# antigravity-cli

Adopt this skill: https://proskills.md/skills/other/scarletkc-agents-skills-antigravity-cli

Page: https://proskills.md/skills/other/scarletkc-agents-skills-antigravity-cli
This document is the public skill bundle for that page. It contains SKILL.md and every helper file the catalog carries. No login is required.

## File: skills/antigravity-cli/SKILL.md
size: 6323
sha: ad2c05d74e824891095d98ad86e1c4d7a49ebaa9
upstream: https://github.com/scarletkc/agents/blob/main/skills/antigravity-cli/SKILL.md

````text
---
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.
````

## File: skills/antigravity-cli/scripts/agent_task_runtime.py
size: 9974
sha: 41242d8fb9744cbabe2e9f15adb014bf0b001452
upstream: https://github.com/scarletkc/agents/blob/main/skills/antigravity-cli/scripts/agent_task_runtime.py

```text
# 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
```

## File: skills/antigravity-cli/scripts/agy.py
size: 27786
sha: 0b982f44362e41183c4b55c51e1edc845edfff88
upstream: https://github.com/scarletkc/agents/blob/main/skills/antigravity-cli/scripts/agy.py

```text
#!/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())
```

## File: skills/antigravity-cli/references/client.md
size: 6033
sha: 87c84810308fbab4fb30933667ef1a16f3170a4b
upstream: https://github.com/scarletkc/agents/blob/main/skills/antigravity-cli/references/client.md

````text
# 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/).
````

## File: README.md
upstream: https://github.com/scarletkc/agents/blob/main/README.md

````text
# Agents

Shared standards and skills for AI coding agents: Claude Code, Codex CLI,
and anything else that reads `AGENTS.md` or the
[SKILL.md](https://agentskills.io) format.

## Featured in

- [Awesome Agent Skills](https://github.com/VoltAgent/awesome-agent-skills/blob/main/README.md#development-and-testing)

## Contents

- [`AGENTS.md`](AGENTS.md): project-agnostic base guidelines. Copy it into a
  repository and prepend the project-specific sections (structure, commands,
  domain notes).
- [`user/AGENTS.md`](user/AGENTS.md): user-level collaboration preferences,
  kept separate from the project-level guidelines.
- [`skills/`](skills/): cross-agent skills; each skill is a directory
  containing a `SKILL.md` plus optional resources.

## Skills

Each summary is generated from the skill's own metadata. Open its `SKILL.md`
for the complete activation description and instructions.

<!-- skills:start -->
| Skill | Summary |
|-------|---------|
| [antigravity-cli](skills/antigravity-cli/SKILL.md) | Delegate work to Antigravity CLI, inspect progress and tool failures, and continue the same conversation across follow-up tasks. |
| [ask-to-plan](skills/ask-to-plan/SKILL.md) | Turn a rough idea into clear goals, requirements, a solution, and an actionable plan through guided questions and native choices. |
| [codex-cli](skills/codex-cli/SKILL.md) | Reach for the Codex CLI when a task is hard enough to earn it: second-model review, bounded hand-offs, sandbox permissions, and a model and effort matched to the difficulty. |
| [devin-cli](skills/devin-cli/SKILL.md) | Hand bounded tasks to Devin CLI over ACP, keep its permissions single-use, watch progress, and resume the same session for follow-ups. |
| [grok-cli](skills/grok-cli/SKILL.md) | Delegate work to Grok Build, keep its session for follow-ups, and inspect results while the supervising agent continues working. |
| [marketing-copy](skills/marketing-copy/SKILL.md) | Write outbound promo copy that stays truthful, discloses only what may be public, and earns attention without hype. |
| [product-writing](skills/product-writing/SKILL.md) | Write accurate product copy and useful technical docs. |
| [scoped-change](skills/scoped-change/SKILL.md) | Hold a change to the size the request defines: no unrequested surfaces, no speculative layers, no half-applied edits. |
| [talk-like-scarletkc](skills/talk-like-scarletkc/SKILL.md) | Write and translate in scarletkc's natural voice without generic AI phrasing. |
| [ux-writing](skills/ux-writing/SKILL.md) | Review user-facing copy and documentation for clarity, consistency, facts that do not go stale, and no leftover intermediate state. |
| [worktree-pr](skills/worktree-pr/SKILL.md) | Optionally run a task in its own worktree: branch from the integration branch, compare against the baseline, and prepare it for a PR. |
| [x-content](skills/x-content/SKILL.md) | Find worthwhile X topics and write posts that earn attention and keep readers engaged. |
<!-- skills:end -->

[`product-writing`](skills/product-writing/SKILL.md) is a compact alternative to
[`ux-writing`](skills/ux-writing/SKILL.md) for product copy and technical docs.
Use one of these guides for a task. `ux-writing` retains the original, more
detailed rules; `product-writing` leaves more structural and editorial choices
to the task.

## Install skills

Send this prompt to your coding agent:

> Install skills from https://github.com/scarletkc/agents. Show me the available
> skills and let me choose which ones to install.

You can also use either CLI directly. Both installers let you choose individual
skills. Replace `codex` with another supported agent, such as `claude-code`,
when needed.

### GitHub CLI

Browse the repository and install a skill for the current user:

```sh
gh skill install scarletkc/agents --agent codex --scope user
```

For a non-interactive installation, provide the skill name:

```sh
gh skill install scarletkc/agents <skill> --agent codex --scope user
```

Update installed skills with `gh skill update --all`.

### skills CLI

Browse the repository and choose skills and agents interactively:

```sh
npx skills add scarletkc/agents -g
```

For a non-interactive installation, provide the skill and agent:

```sh
npx skills add scarletkc/agents --skill <skill> --agent codex -g -y
```

Update global skills with `npx skills update -g -y`.

## Package a skill

Create a ZIP for a web app that accepts skill uploads:

```sh
python scripts/package_skill.py <skill>
```

The archive is written to `dist/<skill>.zip`. Its root contains `SKILL.md`
and the skill's supporting files, ready to upload as one file. Root-level
`evals/` and local build artifacts are excluded.

The CLI skills bundle their shared task runtime so each skill can be installed
independently. After editing `scripts/agent_task_runtime.py`, run
`python scripts/sync_agent_runtime.py` to update the bundled copies. The test
suite checks that those copies match the maintained source.

## Feedback and contributions

Report problems or suggest improvements through
[GitHub Issues](https://github.com/scarletkc/agents/issues). Pull requests are
welcome.

## License

[Apache-2.0](LICENSE). Attribution information is provided in
[`NOTICE`](NOTICE).
````