antigravity-cli

5 files shipped with this skill. Inline text is the catalog copy. The upstream link is the source repository.

• skills/antigravity-cli/SKILL.md6.2 KBad2c05d74e82upstream ↗

  ---
  name: antigravity-cli
  description: "Delegate bounded investigation, review, or implementation tasks to Antigravity CLI from a supervising agent. Includes a background client for task progress, follow-up prompts, conversation recovery, cancellation, and visible failures."
  license: Apache-2.0
  metadata:
    author: scarletkc
    source: https://github.com/scarletkc/agents
    summary: "Delegate work to Antigravity CLI, inspect progress and tool failures, and continue the same conversation across follow-up tasks."
  ---
  
  # Antigravity CLI
  
  Use the installed `agy` CLI as an external coding agent. The bundled client
  returns a task ID immediately and keeps a background worker available for
  follow-ups. It runs each turn with official streaming JSON input/output and
  resumes the exact Antigravity conversation on the next turn. Waiting for the
  process and stderr to finish makes permission denials and partial-output
  timeouts visible; each follow-up incurs CLI startup time.
  
  Delegate when the user requests Antigravity or another agent can add value
  that you can verify. Keep routine work with the supervising agent when the
  extra startup and review time would outweigh that benefit.
  
  Antigravity CLI must already be installed and authenticated. Run
  [scripts/agy.py](scripts/agy.py) with the project's Python 3.12+ interpreter.
  The helper uses only the standard library; when the project has no Python
  environment, `uv run --python 3.12 <script> ...` also works. Resolve the
  script relative to this skill, not the project receiving the task.
  
  ## Assign and follow up
  
  Provide the goal, absolute working directory, relevant files and evidence,
  scope, and completion check. Antigravity does not inherit the supervising
  agent's conversation. Use a UTF-8 prompt file for substantial instructions:
  
  ```sh
  python <script> start --cwd <absolute-project-path> --prompt-file <task-file>
  python <script> wait <task_id> --timeout 30
  python <script> status <task_id>
  python <script> reply <task_id> --prompt-file <follow-up-file>
  python <script> cancel <task_id>
  python <script> close <task_id>
  ```
  
  Keep independent work with the supervising agent while the task runs.
  `wait` timing out only returns a snapshot; it does not cancel the task.
  `reply` is accepted after the current turn finishes. Retain the `task_id`
  and close the worker when no more follow-ups are needed.
  
  The helper supplies both the process working directory and `--add-dir`.
  For concurrent edits to overlapping files, use separate worktrees and pass
  their paths as `--cwd`; the helper does not create worktrees. Check actual
  output paths, since Antigravity can also write to its conversation artifacts.
  
  ## Configuration and permissions
  
  Inherit Antigravity's authentication, model, and permission configuration.
  Subscription and API authentication remain Antigravity's responsibility.
  Do not switch providers, change authentication, or rewrite its global
  settings to conceal a failure.
  
  `start --model`, `--effort`, `--mode`, and `--sandbox` are task-specific
  overrides. Honor the user's model or effort choice. Otherwise choose effort
  according to complexity, error cost, and ease of verification; omit the
  override when the model's support or the appropriate choice is uncertain.
  Discover models with `agy --output-format json models`.
  
  Match permissions to the authorized work. `--mode plan` is a planning
  instruction, not enforced read-only access: headless plan review can proceed
  automatically. The supervising agent's sandbox does not configure AGY's.
  Do not promise isolation without checking the target platform's support.
  
  For authorized workspace edits, `--mode accept-edits` allows file changes
  without interactive diff review; shell commands still follow tool permission
  rules. If a file tool rejects the target as an invalid artifact path, ask
  Antigravity to write a regular workspace file instead of an artifact. Inspect
  that tool error before attempting a shell workaround or expanding permissions.
  
  Headless AGY cannot accept programmatic approval responses. When the helper
  reports `blocked`, inspect `denied_actions` and the partial result. If the
  task requires those tools, use scoped Antigravity permission rules or an
  interactive AGY session to establish the required permissions, following the
  original authorization. Do not repeatedly submit a blocked operation with
  unchanged permissions. The helper exposes `--dangerously-skip-permissions`
  only as an explicit override; do not use it merely to avoid an approval.
  Scoped rules limit which actions match; they are stored in Antigravity's
  global settings, not as temporary grants managed by this helper.
  
  ## Check results and recover
  
  Read `status`, `closed`, `stop_reason`, `error`, `denied_actions`, and
  `tool_errors` together. `completed` means the turn returned a response, not
  that its claims or edits were verified. Inspect the diff, output files, and
  relevant checks. AGY may recover from an individual tool error, so tool
  errors remain visible even when the final turn completes.
  
  The helper does not equate native `SUCCESS` or exit code 0 with completion.
  It checks permission denials and stderr timeout diagnostics, and enforces
  its own startup, turn, and shutdown deadlines.
  
  `cancel` stops that task's AGY process tree and closes its worker. It does
  not undo edits or external actions already performed. Read partial results
  before retrying. To continue a closed conversation, explicitly start a new
  task with its returned `session_id`:
  
  ```sh
  python <script> start --cwd <same-project-path> --conversation <session_id> --prompt-file <follow-up-file>
  ```
  
  An open task can use `reply` after a provider failure. This is an explicit
  new turn in the same conversation; there is no automatic retry. If startup
  failed before a session ID was obtained, resolve the error and start again.
  
  If installation, authentication, network, permissions, or execution blocks
  the task, tell the user what failed and what remains incomplete. Do not
  silently replace the requested Antigravity contribution with your own work
  or another provider. Committing, publishing, and other external actions
  retain the authorization boundaries of the original assignment.
  
  Use `<script> --help` for commands and
  [references/client.md](references/client.md) for state fields, timeouts,
  storage, and diagnostics.
  

