antigravity-bridge
One-directional knowledge bridge from Google Antigravity IDE to OpenClaw. Syncs only .md documentation files from Antigravity projects into OpenClaw workspace for native vector indexing. No secrets, credentials, or binary state are synced — rsync filters enforce .md-only transfer. Supports multiple projects.
Packaged view
This page reorganizes the original catalog entry around fit, installability, and workflow context first. The original raw source lives below.
Install command
npx @skill-hub/cli install openclaw-skills-antigravity-bridge
Repository
Skill path: skills/heintonny/antigravity-bridge
One-directional knowledge bridge from Google Antigravity IDE to OpenClaw. Syncs only .md documentation files from Antigravity projects into OpenClaw workspace for native vector indexing. No secrets, credentials, or binary state are synced — rsync filters enforce .md-only transfer. Supports multiple projects.
Open repositoryBest for
Primary workflow: Ship Full Stack.
Technical facets: Full Stack.
Target audience: everyone.
License: Unknown.
Original source
Catalog source: SkillHub Club.
Repository owner: openclaw.
This is still a mirrored public skill entry. Review the repository before installing into production workflows.
What it helps with
- Install antigravity-bridge into Claude Code, Codex CLI, Gemini CLI, or OpenCode workflows
- Review https://github.com/openclaw/skills before adding antigravity-bridge to shared team environments
- Use antigravity-bridge for development workflows
Works across
Favorites: 0.
Sub-skills: 0.
Aggregator: No.
Original source / Raw SKILL.md
---
name: antigravity-bridge
version: 2.0.1
description: "One-directional knowledge bridge from Google Antigravity IDE to OpenClaw. Syncs only .md documentation files from Antigravity projects into OpenClaw workspace for native vector indexing. No secrets, credentials, or binary state are synced — rsync filters enforce .md-only transfer. Supports multiple projects."
author: heintonny
metadata: {"openclaw":{"emoji":"🌉","tags":["antigravity","gemini","knowledge-sync","multi-agent","bridge","ide","sync"],"requires":{"bins":["rsync","yq"]},"install":[{"id":"yq","kind":"brew","package":"yq","bins":["yq"],"label":"Install yq YAML parser"},{"id":"rsync","kind":"system","package":"rsync","bins":["rsync"],"label":"rsync (usually pre-installed)"}]}}
---
# Antigravity Bridge
One-directional knowledge bridge from Google Antigravity IDE to OpenClaw.
Syncs `.md` files from your Antigravity/Gemini projects into the OpenClaw workspace, where they are natively embedded and indexed for `memory_search`. No MEMORY.md dumping, no custom state tracking — just files on disk, indexed automatically.
## When to Use
- User says "sync antigravity", "bridge sync", "pull antigravity docs"
- You need cross-project awareness of Antigravity-managed context
- After an Antigravity session, to surface new decisions/rules/tasks to OpenClaw
- Scheduled (cron) to keep the knowledge fresh automatically
## When NOT to Use
- Primary coding work (use Antigravity for that — it has IDE, LSP, deep code awareness)
- Writing back to Antigravity projects (this is **one-way only**: Antigravity → OpenClaw)
- Querying the synced knowledge (just use `memory_search` — the files are already indexed)
---
## Architecture
```
Antigravity IDE OpenClaw Workspace
───────────────── ─────────────────────────────
~/repo/acme-corp/acme-platform/
.agent/memory/ ──rsync──► antigravity/acme-platform/
.agent/rules/ ──rsync──► .agent/memory/
.agent/skills/ ──rsync──► .agent/rules/
.agent/sessions/ ──rsync──► .agent/skills/
.agent/tasks.md ──rsync──► .agent/sessions/
.gemini/GEMINI.md ──rsync──► .agent/tasks.md
docs/ ──rsync──► .gemini/GEMINI.md
*.md (root) ──rsync──► docs/
*.md (root)
~/.gemini/antigravity/
knowledge/ ──rsync──► antigravity/gemini-knowledge/
───────────────── ─────────────────────────────
│
OpenClaw native embedder
(memorySearch.extraPaths)
│
memory_search queries ✓
```
**Key design decisions:**
- Files land in `antigravity/<project-name>/` under the OpenClaw workspace destination
- OpenClaw's native embedder indexes them automatically via `memorySearch.extraPaths`
- `sync.sh` is idempotent — safe to run repeatedly or on cron
- Source paths that don't exist are skipped gracefully (no failure)
---
## Setup Guide
### Step 1: Prerequisites
```bash
# rsync (usually pre-installed on macOS/Linux)
rsync --version
# yq — YAML parser (required)
brew install yq # macOS
# or: sudo apt install yq # Ubuntu/Debian
# or: snap install yq # Ubuntu snap
yq --version
```
### Step 2: Copy the config template
```bash
cp ~/.openclaw/workspace/skills/antigravity-bridge/config-template.yaml \
~/.openclaw/workspace/antigravity-bridge.yaml
```
### Step 3: Edit the config
Open `~/.openclaw/workspace/antigravity-bridge.yaml` and configure your projects:
```yaml
projects:
- name: acme-platform
repo: ~/repo/acme-corp/acme-platform
paths:
- .agent/memory
- .agent/rules
- .agent/skills
- .agent/tasks.md
- .gemini/GEMINI.md
- docs
include_root_md: true
knowledge:
- name: gemini-knowledge
path: ~/.gemini/antigravity/knowledge
destination: antigravity
```
- **`projects`** — list of Antigravity-managed repos
- **`knowledge`** — standalone knowledge directories (e.g. Gemini's global knowledge store)
- **`destination`** — subfolder within the OpenClaw workspace (default: `antigravity`)
### Step 4: Configure OpenClaw extraPaths
Tell OpenClaw to index the synced directory. In your OpenClaw config (`~/.openclaw/config.yaml` or equivalent), add:
```yaml
memorySearch:
extraPaths:
- ~/path/to/openclaw/workspace/antigravity
```
Replace with the actual workspace path. After saving, restart OpenClaw or reload memory indexing.
### Step 5: Test with --dry-run
```bash
~/.openclaw/workspace/skills/antigravity-bridge/sync.sh --dry-run --verbose
```
You'll see what *would* be synced without touching anything.
### Step 6: Run for real
```bash
~/.openclaw/workspace/skills/antigravity-bridge/sync.sh --verbose
```
### Step 7: Verify with memory_search
After syncing, query OpenClaw memory to confirm indexing:
```
memory_search: "acme-platform agent rules"
memory_search: "GEMINI.md"
```
If results come back from the synced files, the bridge is working.
---
## Config Reference
```yaml
# ~/.openclaw/workspace/antigravity-bridge.yaml
projects:
- name: <string> # Identifier — used as subfolder name
repo: <path> # Root of the Antigravity project (~ expanded)
paths: # List of paths relative to repo root
- .agent/memory # Directory → recursively sync *.md
- .agent/tasks.md # Single file → synced directly
- docs # Directory → recursively sync *.md
include_root_md: true # Also sync *.md files at repo root (optional, default: false)
knowledge:
- name: <string> # Identifier — used as subfolder name
path: <path> # Source path to rsync *.md from (~ expanded)
destination: antigravity # Target subfolder in OpenClaw workspace
# Full path: <workspace>/<destination>/
```
**Path types:**
- **Directory** — rsync runs with `--include='*.md' --exclude='*'` recursively
- **Single file** — rsync copies the file directly (must end in `.md`)
**Missing sources:** If a configured path doesn't exist, sync.sh logs a warning and skips it. Other paths continue normally. Exit code remains 0.
---
## CLI Reference
```
sync.sh [options]
Options:
--config <path> Config file (default: ~/.openclaw/workspace/antigravity-bridge.yaml)
--project <name> Sync only this project (by name)
--dry-run Show what would be synced, without making changes
--verbose Show rsync output and detailed progress
--help Show this help
```
Examples:
```bash
# Sync everything
sync.sh
# Sync one project only
sync.sh --project acme-platform
# Preview without touching files
sync.sh --dry-run --verbose
# Use a custom config
sync.sh --config ~/my-bridge.yaml
```
---
## Cron Integration
Add to crontab (`crontab -e`) for automatic syncing:
```cron
# Antigravity Bridge — hourly during business hours (Mon-Fri, 08:00-18:00)
0 8-18 * * 1-5 ~/.openclaw/workspace/skills/antigravity-bridge/sync.sh >> ~/.openclaw/logs/antigravity-bridge.log 2>&1
# Nightly full sync (all days, 02:00)
0 2 * * * ~/.openclaw/workspace/skills/antigravity-bridge/sync.sh --verbose >> ~/.openclaw/logs/antigravity-bridge.log 2>&1
```
Create the log directory first:
```bash
mkdir -p ~/.openclaw/logs
```
---
## Troubleshooting
**`yq: command not found`**
Install yq: `brew install yq` (macOS) or see https://github.com/mikefarah/yq
**`Config file not found`**
Copy the template: `cp config-template.yaml ~/.openclaw/workspace/antigravity-bridge.yaml`
**`rsync: command not found`**
Install rsync: `brew install rsync` (macOS) or `sudo apt install rsync`
**No results from memory_search**
- Check that `memorySearch.extraPaths` includes the destination folder
- Restart OpenClaw after changing extraPaths
- Verify files landed in the right place: `ls ~/.openclaw/workspace/antigravity/`
**Files not updating**
- Run with `--verbose` to see rsync output
- Check source paths exist: `ls ~/repo/acme-corp/acme-platform/.agent/memory/`
**Wrong files synced**
- Only `.md` files are synced (rsync filter: `--include='*.md' --exclude='*'`)
- To sync other file types, edit sync.sh patterns
---
## Security & Privacy
- **All data stays local.** No external API calls, no cloud sync, no network access.
- **Only `.md` files are synced.** rsync filters (`--filter='+ *.md' --filter='- *'`) enforce markdown-only transfer. No secrets, credentials, API keys, binary state, session tokens, or config files are ever copied.
- `.agent/` and `.gemini/` directories are Antigravity's documentation folders containing markdown notes about rules, tasks, memory, and project context. They do **not** contain credentials or sensitive runtime state — those are stored elsewhere by the IDE.
- sync.sh only reads from user-configured source paths and writes to a designated OpenClaw workspace subfolder.
- No credentials or tokens required to run.
- Safe to run with `--dry-run` to inspect behavior before committing.
- **Dependencies:** `rsync` (system), `yq` (YAML parser) — both declared in manifest metadata.
---
## Skill Companion Files
> Additional files collected from the skill directory layout.
### README.md
```markdown
# 🌉 Antigravity Bridge
**Bidirectional knowledge bridge between [OpenClaw](https://github.com/openclaw/openclaw) and [Google Antigravity IDE](https://antigravityide.org/).**
Two AI agents. One shared brain. Zero manual sync.
## What This Does
If you use **Antigravity** (Gemini) for coding and **OpenClaw** (any LLM) as your life/DevOps agent, this skill bridges their knowledge systems so both agents share context:
```
┌─────────────────────┐ ┌─────────────────────┐
│ Antigravity │ │ OpenClaw │
│ (Gemini) │ │ (Any LLM) │
│ │ │ │
│ knowledge/ ◄──┼───────┼──► MEMORY.md │
│ .agent/tasks.md ◄──┼───────┼──► coding sub-agents │
│ .agent/memory/ ◄──┼───────┼──► memory/*.md │
│ .agent/sessions/ ◄──┼───────┼──► session handoffs │
└─────────────────────┘ └─────────────────────┘
```
## Features
- **📥 Sync Knowledge** — Pull Antigravity's Knowledge Items, tasks, and lessons into OpenClaw
- **📋 Pick Tasks** — Let OpenClaw coding sub-agents pick up tasks from `.agent/tasks.md`
- **🔄 Cross-Agent Self-Improve** — Update both knowledge systems simultaneously
- **📝 Write Knowledge Items** — Write to Antigravity's native KI format programmatically
- **🌉 Reverse Bridge** — Auto-generate an Antigravity skill for delegating tasks to OpenClaw
## Install
### Via ClawHub (recommended)
```bash
clawhub install antigravity-bridge
```
### Manual
```bash
git clone https://github.com/heintonny/antigravity-bridge.git
cp -r antigravity-bridge ~/.openclaw/skills/
```
## Setup
```bash
python3 ~/.openclaw/skills/antigravity-bridge/scripts/configure.py
```
This creates `~/.openclaw/antigravity-bridge.json` pointing to your Antigravity knowledge directory and project.
## Usage
Talk to your OpenClaw agent naturally:
| You say | What happens |
|---|---|
| "Sync with Antigravity" | Pulls all KIs, tasks, memory into context |
| "Pick up next task" | Selects a `[ ]` task, prepares coding brief |
| "Self-improve" | Updates both OpenClaw and Antigravity knowledge |
| "What did Antigravity do?" | Shows task diffs and new learnings |
| "Set up the reverse bridge" | Creates Antigravity skill for OpenClaw delegation |
## Prerequisites
- Google Antigravity IDE with `~/.gemini/antigravity/` directory
- A project with `.agent/` directory (Antigravity standard)
- Python 3.10+
- OpenClaw installed and running
## Architecture
See [references/architecture.md](references/architecture.md) for detailed documentation of Antigravity's knowledge system and how the bridge maps between both systems.
## Security & Privacy
- **100% local.** No external API calls. No cloud sync. No telemetry.
- Scripts only read/write within your configured directories.
- No credentials or tokens required.
- The skill never modifies source code — only knowledge, task, and memory files.
## Contributing
Issues and PRs welcome. This is the first multi-agent knowledge bridge — there's a lot of room to improve.
## License
MIT
```
### _meta.json
```json
{
"owner": "heintonny",
"slug": "antigravity-bridge",
"displayName": "Antigravity Bridge",
"latest": {
"version": "2.0.1",
"publishedAt": 1773530225653,
"commit": "https://github.com/openclaw/skills/commit/0a0e57be2117d02eb155e8f338e6a95a3c3ee0bb"
},
"history": [
{
"version": "1.2.2",
"publishedAt": 1772922871662,
"commit": "https://github.com/openclaw/skills/commit/27d7064d605645d46d69b0ed7ddeb900b07080ba"
},
{
"version": "1.1.0",
"publishedAt": 1772838002202,
"commit": "https://github.com/openclaw/skills/commit/63e4e340b84c520bcaff003ec58a1a4438384bc0"
},
{
"version": "1.0.1",
"publishedAt": 1772836464679,
"commit": "https://github.com/openclaw/skills/commit/53e1b28431776d1f93851fa639a92fcec74c632e"
}
]
}
```
### references/architecture.md
```markdown
# Antigravity Bridge — Architecture Reference
## Antigravity Knowledge System
Google Antigravity IDE uses a two-tier knowledge system:
### Tier 1: Native Knowledge Items (KIs)
**Location:** `~/.gemini/antigravity/knowledge/`
Permanent, distilled knowledge organized by topic:
```
knowledge/
├── <topic_name>/
│ ├── metadata.json # title, summary, conversation references
│ ├── timestamps.json # creation/update timestamps
│ └── artifacts/ # detailed markdown documents
│ ├── overview.md
│ ├── patterns.md
│ └── hazards/
│ └── specific_hazard.md
└── knowledge.lock # prevents concurrent writes
```
**metadata.json format:**
```json
{
"title": "Topic Title",
"summary": "What this knowledge covers...",
"references": [
{"type": "conversation_id", "value": "<uuid>"},
{"type": "file", "value": "/path/to/relevant/file"}
]
}
```
**timestamps.json format:**
```json
{
"created": "2026-01-15T10:30:00Z",
"updated": "2026-03-06T22:00:00Z"
}
```
### Tier 2: Repo-Local Agent Config
**Location:** `<project>/.agent/`
Project-specific, git-tracked knowledge:
```
.agent/
├── tasks.md # SSoT for project roadmap
│ # [x] done, [ ] todo, [>] active, [-] skipped
├── rules/ # Always-on behavior rules
│ ├── go-standards.md
│ ├── typescript-standards.md
│ └── ...
├── skills/ # Contextual knowledge packages
│ └── <skill-name>/
│ └── SKILL.md
├── workflows/ # Saved prompts (slash commands)
│ ├── next-task.md
│ ├── finish.md
│ ├── self-improve.md
│ └── ...
├── sessions/ # Continuation prompts
│ └── next-session-*.md
└── memory/ # Lessons learned (git-tracked)
└── lessons-learned-*.md
```
### Supporting Systems
**Brain:** `~/.gemini/antigravity/brain/<uuid>/`
Session artifacts (tasks, plans, walkthroughs, screenshots).
Ephemeral — linked to KIs via conversation_id in metadata.json.
**Code Tracker:** `~/.gemini/antigravity/code_tracker/`
Content-hash snapshots of files modified by the agent. Internal to Antigravity.
**GEMINI.md:** `<project>/.gemini/GEMINI.md`
Global project context loaded every session.
## Knowledge Flow
```
Session Start:
GEMINI.md → KI summaries → /next-task workflow
→ reads rules → reads tasks.md → reads git log
→ reads memory → reads skills → recommends tasks
During Session:
Agent works → creates brain artifacts
→ references Context7 for docs → code_tracker snapshots
Session End (/finish):
1. Run tests
2. /audit-agent-memory (grep stale refs)
3. /self-improve (lessons → memory → promote to KIs)
4. /update-docs
5. /commit
6. /deploy (if cluster running)
7. Generate continuation prompt if context > 70%
```
## Communication Channels
### Antigravity → OpenClaw
1. `curl localhost:18789/api/message` (Antigravity has terminal access)
2. File drop in `~/.openclaw/workspace/inbox/`
3. Git commit → OpenClaw heartbeat detects changes
### OpenClaw → Antigravity
1. Write to `.agent/sessions/` (read at session start)
2. Write to `.agent/memory/` (visible via /self-improve)
3. Write to `knowledge/` (native KI system)
4. Update `tasks.md` (visible via /next-task)
## Mapping: Antigravity ↔ OpenClaw Concepts
| Antigravity | OpenClaw |
|---|---|
| Knowledge Items (KIs) | MEMORY.md |
| .agent/memory/ | memory/*.md |
| .agent/tasks.md | Task tracking (manual/memory) |
| .agent/rules/ | AGENTS.md, SOUL.md |
| .agent/skills/ | OpenClaw skills |
| .agent/workflows/ | Heartbeat, cron jobs |
| .gemini/GEMINI.md | Workspace context files |
| brain/ (sessions) | Daily memory logs |
| /self-improve | Nightly consolidation cron |
| /next-task | Session startup sequence |
```
### references/ki-format.md
```markdown
# Knowledge Item (KI) Format Reference
## Directory Structure
Each KI lives in its own topic directory:
```
knowledge/<topic_snake_case>/
├── metadata.json
├── timestamps.json
└── artifacts/
├── overview.md
├── <artifact_name>.md
└── <subdirectory>/
└── <artifact_name>.md
```
## metadata.json
```json
{
"title": "Human-Readable Topic Title",
"summary": "Comprehensive summary of what this knowledge covers. Should be detailed enough for the agent to decide whether to load full artifacts. 2-4 sentences.",
"references": [
{
"type": "conversation_id",
"value": "uuid-of-antigravity-session"
},
{
"type": "file",
"value": "/absolute/path/to/relevant/file"
}
]
}
```
## timestamps.json
```json
{
"created": "2026-01-15T10:30:00Z",
"updated": "2026-03-06T22:00:00Z"
}
```
## Artifact Files
Markdown documents with detailed knowledge. No required format, but common patterns:
- `overview.md` — high-level summary of the topic
- `patterns.md` — implementation patterns and examples
- `hazards/*.md` — known pitfalls and warnings
- `lifecycle/*.md` — process documentation
- `implementation/*.md` — technical details
## Topic Naming
Use `snake_case` for topic directory names:
- ✅ `authentication_and_identity`
- ✅ `frontend_design_and_standards`
- ❌ `auth` (too vague)
- ❌ `Frontend-Design` (wrong case)
## Writing Guidelines
1. **Summaries power discovery.** The agent reads summaries to decide which KIs to load. Make them descriptive.
2. **Artifacts are self-contained.** Each artifact should be understandable without reading others.
3. **Reference, don't duplicate.** Link to other KIs or memory files instead of copying content.
4. **Date your updates.** Include "Updated: YYYY-MM-DD" in artifacts that change frequently.
5. **Respect knowledge.lock.** Check for the lock file before writing. Remove after writing.
```
### scripts/configure.py
```python
#!/usr/bin/env python3
"""Configure antigravity-bridge paths interactively."""
# SECURITY MANIFEST:
# Environment variables accessed: HOME (only)
# External endpoints called: none
# Local files read: ~/.openclaw/antigravity-bridge.json (if exists)
# Local files written: ~/.openclaw/antigravity-bridge.json
import json
import os
import sys
CONFIG_PATH = os.path.expanduser("~/.openclaw/antigravity-bridge.json")
DEFAULT_KNOWLEDGE = "~/.gemini/antigravity/knowledge"
def prompt(msg: str, default: str = "") -> str:
suffix = f" [{default}]" if default else ""
val = input(f"{msg}{suffix}: ").strip()
return val if val else default
def main():
print("🌉 Antigravity Bridge — Configuration\n")
existing = {}
if os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH) as f:
existing = json.load(f)
print(f"Existing config found at {CONFIG_PATH}\n")
knowledge_dir = prompt(
"Antigravity knowledge directory",
existing.get("knowledge_dir", DEFAULT_KNOWLEDGE),
)
project_dir = prompt(
"Project directory (with .agent/)",
existing.get("project_dir", ""),
)
if not project_dir:
print("Error: project directory is required.")
sys.exit(1)
agent_dir = os.path.join(project_dir, ".agent")
gemini_md = os.path.join(project_dir, ".gemini", "GEMINI.md")
# Validate paths
warnings = []
expanded_knowledge = os.path.expanduser(knowledge_dir)
expanded_project = os.path.expanduser(project_dir)
expanded_agent = os.path.expanduser(agent_dir)
if not os.path.isdir(expanded_knowledge):
warnings.append(f"⚠️ Knowledge dir not found: {knowledge_dir}")
if not os.path.isdir(expanded_project):
warnings.append(f"⚠️ Project dir not found: {project_dir}")
if not os.path.isdir(expanded_agent):
warnings.append(f"⚠️ Agent dir not found: {agent_dir}")
if not os.path.isfile(os.path.expanduser(gemini_md)):
warnings.append(f"⚠️ GEMINI.md not found: {gemini_md}")
if warnings:
print("\nValidation warnings:")
for w in warnings:
print(f" {w}")
cont = input("\nContinue anyway? (yes/no) [yes]: ").strip()
if cont.lower() in ("no", "n"):
print("Aborted.")
sys.exit(0)
auto_commit = prompt(
"Auto-commit changes to git? (yes/no)",
"yes" if existing.get("auto_commit") else "no",
)
config = {
"knowledge_dir": knowledge_dir,
"project_dir": project_dir,
"agent_dir": agent_dir,
"gemini_md": gemini_md,
"auto_commit": auto_commit.lower() in ("yes", "y", "true"),
}
os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
with open(CONFIG_PATH, "w") as f:
json.dump(config, f, indent=2)
print(f"\n✅ Config saved to {CONFIG_PATH}")
print(json.dumps(config, indent=2))
if __name__ == "__main__":
main()
```
### scripts/create_agy_skill.py
```python
#!/usr/bin/env python3
"""Generate the reverse-direction Antigravity skill for OpenClaw delegation.
Creates .agent/skills/openclaw-bridge/SKILL.md in your project, teaching
Antigravity how to delegate tasks to OpenClaw agents.
"""
# SECURITY MANIFEST:
# Environment variables accessed: HOME (only)
# External endpoints called: none
# Local files read: ~/.openclaw/antigravity-bridge.json
# Local files written: <project>/.agent/skills/openclaw-bridge/SKILL.md
import json
import os
import sys
from pathlib import Path
CONFIG_PATH = os.path.expanduser("~/.openclaw/antigravity-bridge.json")
SKILL_MD = '''---
description: >-
Delegate tasks to OpenClaw agents running on the same machine.
Use when: (1) you need background work done while continuing your session,
(2) tasks that don't require IDE access (monitoring, notifications, API calls),
(3) parallel coding via Claude Code sub-agents,
(4) cross-system knowledge sync,
(5) user says "delegate to openclaw", "send to clawd", or "background task".
---
# OpenClaw Bridge
Delegate tasks to OpenClaw agents for parallel or background execution.
## Quick Delegation
Send a task to OpenClaw via terminal:
```bash
curl -s http://localhost:18789/api/message \\
-H "Content-Type: application/json" \\
-d '{"message": "<task description>"}'
```
## File-Based Handoff
For complex tasks with context:
```bash
cat > ~/.openclaw/workspace/inbox/task-$(date +%s).md << 'TASK'
# Task from Antigravity
<detailed task description>
## Context
- Source: Antigravity session
- Branch: $(git branch --show-current)
- Priority: <high/medium/low>
- Related files: <list of files>
## Expected Output
- <what OpenClaw should produce>
TASK
```
OpenClaw picks up inbox tasks via heartbeat polling.
## Task Types to Delegate
| Task | Why OpenClaw? |
|---|---|
| Run E2E tests | Headless sub-agents, no IDE needed |
| Monitor deployment | Has Cloudflare + kubectl integrations |
| Send notifications | Telegram/email/Slack channels configured |
| Update docs | Parallel work while you code |
| Security audit | Healthcheck skill built-in |
| Sync knowledge | Updates both agent systems |
| Research | Web search + Context7 + browser relay |
## Triggering Knowledge Sync
Sync both Antigravity and OpenClaw knowledge:
```bash
curl -s http://localhost:18789/api/message \\
-H "Content-Type: application/json" \\
-d '{"message": "Run antigravity bridge self-improve"}'
```
## Receiving Results
OpenClaw writes results to:
- `.agent/sessions/openclaw-result-<timestamp>.md` (for next Antigravity session)
- `.agent/memory/lessons-learned-*.md` (if learnings captured)
- Telegram notification to the user (if configured)
## When NOT to Delegate
- Tasks requiring IDE UI interaction (use Antigravity directly)
- Tasks needing Antigravity's Gemini model specifically
- Quick edits that are faster to do in-session
'''
def main():
config_path = CONFIG_PATH
if not os.path.exists(config_path):
print(f"Error: Config not found at {config_path}")
sys.exit(1)
with open(config_path) as f:
config = json.load(f)
# Create in project's .agent/skills/
agent_dir = Path(os.path.expanduser(config["agent_dir"]))
skill_dir = agent_dir / "skills" / "openclaw-bridge"
skill_dir.mkdir(parents=True, exist_ok=True)
skill_path = skill_dir / "SKILL.md"
skill_path.write_text(SKILL_MD)
print(f"✅ Antigravity skill created: {skill_path}")
# Also create globally if knowledge dir exists
knowledge_dir = Path(os.path.expanduser(config["knowledge_dir"]))
if knowledge_dir.exists():
global_skill_dir = knowledge_dir.parent / "skills" / "openclaw-bridge"
global_skill_dir.mkdir(parents=True, exist_ok=True)
global_skill_path = global_skill_dir / "SKILL.md"
global_skill_path.write_text(SKILL_MD)
print(f"✅ Global skill created: {global_skill_path}")
print("\\n🌉 Antigravity can now delegate tasks to OpenClaw!")
print(" Use in Antigravity: 'delegate this to OpenClaw' or run the curl command.")
if __name__ == "__main__":
main()
```
### scripts/diff_tasks.py
```python
#!/usr/bin/env python3
"""Show task changes since last sync."""
# SECURITY MANIFEST:
# Environment variables accessed: HOME (only)
# External endpoints called: none
# Local files read: .agent/tasks.md, ~/.openclaw/antigravity-bridge-state.json
# Local files written: ~/.openclaw/antigravity-bridge-state.json
import hashlib
import json
import os
import re
import sys
from pathlib import Path
CONFIG_PATH = os.path.expanduser("~/.openclaw/antigravity-bridge.json")
STATE_PATH = os.path.expanduser("~/.openclaw/antigravity-bridge-state.json")
def load_config() -> dict:
if not os.path.exists(CONFIG_PATH):
print(f"Error: Config not found at {CONFIG_PATH}")
sys.exit(1)
with open(CONFIG_PATH) as f:
return json.load(f)
def load_state() -> dict:
if os.path.exists(STATE_PATH):
with open(STATE_PATH) as f:
return json.load(f)
return {}
def save_state(state: dict):
with open(STATE_PATH, "w") as f:
json.dump(state, f, indent=2)
def parse_tasks(content: str) -> dict[str, str]:
"""Parse tasks into {task_text: status} dict."""
tasks = {}
pattern = re.compile(r"- \[(.)\] \*\*(.+?)\*\*")
for line in content.splitlines():
m = pattern.search(line)
if m:
status, name = m.group(1), m.group(2)
tasks[name] = status
return tasks
def main():
config = load_config()
state = load_state()
tasks_path = Path(os.path.expanduser(config["agent_dir"])) / "tasks.md"
if not tasks_path.exists():
print("No tasks.md found")
sys.exit(1)
content = tasks_path.read_text()
current_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
previous_hash = state.get("tasks_hash", "")
if current_hash == previous_hash:
print("📋 No changes since last sync.")
return
current_tasks = parse_tasks(content)
previous_tasks = state.get("tasks_snapshot", {})
# Find changes
newly_done = []
newly_active = []
newly_added = []
for name, status in current_tasks.items():
prev_status = previous_tasks.get(name)
if prev_status is None:
newly_added.append(name)
elif prev_status != "x" and status == "x":
newly_done.append(name)
elif prev_status != ">" and status == ">":
newly_active.append(name)
if newly_done:
print("✅ Newly completed:")
for t in newly_done:
print(f" - {t}")
if newly_active:
print("🔄 Newly active:")
for t in newly_active:
print(f" - {t}")
if newly_added:
print("🆕 New tasks:")
for t in newly_added:
print(f" - {t}")
if not (newly_done or newly_active or newly_added):
print("📝 Tasks changed but no status transitions detected.")
# Update state
state["tasks_hash"] = current_hash
state["tasks_snapshot"] = current_tasks
save_state(state)
print(f"\n📌 State saved. {len(current_tasks)} tasks tracked.")
if __name__ == "__main__":
main()
```
### scripts/pick_task.py
```python
#!/usr/bin/env python3
"""Gather project context for next-task selection.
Mirrors Antigravity's /next-task workflow pattern:
1. Reads tasks, git log, memory, sessions, rules, skills
2. Outputs structured JSON context
3. Agent (LLM) reasons about which tasks to recommend
4. Agent presents 2-3 options to user
5. User picks → agent spawns sub-agent
This script does NOT modify tasks.md or pick a task.
That's the agent's job after user selection.
"""
# SECURITY MANIFEST:
# Environment variables accessed: HOME (only)
# External endpoints called: none
# Local files read: .agent/tasks.md, .agent/memory/*, .agent/sessions/*,
# .agent/rules/*, .agent/skills/*, .gemini/GEMINI.md, git log
# Local files written: none
import json
import os
import re
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from utils import load_config
def read_tasks(agent_dir: str) -> dict:
"""Read tasks.md and extract structured task data."""
tasks_path = Path(os.path.expanduser(agent_dir)) / "tasks.md"
if not tasks_path.exists():
return {"exists": False}
content = tasks_path.read_text()
lines = content.splitlines()
stats = {
"done": content.count("[x]"),
"todo": content.count("[ ]"),
"active": content.count("[>]"),
"skipped": content.count("[-]"),
}
# Extract active tasks
active_tasks = []
for i, line in enumerate(lines):
if "[>]" in line:
active_tasks.append({
"line_num": i + 1,
"text": line.strip().lstrip("- "),
})
# Extract phases/sections with their todo tasks
phases = []
current_phase = None
for i, line in enumerate(lines):
# Detect phase headers (## or ### with phase-like content)
if re.match(r'^#{1,3}\s+', line) and not line.startswith('####'):
current_phase = {
"header": line.strip().lstrip("#").strip(),
"line_num": i + 1,
"todo_tasks": [],
"done_count": 0,
"todo_count": 0,
}
phases.append(current_phase)
elif current_phase:
if "[x]" in line:
current_phase["done_count"] += 1
elif "[ ]" in line:
current_phase["todo_count"] += 1
# Only include first-level tasks (not sub-tasks indented too deep)
indent = len(line) - len(line.lstrip())
if indent <= 4: # Top-level or one indent
task_text = re.sub(r'^[\s\-]*\[\s*\]\s*', '', line).strip()
if task_text:
current_phase["todo_tasks"].append({
"line_num": i + 1,
"text": task_text,
})
# Filter to phases with remaining work
phases_with_work = [p for p in phases if p["todo_count"] > 0]
return {
"exists": True,
"stats": stats,
"active_tasks": active_tasks,
"phases_with_work": phases_with_work[:10], # Top 10 phases
}
def read_git_log(project_dir: str, count: int = 15) -> list[str]:
"""Read recent git commits."""
try:
result = subprocess.run(
["git", "log", f"--oneline", f"-{count}", "--no-decorate"],
cwd=os.path.expanduser(project_dir),
capture_output=True, text=True, timeout=10
)
if result.returncode == 0:
return result.stdout.strip().splitlines()
except (subprocess.TimeoutExpired, FileNotFoundError):
pass
return []
def read_session_handoffs(agent_dir: str) -> list[dict]:
"""Read session continuation prompts."""
sessions_dir = Path(os.path.expanduser(agent_dir)) / "sessions"
if not sessions_dir.exists():
return []
handoffs = []
for f in sorted(sessions_dir.glob("*.md")):
content = f.read_text()
handoffs.append({
"file": f.name,
"preview": content[:800],
})
return handoffs
def read_recent_memory(agent_dir: str, max_files: int = 10) -> list[dict]:
"""Read recent lessons learned (names + first line)."""
memory_dir = Path(os.path.expanduser(agent_dir)) / "memory"
if not memory_dir.exists():
return []
files = sorted(memory_dir.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True)
result = []
for f in files[:max_files]:
first_line = ""
with open(f) as fh:
for line in fh:
line = line.strip()
if line and not line.startswith("#"):
first_line = line[:200]
break
result.append({
"file": f.name,
"first_line": first_line,
})
return result
def list_directory(agent_dir: str, subdir: str) -> list[str]:
"""List files in an .agent/ subdirectory."""
dir_path = Path(os.path.expanduser(agent_dir)) / subdir
if not dir_path.exists():
return []
if subdir == "skills":
# Return skill directory names
return sorted([d.name for d in dir_path.iterdir() if d.is_dir() and not d.name.startswith(".")])
return sorted([f.name for f in dir_path.glob("*.md")])
def main():
config = load_config()
project_dir = config["project_dir"]
agent_dir = config["agent_dir"]
context = {
"gathered_at": datetime.now().isoformat(),
"project_dir": project_dir,
"tasks": read_tasks(agent_dir),
"recent_commits": read_git_log(project_dir),
"session_handoffs": read_session_handoffs(agent_dir),
"recent_memory": read_recent_memory(agent_dir),
"rules": list_directory(agent_dir, "rules"),
"skills": list_directory(agent_dir, "skills"),
"workflows": list_directory(agent_dir, "workflows"),
}
print(json.dumps(context, indent=2))
if __name__ == "__main__":
main()
```
### scripts/self_improve.py
```python
#!/usr/bin/env python3
"""Cross-agent self-improvement: update both OpenClaw and Antigravity knowledge systems."""
# SECURITY MANIFEST:
# Environment variables accessed: HOME (only)
# External endpoints called: none
# Local files read: config
# Local files written: MEMORY.md, memory/YYYY-MM-DD.md, .agent/memory/lessons-learned-*.md,
# ~/.gemini/antigravity/knowledge/<topic>/metadata.json,
# ~/.gemini/antigravity/knowledge/<topic>/timestamps.json,
# ~/.gemini/antigravity/knowledge/<topic>/artifacts/*.md
import argparse
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
CONFIG_PATH = os.path.expanduser("~/.openclaw/antigravity-bridge.json")
def load_config() -> dict:
if not os.path.exists(CONFIG_PATH):
print(f"Error: Config not found at {CONFIG_PATH}")
sys.exit(1)
with open(CONFIG_PATH) as f:
return json.load(f)
def write_ki(knowledge_dir: str, topic: str, title: str, summary: str,
artifact_name: str, content: str):
"""Write or update a Knowledge Item in Antigravity's native format."""
ki_dir = Path(os.path.expanduser(knowledge_dir)) / topic
artifacts_dir = ki_dir / "artifacts"
artifacts_dir.mkdir(parents=True, exist_ok=True)
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
# metadata.json
meta_path = ki_dir / "metadata.json"
if meta_path.exists():
try:
with open(meta_path) as f:
meta = json.load(f)
except json.JSONDecodeError:
with open(meta_path) as f:
raw = f.read().strip()
try:
meta = json.loads(raw[:raw.rindex("}") + 1])
except (ValueError, json.JSONDecodeError):
meta = {"title": title, "summary": summary, "references": []}
# Append to summary if new info
if summary not in meta.get("summary", ""):
meta["summary"] = meta["summary"].rstrip(". ") + ". " + summary
else:
meta = {
"title": title,
"summary": summary,
"references": []
}
with open(meta_path, "w") as f:
json.dump(meta, f, indent=4)
# timestamps.json
ts_path = ki_dir / "timestamps.json"
ts = {"updated": now}
if ts_path.exists():
with open(ts_path) as f:
ts = json.load(f)
ts["updated"] = now
else:
ts["created"] = now
with open(ts_path, "w") as f:
json.dump(ts, f, indent=4)
# artifact
artifact_path = artifacts_dir / f"{artifact_name}.md"
with open(artifact_path, "w") as f:
f.write(content)
print(f" 📚 KI updated: knowledge/{topic}/artifacts/{artifact_name}.md")
def write_agent_memory(agent_dir: str, topic: str, lesson: str):
"""Write a lesson-learned file to .agent/memory/."""
memory_dir = Path(os.path.expanduser(agent_dir)) / "memory"
memory_dir.mkdir(parents=True, exist_ok=True)
filename = f"lessons-learned-{topic.replace('_', '-')}.md"
filepath = memory_dir / filename
now = datetime.now().strftime("%Y-%m-%d")
if filepath.exists():
existing = filepath.read_text()
updated = existing.rstrip() + f"\n\n---\n*Updated {now} by OpenClaw/antigravity-bridge*\n\n{lesson}\n"
else:
updated = f"# Lessons Learned: {topic.replace('_', ' ').title()}\n\n*Created {now} by OpenClaw/antigravity-bridge*\n\n{lesson}\n"
filepath.write_text(updated)
print(f" 📝 Memory: .agent/memory/{filename}")
def write_openclaw_memory(lesson: str, topic: str):
"""Update OpenClaw's daily memory log."""
workspace = os.path.expanduser("~/.openclaw/workspace")
memory_dir = os.path.join(workspace, "memory")
os.makedirs(memory_dir, exist_ok=True)
today = datetime.now().strftime("%Y-%m-%d")
now = datetime.now().strftime("%H:%M")
filepath = os.path.join(memory_dir, f"{today}.md")
entry = f"{now} — [antigravity-bridge] Self-improve: {topic} — {lesson[:200]}\n"
if os.path.exists(filepath):
with open(filepath, "a") as f:
f.write(entry)
else:
with open(filepath, "w") as f:
f.write(f"# {today}\n\n{entry}")
print(f" 🧠 OpenClaw memory: memory/{today}.md")
def main():
parser = argparse.ArgumentParser(description="Cross-agent self-improvement")
parser.add_argument("--topic", required=True, help="Knowledge topic (snake_case)")
parser.add_argument("--title", help="Human-readable title (defaults to topic)")
parser.add_argument("--summary", help="KI summary (2-4 sentences)")
parser.add_argument("--lesson", required=True, help="What was learned")
parser.add_argument("--artifact", help="Artifact filename (defaults to topic)")
parser.add_argument("--content", help="Full artifact content (reads stdin if not provided)")
args = parser.parse_args()
config = load_config()
title = args.title or args.topic.replace("_", " ").title()
summary = args.summary or args.lesson[:200]
artifact = args.artifact or args.topic
content = args.content or args.lesson
print("🔄 Cross-agent self-improvement\n")
# 1. Update Antigravity KI
print("Antigravity Knowledge Items:")
write_ki(config["knowledge_dir"], args.topic, title, summary, artifact, content)
# 2. Update .agent/memory/
print("\nAntigravity Agent Memory:")
write_agent_memory(config["agent_dir"], args.topic, args.lesson)
# 3. Update OpenClaw memory
print("\nOpenClaw Memory:")
write_openclaw_memory(args.lesson, args.topic)
print("\n✅ Both knowledge systems updated.")
# 4. Git commit if configured
if config.get("auto_commit"):
project_dir = os.path.expanduser(config["project_dir"])
os.system(f'cd "{project_dir}" && git add .agent/memory/ && git commit -m "chore: self-improve — {args.topic}" --no-verify 2>/dev/null')
print("📦 Changes committed to git.")
if __name__ == "__main__":
main()
```
### scripts/sync_knowledge.py
```python
#!/usr/bin/env python3
"""Sync Antigravity knowledge into OpenClaw context.
v1.2.0: Produces a structured state file + diff output.
The agent reads the diff and updates MEMORY.md / daily logs.
Outputs JSON to stdout with:
- current: full current state snapshot
- diff: what changed since last sync
- is_first_sync: true if no previous state exists
"""
# SECURITY MANIFEST:
# Environment variables accessed: HOME (only)
# External endpoints called: none
# Local files read: ~/.gemini/antigravity/knowledge/*, .agent/tasks.md,
# .agent/memory/*, .agent/sessions/*, .agent/rules/*, .agent/skills/*,
# .agent/workflows/*, .gemini/GEMINI.md, antigravity-sync-state.json
# Local files written: <workspace>/antigravity-sync-state.json
import hashlib
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from utils import load_config, safe_load_json
STATE_PATH = os.path.expanduser("~/.openclaw/workspace/antigravity-sync-state.json")
def file_hash(path: str) -> str:
"""MD5 hash of file contents for change detection."""
try:
return hashlib.md5(Path(path).read_bytes()).hexdigest()
except (OSError, IOError):
return ""
def read_ki_topics(knowledge_dir: str) -> dict:
"""Read all Knowledge Item metadata into a structured dict."""
ki_dir = Path(os.path.expanduser(knowledge_dir))
if not ki_dir.exists():
return {}
topics = {}
for topic_dir in sorted(ki_dir.iterdir()):
if not topic_dir.is_dir() or topic_dir.name.startswith("."):
continue
meta_path = topic_dir / "metadata.json"
if not meta_path.exists():
continue
meta = safe_load_json(str(meta_path))
if not meta:
continue
# Count artifacts
artifact_dir = topic_dir / "artifacts"
artifacts = []
if artifact_dir.exists():
artifacts = [str(a.relative_to(artifact_dir)) for a in sorted(artifact_dir.rglob("*.md"))]
# Get timestamps
ts_path = topic_dir / "timestamps.json"
updated = ""
if ts_path.exists():
ts = safe_load_json(str(ts_path))
if ts:
updated = ts.get("updated", "")
topics[topic_dir.name] = {
"title": meta.get("title", topic_dir.name),
"summary": meta.get("summary", ""),
"artifact_count": len(artifacts),
"artifacts": artifacts,
"updated": updated,
"meta_hash": file_hash(str(meta_path)),
}
return topics
def read_tasks(agent_dir: str) -> dict:
"""Read tasks.md and extract stats + active tasks."""
tasks_path = Path(os.path.expanduser(agent_dir)) / "tasks.md"
if not tasks_path.exists():
return {"exists": False}
content = tasks_path.read_text()
done = content.count("[x]")
todo = content.count("[ ]")
active = content.count("[>]")
skipped = content.count("[-]")
active_tasks = []
for line in content.splitlines():
if "[>]" in line:
active_tasks.append(line.strip().lstrip("- "))
return {
"exists": True,
"done": done,
"todo": todo,
"active": active,
"skipped": skipped,
"active_tasks": active_tasks,
"file_hash": file_hash(str(tasks_path)),
}
def read_directory_inventory(base_dir: str, subdir: str) -> dict:
"""Read a directory and return file names + hashes."""
dir_path = Path(os.path.expanduser(base_dir)) / subdir
if not dir_path.exists():
return {}
inventory = {}
for f in sorted(dir_path.rglob("*.md")):
rel = str(f.relative_to(dir_path))
inventory[rel] = {
"hash": file_hash(str(f)),
"modified": datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
"size": f.stat().st_size,
}
return inventory
def read_memory_previews(agent_dir: str, max_files: int = 10) -> dict:
"""Read memory files with previews for new file detection."""
memory_dir = Path(os.path.expanduser(agent_dir)) / "memory"
if not memory_dir.exists():
return {}
files = sorted(memory_dir.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True)
result = {}
for f in files[:max_files]:
content = f.read_text()
result[f.name] = {
"hash": file_hash(str(f)),
"modified": datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
"preview": content[:300],
}
return result
def read_sessions(agent_dir: str) -> dict:
"""Read session handoff files."""
sessions_dir = Path(os.path.expanduser(agent_dir)) / "sessions"
if not sessions_dir.exists():
return {}
result = {}
for f in sorted(sessions_dir.glob("*.md")):
content = f.read_text()
result[f.name] = {
"hash": file_hash(str(f)),
"modified": datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
"preview": content[:500],
}
return result
def load_previous_state() -> dict | None:
"""Load previous sync state."""
if not os.path.exists(STATE_PATH):
return None
try:
with open(STATE_PATH) as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
return None
def compute_diff(previous: dict | None, current: dict) -> dict:
"""Compute what changed between previous and current state."""
if previous is None:
return {
"is_first_sync": True,
"summary": "First sync — all data is new.",
"ki_topics": {"added": list(current["ki_topics"].keys()), "removed": [], "changed": []},
"tasks": {"changed": True, "details": current["tasks"]},
"memory": {"added": list(current["memory"].keys()), "removed": [], "changed": []},
"rules": {"added": list(current["rules"].keys()), "removed": [], "changed": []},
"skills": {"added": list(current["skills"].keys()), "removed": [], "changed": []},
"workflows": {"added": list(current["workflows"].keys()), "removed": [], "changed": []},
"sessions": {"added": list(current["sessions"].keys()), "removed": [], "changed": []},
}
diff = {"is_first_sync": False, "changes": []}
# KI topics diff
prev_ki = set(previous.get("ki_topics", {}).keys())
curr_ki = set(current["ki_topics"].keys())
ki_added = sorted(curr_ki - prev_ki)
ki_removed = sorted(prev_ki - curr_ki)
ki_changed = []
for topic in sorted(curr_ki & prev_ki):
prev_hash = previous["ki_topics"][topic].get("meta_hash", "")
curr_hash = current["ki_topics"][topic].get("meta_hash", "")
prev_count = previous["ki_topics"][topic].get("artifact_count", 0)
curr_count = current["ki_topics"][topic].get("artifact_count", 0)
if prev_hash != curr_hash or prev_count != curr_count:
ki_changed.append({
"topic": topic,
"title": current["ki_topics"][topic]["title"],
"artifacts_before": prev_count,
"artifacts_after": curr_count,
})
diff["ki_topics"] = {"added": ki_added, "removed": ki_removed, "changed": ki_changed}
if ki_added or ki_removed or ki_changed:
diff["changes"].append(f"KI: +{len(ki_added)} added, -{len(ki_removed)} removed, ~{len(ki_changed)} changed")
# Tasks diff
prev_tasks = previous.get("tasks", {})
curr_tasks = current["tasks"]
tasks_changed = prev_tasks.get("file_hash") != curr_tasks.get("file_hash")
task_details = {}
if tasks_changed and prev_tasks.get("exists") and curr_tasks.get("exists"):
task_details = {
"done_delta": curr_tasks["done"] - prev_tasks.get("done", 0),
"todo_delta": curr_tasks["todo"] - prev_tasks.get("todo", 0),
"done_before": prev_tasks.get("done", 0),
"done_after": curr_tasks["done"],
"todo_before": prev_tasks.get("todo", 0),
"todo_after": curr_tasks["todo"],
"active_tasks": curr_tasks.get("active_tasks", []),
}
diff["changes"].append(
f"Tasks: {task_details['done_delta']:+d} done, {task_details['todo_delta']:+d} todo "
f"(now {curr_tasks['done']} done / {curr_tasks['todo']} todo)"
)
diff["tasks"] = {"changed": tasks_changed, "details": task_details}
# Generic directory diffs
for key in ["memory", "rules", "skills", "workflows", "sessions"]:
prev_items = set(previous.get(key, {}).keys())
curr_items = set(current[key].keys())
added = sorted(curr_items - prev_items)
removed = sorted(prev_items - curr_items)
changed = []
for item in sorted(curr_items & prev_items):
prev_hash = previous[key][item].get("hash", "")
curr_hash = current[key][item].get("hash", "")
if prev_hash != curr_hash:
changed.append(item)
diff[key] = {"added": added, "removed": removed, "changed": changed}
if added or removed or changed:
diff["changes"].append(f"{key.capitalize()}: +{len(added)} added, -{len(removed)} removed, ~{len(changed)} changed")
# Summary
if diff["changes"]:
diff["summary"] = " | ".join(diff["changes"])
else:
diff["summary"] = "No changes since last sync."
return diff
def save_state(current: dict):
"""Save current state for future diffing."""
state = {
"synced_at": datetime.now().isoformat(),
"ki_topics": current["ki_topics"],
"tasks": current["tasks"],
"memory": {k: {"hash": v["hash"]} for k, v in current["memory"].items()},
"rules": {k: {"hash": v["hash"]} for k, v in current["rules"].items()},
"skills": {k: {"hash": v["hash"]} for k, v in current["skills"].items()},
"workflows": {k: {"hash": v["hash"]} for k, v in current["workflows"].items()},
"sessions": {k: {"hash": v["hash"]} for k, v in current["sessions"].items()},
}
os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
with open(STATE_PATH, "w") as f:
json.dump(state, f, indent=2)
def main():
config = load_config()
# Gather current state
current = {
"ki_topics": read_ki_topics(config["knowledge_dir"]),
"tasks": read_tasks(config["agent_dir"]),
"memory": read_memory_previews(config["agent_dir"]),
"rules": read_directory_inventory(config["agent_dir"], "rules"),
"skills": read_directory_inventory(config["agent_dir"], "skills"),
"workflows": read_directory_inventory(config["agent_dir"], "workflows"),
"sessions": read_sessions(config["agent_dir"]),
}
# Load previous state and compute diff
previous = load_previous_state()
diff = compute_diff(previous, current)
# Save new state
save_state(current)
# Output structured result to stdout
output = {
"synced_at": datetime.now().isoformat(),
"diff": diff,
"current": {
"ki_topic_count": len(current["ki_topics"]),
"ki_topics": {k: {"title": v["title"], "summary": v["summary"], "artifact_count": v["artifact_count"]}
for k, v in current["ki_topics"].items()},
"tasks": current["tasks"],
"memory_count": len(current["memory"]),
"memory_files": list(current["memory"].keys()),
"rules_count": len(current["rules"]),
"rules_files": list(current["rules"].keys()),
"skills_count": len(current["skills"]),
"skills_dirs": list(set(f.split("/")[0] for f in current["skills"].keys())),
"workflows_count": len(current["workflows"]),
"workflows_files": list(current["workflows"].keys()),
"sessions": {k: {"preview": v["preview"]} for k, v in current["sessions"].items()},
},
}
print(json.dumps(output, indent=2))
if __name__ == "__main__":
main()
```
### scripts/utils.py
```python
"""Shared utilities for antigravity-bridge scripts."""
import json
import os
CONFIG_PATH = os.path.expanduser("~/.openclaw/antigravity-bridge.json")
def load_config() -> dict:
if not os.path.exists(CONFIG_PATH):
raise FileNotFoundError(
f"Config not found at {CONFIG_PATH}\nRun: python3 scripts/configure.py"
)
with open(CONFIG_PATH) as f:
return json.load(f)
def safe_load_json(path: str) -> dict | None:
"""Load JSON, handling trailing markdown artifacts."""
try:
with open(path) as f:
return json.load(f)
except json.JSONDecodeError:
with open(path) as f:
raw = f.read().strip()
try:
last_brace = raw.rindex("}")
return json.loads(raw[:last_brace + 1])
except (ValueError, json.JSONDecodeError):
return None
```
### scripts/write_ki.py
```python
#!/usr/bin/env python3
"""Write directly to Antigravity's native Knowledge Item system."""
# SECURITY MANIFEST:
# Environment variables accessed: HOME (only)
# External endpoints called: none
# Local files read: ~/.openclaw/antigravity-bridge.json
# Local files written: ~/.gemini/antigravity/knowledge/<topic>/*
import argparse
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
CONFIG_PATH = os.path.expanduser("~/.openclaw/antigravity-bridge.json")
def load_config() -> dict:
if not os.path.exists(CONFIG_PATH):
print(f"Error: Config not found at {CONFIG_PATH}")
sys.exit(1)
with open(CONFIG_PATH) as f:
return json.load(f)
def write_ki(knowledge_dir: str, topic: str, title: str, summary: str,
artifact_name: str, content: str, references: list | None = None):
"""Write or update a Knowledge Item."""
ki_dir = Path(os.path.expanduser(knowledge_dir)) / topic
artifacts_dir = ki_dir / "artifacts"
artifacts_dir.mkdir(parents=True, exist_ok=True)
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
# metadata.json — merge with existing
meta_path = ki_dir / "metadata.json"
if meta_path.exists():
try:
with open(meta_path) as f:
meta = json.load(f)
except json.JSONDecodeError:
with open(meta_path) as f:
raw = f.read().strip()
try:
meta = json.loads(raw[:raw.rindex("}") + 1])
except (ValueError, json.JSONDecodeError):
meta = {"title": title, "summary": summary, "references": references or []}
if title:
meta["title"] = title
if summary and summary not in meta.get("summary", ""):
meta["summary"] = summary
if references:
existing_refs = {json.dumps(r, sort_keys=True) for r in meta.get("references", [])}
for ref in references:
if json.dumps(ref, sort_keys=True) not in existing_refs:
meta.setdefault("references", []).append(ref)
else:
meta = {
"title": title,
"summary": summary,
"references": references or []
}
with open(meta_path, "w") as f:
json.dump(meta, f, indent=4)
# timestamps.json
ts_path = ki_dir / "timestamps.json"
if ts_path.exists():
with open(ts_path) as f:
ts = json.load(f)
ts["updated"] = now
else:
ts = {"created": now, "updated": now}
with open(ts_path, "w") as f:
json.dump(ts, f, indent=4)
# artifact
artifact_path = artifacts_dir / f"{artifact_name}.md"
with open(artifact_path, "w") as f:
f.write(content)
print(f"✅ KI written: knowledge/{topic}/artifacts/{artifact_name}.md")
print(f" Title: {meta['title']}")
print(f" Artifacts: {len(list(artifacts_dir.rglob('*.md')))}")
def main():
parser = argparse.ArgumentParser(description="Write to Antigravity Knowledge Items")
parser.add_argument("--topic", required=True, help="Topic directory name (snake_case)")
parser.add_argument("--title", required=True, help="Human-readable title")
parser.add_argument("--summary", required=True, help="KI summary (2-4 sentences)")
parser.add_argument("--artifact", required=True, help="Artifact filename (without .md)")
parser.add_argument("--content", help="Artifact content (reads stdin if omitted)")
parser.add_argument("--ref-file", action="append", help="Add file reference")
parser.add_argument("--ref-conversation", action="append", help="Add conversation ID reference")
args = parser.parse_args()
config = load_config()
content = args.content
if content is None:
if not sys.stdin.isatty():
content = sys.stdin.read()
else:
print("Error: provide --content or pipe content via stdin")
sys.exit(1)
references = []
for ref in (args.ref_file or []):
references.append({"type": "file", "value": ref})
for ref in (args.ref_conversation or []):
references.append({"type": "conversation_id", "value": ref})
write_ki(
config["knowledge_dir"],
args.topic,
args.title,
args.summary,
args.artifact,
content,
references,
)
if __name__ == "__main__":
main()
```