feat(ide): add Antigravity CLI adapter and session parser

Add full IDE adapter for Google Antigravity CLI (agy), including:
- CLI adapter: scan home, scan project, MCP discovery, hook detection
- Server adapter: config file generation for agent installs
- Session parser: parse brain/<id>/transcript.jsonl into normalized traces
- Hook spec: hooks.json format with PreInvocation + Stop events
- Session push hook: for when agy hook execution is functional
- Reconcile command: observal reconcile --ide antigravity for manual push
- WSL path resolution: transparent Windows/Linux/macOS support
- IDE registry entries on both server and CLI sides
- 19 tests covering adapter protocol, scanning, and hook detection

SPDX-FileCopyrightText: 2026 Hari Srinivasan <harisrini21@gmail.com>
SPDX-License-Identifier: AGPL-3.0-only

Author: shreem26

observal-server/schemas/ide_registry.py

@@ -316,6 +316,46 @@
         "accepts_model_choice": True,
         "auto_sentinel": {"omit_field": True},
     },
+    "antigravity": {
+        "display_name": "Antigravity",
+        "features": {"hooks", "mcp_servers", "skills"},
+        "session_parser": "antigravity",
+        "scopes": ["project", "user"],
+        "default_scope": "project",
+        "scope_labels": ("project (.agents/)", "user (~/.gemini/antigravity-cli/)"),
+        "rules_file": {
+            "project": "AGENTS.md",
+            "user": "~/.gemini/GEMINI.md",
+        },
+        "rules_format": "markdown",
+        "mcp_config_path": {
+            "project": ".agents/mcp_config.json",
+            "user": "~/.gemini/antigravity-cli/mcp_config.json",
+        },
+        "mcp_servers_key": "mcpServers",
+        "skill_file": {
+            "project": ".agents/skills/{name}/SKILL.md",
+            "user": "~/.gemini/antigravity-cli/skills/{name}/SKILL.md",
+        },
+        "skill_format": "markdown",
+        "home_mcp_config": "~/.gemini/antigravity-cli/mcp_config.json",
+        "hook_type": "command",
+        "hook_config_path": {
+            "project": ".agents/hooks.json",
+            "user": "~/.gemini/config/hooks.json",
+        },
+        "hook_scripts_dir": ".agents/hooks",
+        "hook_events_map": {
+            "PreToolUse": "PreToolUse",
+            "PostToolUse": "PostToolUse",
+            "Stop": "Stop",
+            "SessionStart": "SessionStart",
+            "UserPromptSubmit": "PreInvocation",
+        },
+        "config_dir": ".agents",
+        "accepts_model_choice": True,
+        "auto_sentinel": {"omit_setting": True},
+    },
     "pi": {
         "display_name": "Pi",
         "features": {"skills", "hooks", "mcp_servers"},

observal-server/services/ide/antigravity.py

@@ -0,0 +1,65 @@
+# SPDX-FileCopyrightText: 2026 Hari Srinivasan <harisrini21@gmail.com>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+"""Antigravity CLI server-side config generator."""
+
+from __future__ import annotations
+
+from loguru import logger
+
+from schemas.ide_registry import IDE_REGISTRY
+from services.ide import ConfigContext, register_adapter
+
+
+class AntigravityAdapter:
+    """Antigravity CLI IDE adapter for agent config generation."""
+
+    @property
+    def ide_name(self) -> str:
+        return "antigravity"
+
+    def format_config(self, ctx: ConfigContext) -> dict:
+        logger.debug("format_config: agent={}", ctx.safe_name)
+        spec = IDE_REGISTRY["antigravity"]
+        options = ctx.options
+        scope = options.get("scope", spec["default_scope"])
+
+        # MCP config
+        mcp_path = spec["mcp_config_path"].get(scope, spec["mcp_config_path"]["project"])
+        mcp_content = {spec["mcp_servers_key"]: ctx.mcp_configs}
+
+        # Rules file
+        rules_path = spec["rules_file"].get(scope, spec["rules_file"]["project"])
+        rules_path = rules_path.replace("{name}", ctx.safe_name)
+
+        # Skill files
+        skill_files = []
+        skill_path_template = spec["skill_file"].get(scope, spec["skill_file"]["project"])
+        for skill in ctx.skill_configs:
+            skill_name = skill.get("name", "unnamed")
+            skill_path = skill_path_template.replace("{name}", skill_name)
+            skill_files.append({"path": skill_path, "content": skill.get("content", "")})
+
+        result: dict = {
+            "rules_file": {"path": rules_path, "content": ctx.rules_content},
+            "mcp_config": {"path": mcp_path, "content": mcp_content},
+            "scope": scope,
+        }
+
+        if skill_files:
+            result["skill_files"] = skill_files
+
+        # Model choice
+        model = options.get("_resolved_model")
+        if model:
+            mcp_content["model"] = model
+
+        warnings = list(ctx.compatibility_warnings)
+        warnings.extend(options.get("_model_warnings") or [])
+        if warnings:
+            result["_warnings"] = warnings
+
+        return result
+
+
+register_adapter(AntigravityAdapter())

observal-server/services/ide/load_all.py

@@ -7,6 +7,7 @@
 Each adapter module auto-registers itself on import.
 """
 
+from services.ide import antigravity as _antigravity  # noqa: F401
 from services.ide import claude_code as _claude_code  # noqa: F401
 from services.ide import codex as _codex  # noqa: F401
 from services.ide import copilot as _copilot  # noqa: F401

observal-server/services/session_parsers/__init__.py

@@ -27,6 +27,7 @@
 from .cursor import parse_rows as _parse_cursor
 from .kiro import parse_rows as _parse_kiro
 from .pi import parse_rows as _parse_pi
+from .antigravity import parse_rows as _parse_antigravity
 
 # Maps session_parser ID -> parse_rows callable.
 # Add new entries here when implementing a new JSONL format.
@@ -36,6 +37,7 @@
     "cursor": _parse_cursor,
     "kiro": _parse_kiro,
     "pi": _parse_pi,
+    "antigravity": _parse_antigravity,
 }
 
 

observal-server/services/session_parsers/antigravity.py

@@ -0,0 +1,149 @@
+# SPDX-FileCopyrightText: 2026 Hari Srinivasan <harisrini21@gmail.com>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+"""Antigravity CLI JSONL session parser.
+
+Real transcript format (brain/<id>/.system_generated/logs/transcript.jsonl):
+  {
+    "step_index": 0,
+    "source": "USER_EXPLICIT" | "MODEL" | "SYSTEM",
+    "type": "USER_INPUT" | "PLANNER_RESPONSE" | "LIST_DIRECTORY" | "CONVERSATION_HISTORY" | ...,
+    "status": "DONE" | "IN_PROGRESS" | "ERROR",
+    "created_at": "2026-05-31T17:54:04Z",
+    "content": "...",           # present on USER_INPUT and PLANNER_RESPONSE
+    "tool_calls": [             # present on PLANNER_RESPONSE when tools are invoked
+      {"name": "list_dir", "args": {...}}
+    ]
+  }
+
+Tool results come as separate rows with type matching the tool name
+(e.g. "LIST_DIRECTORY") and content holding the output.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+
+from .base import basic_event, pick_timestamp
+
+# Types that represent user prompts
+_USER_TYPES = {"USER_INPUT"}
+
+# Types that represent model text responses (no tool calls)
+_ASSISTANT_TYPES = {"PLANNER_RESPONSE"}
+
+# Types that are tool results (source=MODEL, type != USER_INPUT/PLANNER_RESPONSE/SYSTEM types)
+_SYSTEM_TYPES = {"CONVERSATION_HISTORY", "SYSTEM_PROMPT"}
+
+_USER_REQUEST_RE = re.compile(r"<USER_REQUEST>\s*(.*?)\s*</USER_REQUEST>", re.DOTALL)
+
+
+def _extract_user_text(content: str) -> str:
+    """Strip XML wrapper tags agy injects around user prompts."""
+    m = _USER_REQUEST_RE.search(content)
+    return m.group(1).strip() if m else content.strip()
+
+
+def parse_rows(rows: list[dict]) -> list[dict]:
+    """Parse raw_line Antigravity transcript rows into normalised frontend events."""
+    events: list[dict] = []
+    # Maps step_index of a PLANNER_RESPONSE with tool_calls -> event index
+    # so we can attach tool results back to the tool call event
+    tool_step_index: dict[int, int] = {}
+
+    for row in rows:
+        raw_line = row.get("raw_line", "")
+        ingested_at = row.get("ingested_at", "")
+        row_ts = row.get("timestamp", "")
+        ide = row.get("ide", "antigravity")
+
+        if not raw_line:
+            events.append(basic_event(row))
+            continue
+
+        try:
+            line = json.loads(raw_line)
+        except (json.JSONDecodeError, ValueError):
+            events.append(basic_event(row))
+            continue
+
+        line_type = line.get("type", "")
+        source = line.get("source", "")
+        status = line.get("status", "")
+        content = line.get("content", "")
+        tool_calls = line.get("tool_calls", [])
+        step_index = line.get("step_index", -1)
+        jsonl_ts = line.get("created_at")
+        ts = pick_timestamp(jsonl_ts, row_ts, ingested_at)
+
+        # Skip system/history lines
+        if line_type in _SYSTEM_TYPES or source == "SYSTEM":
+            continue
+
+        # User prompt
+        if line_type in _USER_TYPES and source in ("USER_EXPLICIT", "USER_IMPLICIT"):
+            text = _extract_user_text(content) if content else ""
+            if text:
+                events.append({
+                    "timestamp": ts,
+                    "event_name": "hook_userpromptsubmit",
+                    "body": text[:120],
+                    "attributes": {"tool_input": text},
+                    "service_name": ide,
+                })
+
+        # Model response with tool calls
+        elif line_type in _ASSISTANT_TYPES and tool_calls:
+            for tc in tool_calls:
+                tool_name = tc.get("name", "")
+                args = tc.get("args", {})
+                idx = len(events)
+                events.append({
+                    "timestamp": ts,
+                    "event_name": "hook_posttooluse",
+                    "body": tool_name,
+                    "attributes": {
+                        "tool_name": tool_name,
+                        "tool_input": json.dumps(args) if isinstance(args, dict) else str(args),
+                    },
+                    "service_name": ide,
+                })
+                tool_step_index[step_index] = idx
+
+        # Model text response (no tool calls)
+        elif line_type in _ASSISTANT_TYPES and content and not tool_calls:
+            events.append({
+                "timestamp": ts,
+                "event_name": "hook_assistant_response",
+                "body": content[:120],
+                "attributes": {"tool_response": content},
+                "service_name": ide,
+            })
+
+        # Tool result — type is the tool name (e.g. LIST_DIRECTORY)
+        # source=MODEL, step_index follows the PLANNER_RESPONSE that called it
+        elif source == "MODEL" and line_type not in _ASSISTANT_TYPES and content:
+            # Find the preceding tool call event (step_index - 1 or step_index - 2)
+            parent_idx = tool_step_index.get(step_index - 1) or tool_step_index.get(step_index - 2)
+            if parent_idx is not None and parent_idx < len(events):
+                events[parent_idx]["attributes"]["tool_response"] = content[:500]
+                if status == "ERROR":
+                    events[parent_idx]["attributes"]["tool_status"] = "error"
+            else:
+                # No matching tool call — emit as standalone result
+                events.append({
+                    "timestamp": ts,
+                    "event_name": "hook_posttooluse",
+                    "body": line_type,
+                    "attributes": {
+                        "tool_name": line_type,
+                        "tool_response": content[:500],
+                    },
+                    "service_name": ide,
+                })
+
+        else:
+            events.append(basic_event(row))
+
+    return events

observal_cli/cmd_doctor.py

@@ -612,6 +612,9 @@ def doctor_patch(
             elif target == "pi":
                 changed = _patch_pi(dry_run)
                 any_changes = any_changes or changed
+            elif target == "antigravity":
+                changed = _patch_antigravity(dry_run)
+                any_changes = any_changes or changed
 
         # ── Shims (all IDEs with home MCP config) ──
         if do_shims:
@@ -788,6 +791,44 @@ def _patch_cursor(dry_run: bool) -> bool:
     return True
 
 
+def _patch_antigravity(dry_run: bool) -> bool:
+    """Install session push hooks into ~/.gemini/config/hooks.json."""
+    from observal_cli.ide_specs.antigravity_hooks_spec import (
+        _OBSERVAL_HOOK_NAME,
+        build_antigravity_hooks,
+    )
+    from observal_cli.shared.utils import resolve_antigravity_config_dir
+
+    rprint("[cyan]Antigravity - session push hooks[/cyan]")
+
+    config_dir = resolve_antigravity_config_dir()
+    if config_dir is None:
+        rprint("  [dim]No ~/.gemini/config/ directory - skipping[/dim]")
+        return False
+
+    hooks_path = config_dir / "hooks.json"
+    desired = build_antigravity_hooks()
+
+    existing: dict = {}
+    if hooks_path.exists():
+        try:
+            existing = json.loads(hooks_path.read_text())
+        except (json.JSONDecodeError, OSError):
+            existing = {}
+
+    if _OBSERVAL_HOOK_NAME in existing:
+        rprint("  [dim]Already up to date[/dim]")
+        return False
+
+    existing.update(desired)
+    if not dry_run:
+        hooks_path.write_text(json.dumps(existing, indent=2) + "\n")
+
+    verb = "Would install" if dry_run else "Installed"
+    rprint(f"  {verb} hooks in {hooks_path}")
+    return True
+
+
 def _patch_pi(dry_run: bool) -> bool:
     """Install observal-pi into ~/.pi/agent/settings.json packages."""
     optic.trace("dry_run={}", dry_run)