• skills/antigravity-cli/scripts/agent_task_runtime.py9.7 KB41242d8fb974upstream ↗

  # Generated by scripts/sync_agent_runtime.py; edit scripts/agent_task_runtime.py.
  """Shared task storage, worker launch, and Windows process ownership for CLI skills.
  
  Run scripts/sync_agent_runtime.py after editing this canonical source.
  """
  
  from __future__ import annotations
  
  import argparse
  import ctypes
  import json
  import math
  import os
  import sqlite3
  import subprocess
  import sys
  import threading
  import time
  import uuid
  from ctypes import wintypes
  from pathlib import Path
  from typing import Any
  
  HEARTBEAT_TIMEOUT = 15
  
  
  class TaskError(RuntimeError):
      pass
  
  
  def encode(value: Any) -> str:
      # ASCII escapes keep redirected JSON usable on non-UTF-8 Windows terminals.
      return json.dumps(value, ensure_ascii=True, allow_nan=False)
  
  
  class Store:
      def __init__(self, root: Path):
          self.root = root.expanduser().resolve()
          self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
          self.db = sqlite3.connect(self.root / "tasks.sqlite3", timeout=10)
          self.db.execute("PRAGMA journal_mode=WAL")
          self.db.executescript("""
              CREATE TABLE IF NOT EXISTS tasks (
                  id TEXT PRIMARY KEY, config TEXT NOT NULL, state TEXT NOT NULL
              );
              CREATE TABLE IF NOT EXISTS commands (
                  id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL,
                  payload TEXT NOT NULL, result TEXT
              );
          """)
  
      def close(self) -> None:
          self.db.close()
  
      def create(self, config: dict[str, Any]) -> dict[str, Any]:
          task_id = uuid.uuid4().hex
          directory = self.root / task_id
          directory.mkdir(mode=0o700)
          state = {
              "task_id": task_id,
              "session_id": None,
              "status": "starting",
              "closed": False,
              "turn": 0,
              "text": "",
              "pending_permissions": [],
              "tools": [],
              "cwd": config["cwd"],
              "updated_at": time.time(),
              "log_dir": str(directory),
          }
          with self.db:
              self.db.execute(
                  "INSERT INTO tasks VALUES (?, ?, ?)",
                  (task_id, encode(config), encode(state)),
              )
          return state
  
      def get(self, task_id: str) -> tuple[dict[str, Any], dict[str, Any]]:
          if len(task_id) != 32 or any(c not in "0123456789abcdef" for c in task_id):
              raise TaskError("Invalid task ID; use the task_id returned by start")
          row = self.db.execute(
              "SELECT config, state FROM tasks WHERE id=?", (task_id,)
          ).fetchone()
          if row is None:
              raise TaskError("Unknown task ID; use the same --state-dir as start")
          return json.loads(row[0]), json.loads(row[1])
  
      def save(self, state: dict[str, Any]) -> None:
          state["updated_at"] = time.time()
          with self.db:
              self.db.execute(
                  "UPDATE tasks SET state=? WHERE id=?", (encode(state), state["task_id"])
              )
  
      def status(self, task_id: str) -> dict[str, Any]:
          _, state = self.get(task_id)
          if (
              not state["closed"]
              and time.time() - state["updated_at"] > HEARTBEAT_TIMEOUT
          ):
              state = {
                  **state,
                  "status": "unresponsive",
                  "error": "Worker heartbeat expired; inspect worker.log before starting another task",
              }
          return state
  
      def submit(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
          state = self.status(task_id)
          if state["closed"] or state["status"] == "unresponsive":
              raise TaskError(
                  "Task is closed or unresponsive; its recorded result remains available via status"
              )
          with self.db:
              cursor = self.db.execute(
                  "INSERT INTO commands(task_id,payload) VALUES (?,?)",
                  (task_id, encode(payload)),
              )
              command_id = cursor.lastrowid
          deadline = time.monotonic() + 5
          while time.monotonic() < deadline:
              row = self.db.execute(
                  "SELECT result FROM commands WHERE id=?", (command_id,)
              ).fetchone()
              if row[0] is not None:
                  result = json.loads(row[0])
                  if "command_error" in result:
                      raise TaskError(result["command_error"])
                  return {"command_id": command_id, **result}
              time.sleep(0.05)
          return {
              "task_id": task_id,
              "command_id": command_id,
              "command_status": "queued",
              "note": "Command not yet acknowledged; check status before retrying",
          }
  
      def commands(self, task_id: str) -> list[tuple[int, str]]:
          return self.db.execute(
              "SELECT id,payload FROM commands WHERE task_id=? AND result IS NULL ORDER BY id",
              (task_id,),
          ).fetchall()
  
      def acknowledge(self, command_id: int, result: dict[str, Any]) -> None:
          with self.db:
              self.db.execute(
                  "UPDATE commands SET result=? WHERE id=?", (encode(result), command_id)
              )
  
  
  def launch_worker(store: Store, config: dict[str, Any], script: Path) -> dict[str, Any]:
      state = store.create(config)
      command = [
          sys.executable,
          str(script.resolve()),
          "--state-dir",
          str(store.root),
          "--worker",
          state["task_id"],
      ]
      options: dict[str, Any] = {}
      if os.name == "nt":
          options["creationflags"] = (
              subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
          )
      else:
          options["start_new_session"] = True
      try:
          with (store.root / state["task_id"] / "worker.log").open("ab") as log:
              process = subprocess.Popen(
                  command,
                  stdin=subprocess.DEVNULL,
                  stdout=log,
                  stderr=log,
                  cwd=store.root,
                  close_fds=True,
                  **options,
              )
          # Retain no inherited pipes: a caller can exit while the worker lives.
          threading.Thread(target=process.wait, daemon=True).start()
          return {**state, "worker_pid": process.pid}
      except OSError as exc:
          state.update(status="failed", closed=True, error=str(exc))
          store.save(state)
          raise
  
  
  def read_prompt(args: argparse.Namespace) -> str:
      if args.prompt_file is not None:
          text = args.prompt_file.read_text(encoding="utf-8-sig")
      elif args.prompt is not None:
          text = args.prompt
      else:
          text = sys.stdin.buffer.read().decode("utf-8-sig")
      if not text.strip():
          raise TaskError(
              "Prompt is empty; provide --prompt, --prompt-file, or UTF-8 stdin"
          )
      return text
  
  
  def seconds(value: str) -> float:
      number = float(value)
      if not math.isfinite(number) or number <= 0:
          raise argparse.ArgumentTypeError(
              "timeout must be a finite number greater than zero"
          )
      return number
  
  
  class BasicLimits(ctypes.Structure):
      _fields_ = [
          ("process_time", ctypes.c_int64),
          ("job_time", ctypes.c_int64),
          ("flags", wintypes.DWORD),
          ("min_working_set", ctypes.c_size_t),
          ("max_working_set", ctypes.c_size_t),
          ("active_processes", wintypes.DWORD),
          ("affinity", ctypes.c_size_t),
          ("priority", wintypes.DWORD),
          ("scheduling", wintypes.DWORD),
      ]
  
  
  class ExtendedLimits(ctypes.Structure):
      _fields_ = [
          ("basic", BasicLimits),
          ("io_counters", ctypes.c_uint64 * 6),
          ("process_memory", ctypes.c_size_t),
          ("job_memory", ctypes.c_size_t),
          ("peak_process_memory", ctypes.c_size_t),
          ("peak_job_memory", ctypes.c_size_t),
      ]
  
  
  class WindowsJob:
      def __init__(self, pid: int):
          self.kernel = ctypes.WinDLL("kernel32", use_last_error=True)
          signatures = {
              "CreateJobObjectW": ([ctypes.c_void_p, wintypes.LPCWSTR], wintypes.HANDLE),
              "SetInformationJobObject": (
                  [wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD],
                  wintypes.BOOL,
              ),
              "OpenProcess": (
                  [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD],
                  wintypes.HANDLE,
              ),
              "AssignProcessToJobObject": (
                  [wintypes.HANDLE, wintypes.HANDLE],
                  wintypes.BOOL,
              ),
              "CloseHandle": ([wintypes.HANDLE], wintypes.BOOL),
          }
          for name, (args, result) in signatures.items():
              function = getattr(self.kernel, name)
              function.argtypes = args
              function.restype = result
          self.handle = self.kernel.CreateJobObjectW(None, None)
          if not self.handle:
              raise ctypes.WinError(ctypes.get_last_error())
          try:
              limits = ExtendedLimits()
              limits.basic.flags = 0x2000  # JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
              if not self.kernel.SetInformationJobObject(
                  self.handle, 9, ctypes.byref(limits), ctypes.sizeof(limits)
              ):
                  raise ctypes.WinError(ctypes.get_last_error())
              process = self.kernel.OpenProcess(
                  0x0101, False, pid
              )  # SET_QUOTA | TERMINATE
              if not process:
                  raise ctypes.WinError(ctypes.get_last_error())
              try:
                  if not self.kernel.AssignProcessToJobObject(self.handle, process):
                      raise ctypes.WinError(ctypes.get_last_error())
              finally:
                  self.kernel.CloseHandle(process)
          except BaseException:
              self.close()
              raise
  
      def close(self):
          if self.handle:
              if not self.kernel.CloseHandle(self.handle):
                  raise ctypes.WinError(ctypes.get_last_error())
              self.handle = None
  

• skills/antigravity-cli/scripts/agy.py27 KB0b982f44362eupstream ↗

  #!/usr/bin/env python3
  """Delegate background tasks to Antigravity CLI. Requires Python 3.12+."""
  
  from __future__ import annotations
  
  import argparse
  import json
  import os
  import queue
  import shutil
  import signal
  import sqlite3
  import subprocess
  import threading
  import time
  import traceback
  import uuid
  from pathlib import Path
  from typing import Any
  
  from agent_task_runtime import (
      Store,
      TaskError,
      encode,
      launch_worker,
      read_prompt,
      seconds,
  )
  
  FINISHED = {"completed", "blocked", "failed", "cancelled", "timed_out"}
  FAILURES = {"blocked", "failed", "timed_out", "unresponsive"}
  
  
  def default_state_dir() -> Path:
      if os.name == "nt":
          return (
              Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData/Local"))
              / "antigravity-tasks"
          )
      return (
          Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local/state"))
          / "antigravity-tasks"
      )
  
  
  def conversation_id(value: str) -> str:
      try:
          return str(uuid.UUID(value))
      except (ValueError, TypeError, AttributeError) as exc:
          raise TaskError(
              "Invalid conversation ID; use the session_id returned by a previous task"
          ) from exc
  
  
  class TurnProcess:
      """One NDJSON turn; collect both channels before reporting completion."""
  
      def __init__(self, command: list[str], cwd: str, log_path: Path):
          self.messages: queue.Queue[tuple[str, Any]] = queue.Queue()
          self.log_path = log_path
          self.log = log_path.open("ab")
          self.log_offset = self.log.tell()
          self.job = None
          self.closed = False
          self.writer: threading.Thread | None = None
          options = (
              {"creationflags": subprocess.CREATE_NO_WINDOW}
              if os.name == "nt"
              else {"start_new_session": True}
          )
          try:
              self.process = subprocess.Popen(
                  command,
                  cwd=cwd,
                  stdin=subprocess.PIPE,
                  stdout=subprocess.PIPE,
                  stderr=self.log,
                  text=True,
                  encoding="utf-8",
                  errors="strict",
                  bufsize=1,
                  **options,
              )
          except BaseException:
              self.log.close()
              raise
          if os.name == "nt":
              from agent_task_runtime import WindowsJob
  
              try:
                  self.job = WindowsJob(self.process.pid)
              except BaseException:
                  self.process.kill()
                  self.process.wait(timeout=10)
                  self.process.stdin.close()
                  self.process.stdout.close()
                  self.log.close()
                  raise
          self.reader = threading.Thread(target=self._read, daemon=True)
          self.reader.start()
  
      def _read(self) -> None:
          try:
              for line in self.process.stdout:
                  event = json.loads(line)
                  if not isinstance(event, dict) or not isinstance(
                      event.get("event"), str
                  ):
                      raise TypeError("expected an event object")
                  self.messages.put(("event", event))
          except (TypeError, ValueError, OSError) as exc:
              self.messages.put(("error", f"Could not read Antigravity stream: {exc}"))
          finally:
              self.messages.put(("eof", None))
  
      def prompt(self, text: str) -> None:
          # A peer that stops reading stdin must not block the worker's watchdog
          # or its command queue. The writer owns stdin until EOF or termination.
          self.writer = threading.Thread(
              target=self._send_prompt, args=(text,), daemon=True
          )
          self.writer.start()
  
      def _send_prompt(self, text: str) -> None:
          error = None
          try:
              self.process.stdin.write(
                  encode({"event": "user", "message": {"content": text}}) + "\n"
              )
              self.process.stdin.flush()
          except (OSError, ValueError) as exc:
              error = str(exc)
          finally:
              try:
                  # EOF ends this process after its turn. Replies resume the ID.
                  self.process.stdin.close()
              except (OSError, ValueError) as exc:
                  error = error or str(exc)
          if error is not None:
              self.messages.put(
                  (
                      "error",
                      f"Could not send the prompt to Antigravity: {error}; inspect agy.log",
                  )
              )
  
      def close(self) -> None:
          if self.closed:
              return
          if self.job is not None:
              self.job.close()
          else:
              try:
                  os.killpg(self.process.pid, signal.SIGKILL)
              except ProcessLookupError:
                  pass
          self.process.wait(timeout=10)
          if self.writer is not None:
              self.writer.join(timeout=2)
              if self.writer.is_alive():
                  raise TaskError(
                      "Antigravity stdin did not close after stopping its process tree"
                  )
          self.reader.join(timeout=2)
          if self.reader.is_alive():
              raise TaskError(
                  "Antigravity stdout did not close after stopping its process tree"
              )
          self.process.stdin.close()
          self.process.stdout.close()
          self.log.close()
          self.closed = True
  
      def diagnostics(self) -> tuple[str, str | None, str | None]:
          timeout = None
          error = None
          tail = ""
          with self.log_path.open("rb") as stream:
              stream.seek(self.log_offset)
              for raw in stream:
                  line = raw.decode("utf-8", errors="replace")
                  tail = (tail + line)[-8192:]
                  if line.startswith("[agy] print timeout after "):
                      timeout = line.strip()
                  elif line.startswith("error:"):
                      error = line.strip()
          return tail, timeout, error
  
  
  class Worker:
      def __init__(self, store: Store, task_id: str):
          self.store = store
          self.config, self.state = store.get(task_id)
          self.state.pop("pending_permissions", None)
          self.state["session_id"] = self.config.get("conversation")
          self.state["denied_actions"] = []
          self.directory = store.root / task_id
          self.process: TurnProcess | None = None
          self.pending_prompt: str | None = self.config["prompt"]
          self.initialized = False
          self.result: dict[str, Any] | None = None
          self.started = 0.0
          self.turn_started: float | None = None
          self.result_at: float | None = None
          self.idle_since = time.monotonic()
          self.last_save = 0.0
          self.stop = False
  
      def save(self) -> None:
          self.store.save(self.state)
          self.last_save = time.monotonic()
  
      def record(self, value: dict[str, Any]) -> None:
          with (self.directory / "events.jsonl").open("a", encoding="utf-8") as log:
              log.write(
                  encode({"time": time.time(), "turn": self.state["turn"], **value})
                  + "\n"
              )
  
      def start_process(self) -> None:
          command = self.config["command"].copy()
          if self.state["session_id"]:
              command.extend(["--conversation", self.state["session_id"]])
          self.initialized = False
          self.result = None
          self.result_at = None
          self.turn_started = None
          self.started = time.monotonic()
          self.state.update(status="starting", text="", tools=[], denied_actions=[])
          for key in (
              "error",
              "stop_reason",
              "raw_status",
              "result",
              "stderr_tail",
              "exit_code",
              "tool_errors",
          ):
              self.state.pop(key, None)
          self.process = TurnProcess(
              command, self.config["cwd"], self.directory / "agy.log"
          )
          self.state["agy_pid"] = self.process.process.pid
          self.record({"type": "process_start", "pid": self.state["agy_pid"]})
          self.save()
  
      def handle(self, event: dict[str, Any]) -> None:
          self.record({"type": "event", "message": event})
          kind = event["event"]
          if kind == "init":
              if self.initialized or self.result is not None:
                  raise TaskError("Antigravity sent an unexpected init event")
              session = conversation_id(event.get("conversation_id"))
              expected = self.state["session_id"]
              if expected and expected != session:
                  raise TaskError(
                      "Antigravity resumed a different conversation; no prompt was sent"
                  )
              init = event.get("init")
              if not isinstance(init, dict) or not isinstance(init.get("cwd"), str):
                  raise TaskError(
                      "Antigravity init did not identify its working directory"
                  )
              if os.path.normcase(os.path.realpath(init["cwd"])) != os.path.normcase(
                  self.config["cwd"]
              ):
                  raise TaskError(
                      "Antigravity reported a different working directory; no prompt was sent"
                  )
              self.initialized = True
              self.state.update(session_id=session, init=init)
              self.state.update(status="running", turn=self.state["turn"] + 1)
              prompt = self.pending_prompt
              self.pending_prompt = None
              self.record({"type": "prompt", "text": prompt})
              self.turn_started = time.monotonic()
              self.process.prompt(prompt)
              self.save()
          elif kind == "step_update":
              step = event.get("step_update")
              if (
                  not self.initialized
                  or self.result is not None
                  or not isinstance(step, dict)
              ):
                  raise TaskError("Antigravity sent an unexpected step update")
              if step.get("conversation_id") != self.state["session_id"]:
                  raise TaskError("Antigravity step belongs to a different conversation")
              text = step.get("text_delta", "")
              if not isinstance(text, str):
                  raise TaskError("Antigravity sent an invalid text delta")
              if step.get("step_type") == "agent_response":
                  self.state["text"] += text
              if step.get("step_type") == "tool":
                  summary = {
                      key: step[key]
                      for key in ("step_index", "tool_name", "state")
                      if key in step
                  }
                  info = step.get("tool_info", {})
                  if not isinstance(info, dict):
                      raise TaskError("Antigravity sent invalid tool information")
                  if info.get("error"):
                      summary["error"] = info["error"]
                      self.state.setdefault("tool_errors", []).append(summary.copy())
                      del self.state["tool_errors"][:-10]
                  tools = self.state["tools"]
                  old = next(
                      (
                          t
                          for t in tools
                          if t.get("step_index") == summary.get("step_index")
                      ),
                      None,
                  )
                  if old is None:
                      tools.append(summary)
                      del tools[:-10]
                  else:
                      old.update(summary)
          elif kind == "result":
              result = event.get("result")
              if (
                  self.result is not None
                  or not isinstance(result, dict)
                  or not isinstance(result.get("status"), str)
              ):
                  raise TaskError("Antigravity sent an invalid or duplicate result")
              session = result.get("conversation_id")
              if session and session != self.state["session_id"]:
                  raise TaskError(
                      "Antigravity result belongs to a different conversation"
                  )
              if result["status"] == "SUCCESS" and not self.initialized:
                  raise TaskError(
                      "Antigravity reported success before initializing the conversation"
                  )
              if not isinstance(result.get("response", ""), str):
                  raise TaskError("Antigravity returned an invalid response")
              denied = result.get("denied_actions", [])
              if not isinstance(denied, list) or any(
                  not isinstance(item, dict) for item in denied
              ):
                  raise TaskError("Antigravity returned invalid denied_actions")
              self.result = result
              self.result_at = time.monotonic()
              self.state.update(
                  result=result,
                  raw_status=result["status"],
                  text=result.get("response") or self.state["text"],
                  denied_actions=denied,
              )
              self.pending_prompt = None
              if self.process.writer is None and not self.process.process.stdin.closed:
                  self.process.process.stdin.close()
          else:
              # Keep unknown observational events in the log without assigning
              # them completion, permission, or execution semantics.
              self.state["last_unknown_event"] = kind
  
      def drain(self, limit: int | None = 64) -> None:
          count = 0
          while limit is None or count < limit:
              try:
                  kind, value = self.process.messages.get_nowait()
              except queue.Empty:
                  return
              if kind == "error":
                  raise TaskError(value)
              if kind == "event":
                  self.handle(value)
              count += 1
  
      def finish(self) -> None:
          code = self.process.process.returncode
          self.process.close()
          self.drain(limit=None)
          tail, timeout, error = self.process.diagnostics()
          self.state.update(exit_code=code, stderr_tail=tail)
          self.process = None
          self.idle_since = time.monotonic()
          self.pending_prompt = None
          result = self.result or {}
          raw = result.get("status")
          if timeout:
              self.state.update(
                  status="timed_out", stop_reason="print_timeout", error=timeout
              )
          elif code != 0 or raw != "SUCCESS" or error:
              status = "cancelled" if raw in {"CANCELED", "INTERRUPTED"} else "failed"
              self.state.update(
                  status=status,
                  stop_reason=raw or "missing_result",
                  error=result.get("error")
                  or error
                  or f"Antigravity exited {code} with status {raw!r}; inspect agy.log",
              )
          elif self.state["denied_actions"]:
              self.state.update(
                  status="blocked",
                  stop_reason="permission_denied",
                  error="Antigravity denied tool actions; inspect denied_actions and configure scoped permissions in Antigravity before continuing",
              )
          elif (
              not result.get("response", "").strip() and "structured_output" not in result
          ):
              self.state.update(
                  status="failed",
                  stop_reason="empty_response",
                  error="Antigravity returned an empty response; inspect tools and agy.log before retrying",
              )
          else:
              self.state.update(status="completed", stop_reason="SUCCESS")
          if not self.initialized:
              self.stop = True
          self.save()
  
      def terminate(self, status: str, reason: str, error: str | None = None) -> None:
          self.state.update(status=status, stop_reason=reason, closing=True)
          if error:
              self.state["error"] = error
          self.stop = True
          self.save()
  
      def control(self, payload: dict[str, Any]) -> dict[str, Any]:
          if self.stop:
              raise TaskError("Task is closing")
          action = payload["action"]
          if action == "reply":
              if (
                  self.process is not None
                  or self.pending_prompt is not None
                  or self.state["status"] not in FINISHED
              ):
                  raise TaskError("Wait for the current turn to finish before replying")
              self.pending_prompt = payload["prompt"]
              self.state["status"] = "starting"
          elif action == "cancel":
              if self.process is None and self.pending_prompt is None:
                  raise TaskError(
                      "No active turn to cancel; use close to release an idle task"
                  )
              self.terminate("cancelled", "forced_stop")
          elif action == "close":
              if self.process is not None or self.pending_prompt is not None:
                  self.terminate("cancelled", "forced_stop")
              else:
                  self.state.update(closing=True, close_reason="requested")
                  self.stop = True
          else:
              raise TaskError(f"Unsupported worker command: {action}")
          self.save()
          return self.state.copy()
  
      def run(self) -> None:
          self.state["worker_pid"] = os.getpid()
          self.save()
          try:
              while not self.stop:
                  for command_id, raw in self.store.commands(self.state["task_id"]):
                      try:
                          response = self.control(json.loads(raw))
                      except TaskError as exc:
                          response = {"command_error": str(exc)}
                      self.store.acknowledge(command_id, response)
                  if self.stop:
                      break
                  if self.process is None and self.pending_prompt is not None:
                      self.start_process()
                  if self.process is not None:
                      self.drain()
                      now = time.monotonic()
                      if (
                          self.result is None
                          and self.turn_started is None
                          and now - self.started >= self.config["startup_timeout"]
                      ):
                          self.terminate(
                              "timed_out",
                              "startup_timeout",
                              "Antigravity did not initialize before --startup-timeout; no prompt was sent",
                          )
                      elif (
                          self.result is None
                          and self.turn_started is not None
                          and now - self.turn_started >= self.config["turn_timeout"]
                      ):
                          self.terminate(
                              "timed_out",
                              "turn_timeout",
                              "Turn exceeded --turn-timeout; its process tree was stopped",
                          )
                      elif self.process.process.poll() is not None:
                          self.finish()
                      elif (
                          self.result_at is not None
                          and now - self.result_at >= self.config["exit_timeout"]
                      ):
                          self.terminate(
                              "timed_out",
                              "exit_timeout",
                              "Antigravity returned a result but did not exit before --exit-timeout",
                          )
                  elif time.monotonic() - self.idle_since >= self.config["idle_timeout"]:
                      self.state["close_reason"] = "idle_timeout"
                      self.stop = True
                  if time.monotonic() - self.last_save >= 1:
                      self.save()
                  time.sleep(0.05)
          except Exception as exc:
              traceback.print_exc()
              self.state.update(
                  status="failed", error=str(exc), stop_reason="client_error"
              )
              raise
          finally:
              try:
                  if self.process is not None:
                      self.process.close()
                      tail, _, _ = self.process.diagnostics()
                      self.state.update(
                          stderr_tail=tail, exit_code=self.process.process.returncode
                      )
              except Exception as exc:
                  traceback.print_exc()
                  self.state.update(
                      status="failed", error=f"Could not clean up Antigravity: {exc}"
                  )
                  raise
              finally:
                  self.state.update(closed=True, closing=False)
                  self.save()
  
  
  def launch(store: Store, config: dict[str, Any]) -> dict[str, Any]:
      return launch_worker(store, config, Path(__file__))
  
  
  def executable(value: str | None) -> str:
      found = shutil.which(value or "agy")
      if found:
          return str(Path(found).resolve())
      if value is None:
          candidate = (
              Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData/Local"))
              / "agy/bin/agy.exe"
              if os.name == "nt"
              else Path.home() / ".local/bin/agy"
          )
          if candidate.is_file() and os.access(candidate, os.X_OK):
              return str(candidate.resolve())
      raise TaskError(
          "Antigravity CLI executable not found; install agy or pass --agy with its path"
      )
  
  
  def parser() -> argparse.ArgumentParser:
      root = argparse.ArgumentParser(
          description=__doc__, formatter_class=argparse.ArgumentDefaultsHelpFormatter
      )
      root.add_argument(
          "--state-dir",
          type=Path,
          default=default_state_dir(),
          help="task storage directory (reuse it across commands)",
      )
      root.add_argument("--worker", help=argparse.SUPPRESS)
      sub = root.add_subparsers(dest="action")
      start = sub.add_parser(
          "start",
          help="start a background task",
          formatter_class=argparse.ArgumentDefaultsHelpFormatter,
      )
      start.add_argument(
          "--cwd",
          type=Path,
          required=True,
          help="task working directory and Antigravity workspace",
      )
      start.add_argument("--agy", help="Antigravity executable name or path")
      start.add_argument(
          "--conversation", help="explicitly resume a previous session_id in a new task"
      )
      start.add_argument(
          "--model",
          help="per-task model override; otherwise inherit Antigravity configuration",
      )
      start.add_argument(
          "--effort",
          choices=["low", "medium", "high"],
          help="per-task reasoning effort override",
      )
      start.add_argument(
          "--mode",
          choices=["accept-edits", "plan"],
          help="per-task execution mode; plan is not an enforced read-only mode",
      )
      start.add_argument(
          "--sandbox",
          action="store_true",
          help="enable Antigravity terminal sandbox for this task",
      )
      start.add_argument(
          "--dangerously-skip-permissions",
          action="store_true",
          help="explicitly auto-approve all tools for this task",
      )
      for name, default, help_text in [
          ("startup", 60, "seconds for each AGY process to initialize"),
          ("turn", 900, "seconds per turn before stopping its process tree"),
          ("exit", 10, "seconds for AGY to exit after its final result"),
          ("idle", 1800, "seconds to retain a finished task for follow-ups"),
      ]:
          start.add_argument(
              f"--{name}-timeout", type=seconds, default=default, help=help_text
          )
      reply = sub.add_parser("reply", help="continue a finished, open task")
      reply.add_argument("task_id")
      for item in (start, reply):
          prompts = item.add_mutually_exclusive_group()
          prompts.add_argument("--prompt", help="prompt text; defaults to UTF-8 stdin")
          prompts.add_argument(
              "--prompt-file", type=Path, help="UTF-8 prompt file (BOM accepted)"
          )
      for name, help_text in [
          ("status", "read task state and the latest result"),
          ("wait", "wait for a finished turn without cancelling it"),
          ("cancel", "stop active work and close the task"),
          ("close", "release the task; stop active work if necessary"),
      ]:
          item = sub.add_parser(name, help=help_text)
          item.add_argument("task_id")
          if name == "wait":
              item.add_argument(
                  "--timeout",
                  type=seconds,
                  default=30,
                  help="seconds to wait (default: 30); does not stop AGY",
              )
      return root
  
  
  def main(argv: list[str] | None = None) -> int:
      cli = parser()
      args = cli.parse_args(argv)
      if not args.action and not args.worker:
          cli.error("a command is required")
      store = None
      try:
          store = Store(args.state_dir)
          if args.worker:
              Worker(store, args.worker).run()
              return 0
          if args.action == "start":
              cwd = args.cwd.expanduser().resolve(strict=True)
              if not cwd.is_dir():
                  raise TaskError("--cwd must be a directory")
              command = [
                  executable(args.agy),
                  "--add-dir",
                  str(cwd),
                  "--input-format",
                  "stream-json",
                  "--output-format",
                  "stream-json",
              ]
              # Our watchdog fires first. Native timeouts can report SUCCESS with
              # partial output, so their stderr is still checked after process exit.
              command.extend(
                  [
                      "--print-timeout",
                      f"{args.startup_timeout + args.turn_timeout + 60:g}s",
                  ]
              )
              for key in ("model", "effort", "mode"):
                  if getattr(args, key):
                      command.extend(["--" + key, getattr(args, key)])
              for key in ("sandbox", "dangerously_skip_permissions"):
                  if getattr(args, key):
                      command.append("--" + key.replace("_", "-"))
              result = launch(
                  store,
                  {
                      "cwd": str(cwd),
                      "command": command,
                      "prompt": read_prompt(args),
                      "conversation": conversation_id(args.conversation)
                      if args.conversation
                      else None,
                      **{
                          f"{key}_timeout": getattr(args, f"{key}_timeout")
                          for key in ("startup", "turn", "idle", "exit")
                      },
                  },
              )
          elif args.action in {"status", "wait"}:
              deadline = time.monotonic() + (args.timeout if args.action == "wait" else 0)
              while True:
                  result = store.status(args.task_id)
                  if (
                      result["closed"]
                      or (
                          not result.get("closing")
                          and result["status"] in FINISHED | {"unresponsive"}
                      )
                      or time.monotonic() >= deadline
                  ):
                      break
                  time.sleep(0.1)
          else:
              payload = {"action": args.action}
              if args.action == "reply":
                  payload["prompt"] = read_prompt(args)
              result = store.submit(args.task_id, payload)
          print(encode(result))
          return int(result.get("status") in FAILURES)
      except (TaskError, OSError, ValueError, sqlite3.Error) as exc:
          print(encode({"error": str(exc)}))
          return 1
      finally:
          if store is not None:
              store.close()
  
  
  if __name__ == "__main__":
      raise SystemExit(main())
  

• skills/antigravity-cli/references/client.md5.9 KB87c84810308fupstream ↗

  # Client operations
  
  The helper maintains a background worker for each task. Each turn starts a
  private AGY process with `--input-format stream-json --output-format
  stream-json`, supplies `--add-dir`, waits for `init`, sends one prompt, and
  closes stdin. Subsequent turns use the returned conversation ID. Results
  are finalized only after the process exits and both output channels have
  been collected. No MCP server or unofficial ACP adapter is required.
  
  ## Input and output
  
  `start` and `reply` accept `--prompt`, `--prompt-file`, or UTF-8 stdin.
  Prompt files may contain a UTF-8 BOM. Prompts travel as JSON through stdin,
  not shell-interpreted command text. AGY itself determines prompt and skill
  expansion; stream input does not support CLI-handled slash commands.
  
  The helper emits one JSON object on stdout with Unicode escaped for Windows
  pipe compatibility. Usage errors go to stderr. Runtime errors and states
  `blocked`, `failed`, `timed_out`, and `unresponsive` produce a nonzero helper
  exit code. A successful observation of `starting` or `running` does not mean
  the task has finished.
  
  | Field | Meaning |
  | --- | --- |
  | `task_id` | Helper ID for status, wait, reply, cancel, and close |
  | `session_id` | Antigravity conversation ID; absent until initialization succeeds |
  | `status` | Starting, running, completed, blocked, failed, cancelled, timed out, or unresponsive |
  | `closed` | The background worker has ended; no more replies can be accepted. Check `error` for any cleanup failure |
  | `closing` | Cleanup is in progress; wait for `closed` before starting replacement work |
  | `turn` | Prompts sent by this helper task; a recovered conversation can have additional earlier turns |
  | `text` | Latest response or partial streamed text |
  | `tools` | Up to ten recent tool names, indices, states, and errors |
  | `tool_errors` | Up to ten reported tool errors, including errors AGY later recovered from |
  | `denied_actions` | Actions AGY refused under its headless permission policy |
  | `result` | Native result, including its cumulative conversation usage and turn count |
  | `raw_status` | Native AGY status, which can be SUCCESS for blocked or timed-out work |
  | `stop_reason`, `error` | Helper completion reason and failure details |
  | `init` | Configuration AGY advertised, including its cwd, tools, and any explicit model override |
  | `exit_code` | Native exit code; forced termination can still produce 0 on Windows |
  | `stderr_tail` | Recent diagnostics from this turn; the full log is retained |
  | `log_dir` | Absolute path to logs and events |
  
  Native counters in `result.usage`, `result.num_turns`, and
  `result.duration_seconds` describe the whole conversation. Do not add these
  cumulative counters across replies as though they described separate turns.
  
  Control commands return a `command_id` after queuing the request.
  `command_status: queued` means the worker has not acknowledged it yet.
  Inspect state before retrying, particularly for `reply`, to avoid duplicate
  prompts. A stale worker heartbeat produces `unresponsive`; inspect its
  process and logs instead of assuming the saved running state is current.
  
  ## Deadlines and process ownership
  
  `start --help` lists the defaults. `--startup-timeout` applies to each new
  AGY process, `--turn-timeout` to prompt execution, `--exit-timeout` to a
  process that has returned a result but has not exited, and `--idle-timeout`
  to a finished worker waiting for another prompt. The helper gives AGY a
  longer native print timeout so the helper's watchdog normally fires first.
  It also checks AGY's timeout diagnostics after exit, because native print
  timeouts can return SUCCESS and partial output.
  
  An idle timeout releases the worker and preserves its last result. A helper
  startup, turn, or exit timeout stops active work and closes the task. A
  native timeout result can leave the idle worker available for an explicit
  follow-up. No timeout automatically resubmits a prompt.
  
  Windows processes belong to a Job Object; other platforms use a dedicated
  process group. Cleanup stops that task's process tree, including tool
  children that remain after AGY exits. This also stops task-owned development
  servers; use another process for services intended to outlive the task.
  Cancellation cannot undo work already done. No in-process cooperative
  cancel or approval protocol is advertised.
  
  ## Storage and troubleshooting
  
  State defaults to `%LOCALAPPDATA%/antigravity-tasks` on Windows, or
  `$XDG_STATE_HOME/antigravity-tasks` (`~/.local/state/antigravity-tasks` by
  default) elsewhere. To override it, place `--state-dir <absolute-path>`
  before the command and reuse that path for all commands for the task.
  Keep state outside shared or tracked repositories: logs can contain prompts,
  file contents, tool output, and other project data.
  
  Each task retains `events.jsonl` for prompts, process starts, and native
  events; `agy.log` for stderr across all turns; and `worker.log` for helper
  diagnostics. Logs are not automatically deleted. Close tasks before removing
  their state directory.
  
  The helper discovers `agy` in PATH or its standard user install directory.
  An explicit `--agy` path never falls back to another binary. Authentication
  is handled by the installed CLI; finish any required interactive sign-in
  outside the helper. A model list or valid login does not prove model requests
  can complete. Preserve network and provider errors without attributing them
  to credentials or changing providers automatically.
  
  On Windows the standard binary is `%LOCALAPPDATA%/agy/bin/agy.exe`. If a
  new installation is not yet on the current shell's PATH, open a new terminal
  or use the absolute binary path for direct CLI commands, for example:
  
  ```powershell
  & "$env:LOCALAPPDATA\agy\bin\agy.exe" --output-format json models
  ```
  
  Protocol references: [Headless mode](https://www.antigravity.google/docs/cli/headless/),
  [permissions](https://www.antigravity.google/docs/cli/permissions/), and
  [installation and authentication](https://antigravity.google/docs/cli/install/).
  

• README.mdupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

Plain-text copy, no login: /skill-bundles/scarletkc-agents-skills-antigravity-cli.md