From c6ce151fbb7b35d4541520f2377e7a4ed8293c50 Mon Sep 17 00:00:00 2001 From: Alfred Kolakkal Date: Sat, 1 Aug 2026 11:24:21 -0600 Subject: [PATCH] feat(sleep): Antigravity conversation harvest source Adds transcript_source: "antigravity", reading Google Antigravity's per-conversation SQLite trajectory stores (~/.gemini/antigravity/conversations/.db). Step payloads are protobuf with no published schema, so a tolerant string walker recovers user prompts (step type 14), assistant artifacts (5) and tool calls (33); schema drift degrades to fewer strings rather than failing the night. Addresses the review on #153: * Consistent reads, no predictable temp path. Databases are read through a read-only URI connection so SQLite attaches the -wal sidecar itself; copying the bare .db (the previous approach) can miss recent commits or the schema entirely while a writer is attached. Locked or unreadable stores fall back to SQLite's online backup API, snapshotting into a private TemporaryDirectory that is removed as soon as the rows are read. * Real project provenance, enforced scope. The workspace is read from trajectory_metadata_blob (file:// URI, plus git branch), and conversations Antigravity marked "outside-of-project" are recognised as such. Under invoked/list scope only conversations whose recorded workspace contains a selected project are harvested; unknown and outside-of-project sessions are skipped instead of being relabelled as the invoked project. projects: "all" is the explicit opt-in and still reports the real workspace ("" when there is none). On the corpus this was developed against, invoked scope selects 7 of 316 conversations where every one was previously attributed to the invoked project. * Secrets are redacted inside the harvester. Prompts, finals and tool names pass through staging.redact_secrets before they can reach the evidence log or any model-processing path. * Walker hardening. A nested message that happens to decode as UTF-8 no longer surfaces as text with its protobuf framing bytes attached; chunks containing control bytes are re-walked instead. Session start now comes from the recorded creation timestamp (mtime remains the end time and the since_iso key), so lookback filtering no longer treats a long conversation as having started when it last changed. tests/test_antigravity_harvest.py adds 23 tests over synthetic protobuf/SQLite fixtures: step types 14/5/33, Unicode, schema drift and unknown wire types, invoked/all/list scope, outside-of-project and missing-metadata attribution, live WAL reads, the backup fallback, locked/corrupt/truncated stores, temp-file cleanup, long-token and PEM redaction, and the config/source router. --- skillopt_sleep/__main__.py | 12 +- skillopt_sleep/config.py | 10 +- skillopt_sleep/harvest_antigravity.py | 586 ++++++++++++++++++++++++++ skillopt_sleep/harvest_sources.py | 9 + tests/test_antigravity_harvest.py | 557 ++++++++++++++++++++++++ 5 files changed, 1171 insertions(+), 3 deletions(-) create mode 100644 skillopt_sleep/harvest_antigravity.py create mode 100644 tests/test_antigravity_harvest.py diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py index d6fac84f..f1f292ba 100644 --- a/skillopt_sleep/__main__.py +++ b/skillopt_sleep/__main__.py @@ -14,7 +14,7 @@ --target-skill-path PATH explicit live SKILL.md to stage/adopt --tasks-file PATH reviewed TaskRecord JSON file to replay instead of harvesting --backend mock|claude|codex|copilot|cursor|handoff - --source claude|codex|copilot|cursor|auto + --source claude|codex|copilot|cursor|antigravity|auto --vscode-workspace-storage PATH --model NAME --lookback-hours N @@ -80,10 +80,14 @@ def _add_common(p: argparse.ArgumentParser) -> None: p.add_argument("--claude-home", default="", help="override ~/.claude (also isolates state)") p.add_argument("--codex-home", default="", help="override ~/.codex for archived session harvest") p.add_argument("--cursor-home", default="", help="override ~/.cursor for Cursor session harvest") - p.add_argument("--source", default="", choices=["", "claude", "codex", "copilot", "cursor", "auto"], + p.add_argument("--source", default="", + choices=["", "claude", "codex", "copilot", "cursor", + "antigravity", "auto"], help="session transcript source") p.add_argument("--vscode-workspace-storage", default="", help="override VS Code User/workspaceStorage root for copilot source") + p.add_argument("--antigravity-home", default="", + help="override ~/.gemini/antigravity for antigravity source") p.add_argument("--lookback-hours", type=int, default=None, help="harvest window in hours; 0 = scan full history") p.add_argument("--edit-budget", type=int, default=0) @@ -124,6 +128,10 @@ def _cfg_from_args(args, task_meta: Dict[str, Any] | None = None) -> Any: overrides["codex_home"] = os.path.abspath(args.codex_home) if getattr(args, "cursor_home", ""): overrides["cursor_home"] = os.path.abspath(os.path.expanduser(args.cursor_home)) + if getattr(args, "antigravity_home", ""): + overrides["antigravity_home"] = os.path.abspath( + os.path.expanduser(args.antigravity_home) + ) if getattr(args, "source", ""): overrides["transcript_source"] = args.source if getattr(args, "vscode_workspace_storage", ""): diff --git a/skillopt_sleep/config.py b/skillopt_sleep/config.py index 3c105673..a091605f 100644 --- a/skillopt_sleep/config.py +++ b/skillopt_sleep/config.py @@ -20,6 +20,7 @@ CLAUDE_HOME = os.path.expanduser("~/.claude") CODEX_HOME = os.path.expanduser("~/.codex") CURSOR_HOME = os.path.expanduser("~/.cursor") +ANTIGRAVITY_HOME = os.path.expanduser("~/.gemini/antigravity") DEFAULTS: Dict[str, Any] = { @@ -27,8 +28,10 @@ "claude_home": CLAUDE_HOME, "codex_home": CODEX_HOME, "cursor_home": CURSOR_HOME, + "antigravity_home": ANTIGRAVITY_HOME, "vscode_workspace_storage": "", # "" => auto-detect platform defaults - "transcript_source": "claude", # "claude" | "codex" | "copilot" | "cursor" | "auto" + # "claude" | "codex" | "copilot" | "cursor" | "antigravity" | "auto" + "transcript_source": "claude", "projects": "invoked", # "invoked" | "all" | [list of abs paths] "invoked_project": "", # filled at runtime (cwd) when projects == "invoked" "lookback_hours": 72, # harvest window when no prior sleep recorded @@ -128,6 +131,11 @@ def cursor_projects_dir(self) -> str: cursor_home = os.path.abspath(os.path.expanduser(str(self.data["cursor_home"]))) return os.path.join(cursor_home, "projects") + @property + def antigravity_conversations_dir(self) -> str: + home = os.path.abspath(os.path.expanduser(str(self.data["antigravity_home"]))) + return os.path.join(home, "conversations") + @property def vscode_workspace_storage(self) -> str: value = self.data.get("vscode_workspace_storage", "") or "" diff --git a/skillopt_sleep/harvest_antigravity.py b/skillopt_sleep/harvest_antigravity.py new file mode 100644 index 00000000..9f89e973 --- /dev/null +++ b/skillopt_sleep/harvest_antigravity.py @@ -0,0 +1,586 @@ +"""Read Google Antigravity conversations and normalize them into digests. + +Antigravity persists each conversation as a SQLite "trajectory" database in +``~/.gemini/antigravity/conversations/.db``. The ``steps`` table holds +protobuf-encoded step payloads; without the proprietary schema we extract the +human-readable content with a conservative protobuf walker that collects +UTF-8 string fields: + + * step_type 14 -> user messages (the typed prompt, e.g. "/goal ...") + * step_type 5 -> artifact/answer content the agent produced + * step_type 33 -> tool calls (JSON with toolSummary/toolAction) + +Project provenance comes from the ``trajectory_metadata_blob`` row, which +records the workspace the conversation was opened against as a ``file://`` +URI (field 7, mirrored in field 1.1) plus the git branch (field 1.4). +Conversations started outside any workspace carry an explicit +``outside-of-project`` marker (field 18) instead. Sessions whose workspace +cannot be established are *never* relabelled as the invoked project: they are +only harvested under an explicit ``projects: "all"`` opt-in, and even then +keep an empty project rather than borrowing the caller's identity. + +Databases are read through SQLite's online backup API into a private +temporary directory. A plain file copy is not sufficient — Antigravity keeps +these databases in WAL mode with a live writer attached, so the ``.db`` file +on its own can be missing recent commits or the schema entirely. + +Heuristic by design: if Antigravity's schema changes, the walker degrades to +returning fewer strings — never to crashing the night (a session that yields +no user prompts is simply skipped, same as an empty transcript). +""" +from __future__ import annotations + +import json +import os +import re +import sqlite3 +import tempfile +import urllib.parse +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Tuple + +from skillopt_sleep.harvest import _detect_feedback, _is_meta_prompt +from skillopt_sleep.staging import redact_secrets +from skillopt_sleep.types import SessionDigest + +_USER_STEP_TYPES = {14} +_ARTIFACT_STEP_TYPES = {5} +_TOOL_STEP_TYPES = {33} + +# trajectory_metadata_blob protobuf fields (empirically stable across the +# observed corpus; every lookup degrades to "unknown" if they move). +_META_WORKSPACE_FIELD = 7 # "file:///..." workspace root +_META_REPO_FIELD = 1 # submessage: .1 workspace uri, .4 git branch +_META_REPO_URI_SUBFIELD = 1 +_META_REPO_BRANCH_SUBFIELD = 4 +_META_CREATED_FIELD = 2 # submessage: .1 epoch seconds +_META_CREATED_SECONDS_SUBFIELD = 1 +_META_PROJECT_MARKER_FIELD = 18 # "outside-of-project" when there is no workspace +_OUTSIDE_OF_PROJECT = "outside-of-project" + +_CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]") + +_UUID_RE = re.compile( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$") + +# Antigravity injects a system wrapper around /goal tasks, and user steps also +# carry a permission-history of tool echoes like ``read_url(github.com)`` — +# neither is the user's own words. +_BOILERPLATE_MARKERS = ( + "marked this task with /goal", + "The system will force you to continue", +) +_TOOL_ECHO_RE = re.compile(r"^[\w$.\\/-]+\([^()]*\)$") + +_READ_TIMEOUT = 5.0 +_MAX_STRINGS_PER_BLOB = 400 +_MAX_WALK_DEPTH = 6 +_MIN_STRING_LEN = 16 + + +# ── generic protobuf decoding ───────────────────────────────────────────────── + +def _read_varint(buf: bytes, i: int) -> Tuple[Optional[int], int]: + val = 0 + shift = 0 + n = len(buf) + while i < n: + b = buf[i] + i += 1 + val |= (b & 0x7F) << shift + shift += 7 + if not b & 0x80: + return val, i + if shift > 63: + break + return None, i + + +def _proto_fields(buf: bytes) -> Dict[int, Any]: + """Decode one protobuf level into {field_number: value}, first wins. + + Length-delimited fields yield ``bytes``; varints yield ``int``. Unknown + wire types stop the scan rather than raising — a truncated or reshaped + message simply contributes the fields decoded so far. + """ + out: Dict[int, Any] = {} + i, n = 0, len(buf) + while i < n: + tag, i = _read_varint(buf, i) + if tag is None: + break + field, wire = tag >> 3, tag & 7 + if wire == 0: + val, i = _read_varint(buf, i) + if val is None: + break + out.setdefault(field, val) + elif wire == 2: + ln, i = _read_varint(buf, i) + if ln is None or ln < 0 or i + ln > n: + break + out.setdefault(field, buf[i:i + ln]) + i += ln + elif wire == 1: + i += 8 + elif wire == 5: + i += 4 + else: + break + return out + + +def _proto_strings(buf: bytes, depth: int = 0, out: Optional[List[str]] = None) -> List[str]: + """Collect plausible UTF-8 string fields from a protobuf blob (schema-less).""" + if out is None: + out = [] + if depth > _MAX_WALK_DEPTH or len(out) > _MAX_STRINGS_PER_BLOB: + return out + i, n = 0, len(buf) + while i < n: + tag, i = _read_varint(buf, i) + if tag is None: + break + wire = tag & 7 + if wire == 0: + _v, i = _read_varint(buf, i) + if _v is None: + break + elif wire == 1: + i += 8 + elif wire == 5: + i += 4 + elif wire == 2: + ln, i = _read_varint(buf, i) + if ln is None or ln < 0 or i + ln > n: + break + chunk = buf[i:i + ln] + i += ln + try: + text: Optional[str] = chunk.decode("utf-8") + except UnicodeDecodeError: + text = None + if text is not None and len(text) >= _MIN_STRING_LEN and _looks_natural(text): + out.append(text) + else: + # possibly a nested message — recurse; a failed walk just + # contributes nothing + _proto_strings(chunk, depth + 1, out) + else: # unknown/deprecated wire types: bail out of this blob + break + return out + + +def _looks_natural(text: str) -> bool: + """Keep human/markdown text; drop ids, uuids, base64 runs, file URIs.""" + t = text.strip() + if not t or _UUID_RE.match(t): + return False + # A nested message often decodes as valid UTF-8, which would otherwise + # yield the inner text with its protobuf framing bytes glued to the front + # ("\x12\x30Refactor the ..."). Real prose carries no control bytes, so + # rejecting them here sends the chunk back through the walker instead. + if _CONTROL_CHARS_RE.search(t): + return False + if t.startswith(("file:///", "http://", "https://")) and " " not in t: + return False + if " " not in t and len(t) > 40: # long spaceless token: id/base64 + return False + letters = sum(c.isalpha() or c.isspace() for c in t) + return letters / max(1, len(t)) > 0.55 + + +# ── path / provenance helpers ───────────────────────────────────────────────── + +def _path_from_file_uri(uri: str) -> str: + """Convert ``file:///C:/a%20b`` to a native path, cross-platform. + + ``urllib.request.url2pathname`` is platform-dependent, which would make the + same database resolve differently on POSIX and Windows; the recorded URI is + always absolute, so decode it directly instead. + """ + if not uri.startswith("file://"): + return "" + rest = uri[len("file://"):] + if not rest.startswith("/"): # file://host/path — host form is not supported + slash = rest.find("/") + if slash < 0: + return "" + rest = rest[slash:] + path = urllib.parse.unquote(rest) + if re.match(r"^/[A-Za-z]:([/\\]|$)", path): # /C:/... -> C:/... + path = path[1:] + return os.path.normpath(path) if path else "" + + +def _normalized_path(path: str) -> str: + return os.path.normcase(os.path.normpath(os.path.expanduser(path))) + + +def _is_workspace_ancestor(workspace: str, invoked: str) -> bool: + """True when ``invoked`` is ``workspace`` or lives inside it.""" + if not workspace or not invoked: + return False + try: + workspace_norm = _normalized_path(workspace) + invoked_norm = _normalized_path(invoked) + return os.path.commonpath([workspace_norm, invoked_norm]) == workspace_norm + except (OSError, ValueError): # different drives / unrelated roots + return False + + +def _iso_utc(epoch: float) -> str: + return ( + datetime.fromtimestamp(epoch, tz=timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def _mtime(path: str) -> Optional[float]: + try: + return os.path.getmtime(path) + except OSError: + return None + + +def _iso_epoch(value: Optional[str]) -> Optional[float]: + if not value: + return None + try: + normalized = value[:-1] + "+00:00" if value.endswith("Z") else value + return datetime.fromisoformat(normalized).timestamp() + except (TypeError, ValueError, OSError): + return None + + +# ── snapshotting ────────────────────────────────────────────────────────────── + +def _query(con: sqlite3.Connection) -> Dict[str, List[Any]]: + """Pull the rows we need out of an open conversation database.""" + rows: Dict[str, List[Any]] = { + "steps": con.execute( + "SELECT idx, step_type, step_payload FROM steps ORDER BY idx" + ).fetchall(), + } + try: + rows["meta"] = con.execute( + "SELECT data FROM trajectory_metadata_blob" + ).fetchall() + except sqlite3.Error: + rows["meta"] = [] # older/reshaped stores: provenance stays unknown + return rows + + +def _read_direct(path: str) -> Dict[str, List[Any]]: + """Read through a read-only URI connection (raises on lock/corruption). + + SQLite attaches the ``-wal`` sidecar itself, so this read sees every + committed transaction. Copying the bare ``.db`` — the previous approach — + does not: with a live writer attached the copy can miss recent commits, or + the schema entirely. + """ + con = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=_READ_TIMEOUT) + try: + return _query(con) + finally: + try: + con.close() + except sqlite3.Error: + pass + + +def _read_via_backup(path: str, tmpdir: str) -> Dict[str, List[Any]]: + """Read through SQLite's online backup API into a private temp directory. + + The fallback for databases that cannot be opened directly (an exclusive + lock, or a read-only volume with no usable ``-shm``). The snapshot lands in + the caller's ``TemporaryDirectory`` — never a predictable path — and is + removed as soon as the rows are read. + """ + snapshot = os.path.join(tmpdir, "snapshot.db") + source = destination = None + try: + source = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=_READ_TIMEOUT) + destination = sqlite3.connect(snapshot) + source.backup(destination) + return _query(destination) + finally: + for con in (destination, source): + if con is not None: + try: + con.close() + except sqlite3.Error: + pass + try: + if os.path.exists(snapshot): + os.unlink(snapshot) + except OSError: + pass + + +def _read_rows(path: str, tmpdir: str) -> Optional[Dict[str, List[Any]]]: + """Read one conversation database consistently, without copying the file. + + Returns None when the database is locked beyond the timeout, corrupt, or + not a database at all — a bad store skips its session instead of failing + the night. + """ + try: + return _read_direct(path) + except sqlite3.Error: + pass + try: + return _read_via_backup(path, tmpdir) + except sqlite3.Error: + return None + + +def _provenance(meta_rows: List[Any]) -> Tuple[str, str, Optional[float]]: + """Extract (workspace_path, git_branch, created_epoch) from the metadata row. + + ``workspace_path`` is "" both for conversations Antigravity marked + ``outside-of-project`` and for stores whose provenance cannot be read; the + caller treats the two identically (never attributable to a project). + """ + for row in meta_rows or []: + blob = row[0] if isinstance(row, (tuple, list)) else row + if not isinstance(blob, (bytes, bytearray)): + continue + fields = _proto_fields(bytes(blob)) + + marker = fields.get(_META_PROJECT_MARKER_FIELD) + if isinstance(marker, bytes) and marker == _OUTSIDE_OF_PROJECT.encode(): + return "", "", _created_epoch(fields) + + uri = fields.get(_META_WORKSPACE_FIELD) + branch = "" + repo = fields.get(_META_REPO_FIELD) + if isinstance(repo, bytes): + inner = _proto_fields(repo) + if not isinstance(uri, bytes): + uri = inner.get(_META_REPO_URI_SUBFIELD) + raw_branch = inner.get(_META_REPO_BRANCH_SUBFIELD) + if isinstance(raw_branch, bytes): + try: + branch = raw_branch.decode("utf-8") + except UnicodeDecodeError: + branch = "" + + workspace = "" + if isinstance(uri, bytes): + try: + workspace = _path_from_file_uri(uri.decode("utf-8")) + except UnicodeDecodeError: + workspace = "" + + return workspace, branch, _created_epoch(fields) + return "", "", None + + +def _created_epoch(fields: Dict[int, Any]) -> Optional[float]: + """Conversation start time, when the metadata carries a sane timestamp.""" + stamp = fields.get(_META_CREATED_FIELD) + if isinstance(stamp, bytes): + seconds = _proto_fields(stamp).get(_META_CREATED_SECONDS_SUBFIELD) + if isinstance(seconds, int) and 0 < seconds < 4_102_444_800: # < year 2100 + return float(seconds) + return None + + +# ── per-database digestion ──────────────────────────────────────────────────── + +def _clean_user_prompt(text: str) -> str: + t = text.strip() + for prefix in ("/goal ", "/task ", "/ask "): + if t.lower().startswith(prefix): + t = t[len(prefix):] + return t.strip() + + +def _sanitize(text: str) -> str: + """Redact secrets and strip NULs before the text leaves the harvester. + + Every prompt, final and tool name passes through here, so nothing reaches + the evidence log or a model-processing path un-redacted. + """ + return str(redact_secrets(text)).replace("\x00", "").strip() + + +def _digest_with_provenance( + path: str, tmpdir: str +) -> Optional[Tuple[SessionDigest, str]]: + """Digest one database and report the workspace Antigravity recorded. + + Returns ``(digest, workspace)`` where ``workspace`` is "" when the + conversation was started outside any project or its provenance could not be + read. The database is snapshotted exactly once. + """ + rows = _read_rows(path, tmpdir) + if rows is None: + return None + + workspace, branch, created = _provenance(rows.get("meta", [])) + + prompts: List[str] = [] + finals: List[str] = [] + tools: List[str] = [] + for _idx, stype, payload in rows.get("steps", []): + if isinstance(payload, (bytes, bytearray)): + blob = bytes(payload) + elif payload is None: + continue + else: + blob = str(payload).encode("utf-8", "replace") + if not blob: + continue + if stype in _USER_STEP_TYPES: + strs = [ + s for s in _proto_strings(blob) + if not s.startswith("{") + and not any(m in s for m in _BOILERPLATE_MARKERS) + and not _TOOL_ECHO_RE.match(s.strip()) + ] + if strs: + p = _sanitize(_clean_user_prompt(max(strs, key=len))) + if p and not _is_meta_prompt(p): + prompts.append(p) + elif stype in _ARTIFACT_STEP_TYPES: + strs = _proto_strings(blob) + # prefer the artifact body over its ArtifactMetadata JSON envelope + body = [s for s in strs if not s.lstrip().startswith("{")] + if body or strs: + final = _sanitize(max(body or strs, key=len)) + if final: + finals.append(final) + elif stype in _TOOL_STEP_TYPES: + for s in _proto_strings(blob): + if not s.startswith("{"): + continue + try: + obj = json.loads(s) + except ValueError: + continue + if not isinstance(obj, dict): + continue + name = obj.get("toolSummary") or obj.get("toolAction") + if not name: + continue + clean = re.sub(r"[^A-Za-z0-9_.:-]+", "_", _sanitize(str(name)))[:80] + if clean and clean not in tools: + tools.append(clean) + + if not prompts: + return None + + modified = _mtime(path) + ended = _iso_utc(modified) if modified is not None else "" + started = _iso_utc(created) if created is not None else ended + digest = SessionDigest( + session_id=os.path.splitext(os.path.basename(path))[0], + project="", + git_branch=branch, + started_at=started, + ended_at=ended, + user_prompts=prompts, + assistant_finals=finals[-3:], + tools_used=tools[:12], + feedback_signals=_detect_feedback(" \n".join(prompts)), + n_user_turns=len(prompts), + n_assistant_turns=len(finals), + raw_path=path, + ) + return digest, workspace + + +def digest_antigravity_db( + path: str, *, project: str = "", tmpdir: Optional[str] = None +) -> Optional[SessionDigest]: + """Digest one conversation database, or None if it yields no user prompts. + + ``project`` overrides the label; when omitted the digest carries the + workspace Antigravity recorded for the conversation. + """ + if tmpdir is None: + with tempfile.TemporaryDirectory(prefix="skillopt-agy-") as owned: + return digest_antigravity_db(path, project=project, tmpdir=owned) + result = _digest_with_provenance(path, tmpdir) + if result is None: + return None + digest, workspace = result + digest.project = project or workspace + return digest + + +# ── scope selection ─────────────────────────────────────────────────────────── + +def _selected_projects(scope: Any, invoked_project: str) -> List[str]: + if isinstance(scope, (list, tuple)): + return [str(p) for p in scope if str(p).strip()] + return [invoked_project] if invoked_project else [] + + +def harvest_antigravity( + conversations_dir: str, + *, + scope: Any = "invoked", + invoked_project: str = "", + since_iso: Optional[str] = None, + limit: int = 0, +) -> List[SessionDigest]: + """Return Antigravity session digests for the selected workspace scope. + + ``scope="all"`` harvests every conversation, labelling each with the + workspace Antigravity recorded (empty when the conversation was started + outside a workspace). Any other scope keeps only conversations whose + recorded workspace contains one of the selected projects; conversations + with unknown or out-of-project provenance are skipped rather than being + attributed to the invoked project. + """ + if not os.path.isdir(conversations_dir): + return [] + + try: + names = sorted(os.listdir(conversations_dir)) + except OSError: + return [] + + candidates: List[Tuple[str, float]] = [] + for name in names: + if not name.endswith(".db"): + continue + path = os.path.join(conversations_dir, name) + if not os.path.isfile(path): + continue + modified = _mtime(path) + if modified is not None: + candidates.append((path, modified)) + candidates.sort(key=lambda item: (-item[1], item[0])) + + since_epoch = _iso_epoch(since_iso) + wanted = _selected_projects(scope, invoked_project) + harvest_all = scope == "all" + if not harvest_all and not wanted: + return [] # nothing to scope against; never fall back to "everything" + + digests: List[SessionDigest] = [] + with tempfile.TemporaryDirectory(prefix="skillopt-agy-") as tmpdir: + for path, modified in candidates: + if since_epoch is not None and modified <= since_epoch: + continue + result = _digest_with_provenance(path, tmpdir) + if result is None: + continue + digest, workspace = result + if not harvest_all and not any( + _is_workspace_ancestor(workspace, project) for project in wanted + ): + # Unknown, outside-of-project, or a different workspace: skip it + # rather than relabelling it as the invoked project. + continue + digest.project = workspace + digests.append(digest) + if limit and len(digests) >= limit: + break + return digests diff --git a/skillopt_sleep/harvest_sources.py b/skillopt_sleep/harvest_sources.py index a43ebaca..bd8f7127 100644 --- a/skillopt_sleep/harvest_sources.py +++ b/skillopt_sleep/harvest_sources.py @@ -4,6 +4,7 @@ from typing import Optional from skillopt_sleep.harvest import harvest +from skillopt_sleep.harvest_antigravity import harvest_antigravity from skillopt_sleep.harvest_copilot import harvest_copilot from skillopt_sleep.harvest_codex import harvest_codex from skillopt_sleep.harvest_cursor import harvest_cursor @@ -15,6 +16,14 @@ def harvest_for_config(cfg, *, since_iso: Optional[str] = None, limit: int = 0) scope = cfg.get("projects", "invoked") invoked_project = cfg.get("invoked_project", "") + if source == "antigravity": + return harvest_antigravity( + cfg.antigravity_conversations_dir, + scope=scope, + invoked_project=invoked_project, + since_iso=since_iso, + limit=limit, + ) if source == "codex": return harvest_codex( cfg.codex_archived_sessions_dir, diff --git a/tests/test_antigravity_harvest.py b/tests/test_antigravity_harvest.py new file mode 100644 index 00000000..82671af7 --- /dev/null +++ b/tests/test_antigravity_harvest.py @@ -0,0 +1,557 @@ +"""Tests for read-only Google Antigravity conversation harvesting. + +Fixtures are built rather than checked in: Antigravity's trajectory stores are +protobuf blobs inside SQLite, so a synthetic encoder keeps the expected shape +visible in the test itself and lets each case vary one thing (a step type, a +workspace, a WAL state, a secret) without hand-editing binary files. +""" +from __future__ import annotations + +import os +import sqlite3 +import struct +import tempfile +import time +import unittest +from unittest import mock + +from skillopt_sleep.config import load_config +from skillopt_sleep.harvest_antigravity import ( + _path_from_file_uri, + _read_direct, + _read_rows, + _read_via_backup, + digest_antigravity_db, + harvest_antigravity, +) +from skillopt_sleep.harvest_sources import harvest_for_config + + +# ── minimal protobuf encoder (mirrors what Antigravity writes) ──────────────── + +def _varint(value: int) -> bytes: + out = bytearray() + while True: + byte = value & 0x7F + value >>= 7 + out.append(byte | (0x80 if value else 0)) + if not value: + return bytes(out) + + +def _tag(field: int, wire: int) -> bytes: + return _varint((field << 3) | wire) + + +def _bytes_field(field: int, payload: bytes) -> bytes: + return _tag(field, 2) + _varint(len(payload)) + payload + + +def _str_field(field: int, text: str) -> bytes: + return _bytes_field(field, text.encode("utf-8")) + + +def _varint_field(field: int, value: int) -> bytes: + return _tag(field, 0) + _varint(value) + + +def _path_to_uri(path: str) -> str: + from urllib.parse import quote + normalized = path.replace("\\", "/") + if not normalized.startswith("/"): + normalized = "/" + normalized # drive-letter form: /C:/Users/... + return "file://" + quote(normalized, safe="/:") + + +def metadata_blob( + *, + workspace: str = "", + branch: str = "", + created: int = 1_785_483_868, + session_id: str = "11111111-2222-3333-4444-555555555555", +) -> bytes: + """Build a ``trajectory_metadata_blob`` payload. + + With ``workspace`` empty this produces the ``outside-of-project`` form + Antigravity writes for conversations started without a workspace. + """ + blob = b"" + if workspace: + uri = _path_to_uri(workspace) + repo = _str_field(1, uri) + _str_field(2, uri) + if branch: + repo += _str_field(4, branch) + blob += _bytes_field(1, repo) + blob += _bytes_field(2, _varint_field(1, created) + _varint_field(2, 0)) + blob += _str_field(3, "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee") + blob += _str_field(6, session_id) + if workspace: + blob += _str_field(7, _path_to_uri(workspace)) + blob += _str_field(18, "351053ef-3b69-4a5b-b9e5-a16e705ba970") + else: + blob += _str_field(18, "outside-of-project") + return blob + + +def step_payload(text: str) -> bytes: + """A step payload with the text nested one message deep, as observed.""" + return _bytes_field(3, _bytes_field(1, _str_field(2, text))) + + +class AntigravityFixture: + """Builds a conversations directory of synthetic trajectory databases.""" + + USER, ARTIFACT, TOOL = 14, 5, 33 + + def __init__(self, root: str): + self.root = root + os.makedirs(root, exist_ok=True) + + def write( + self, + session_id: str, + steps, + *, + workspace: str = "", + branch: str = "", + created: int = 1_785_483_868, + journal_mode: str = "delete", + keep_open: bool = False, + with_metadata: bool = True, + mtime: float = 0.0, + ): + path = os.path.join(self.root, f"{session_id}.db") + con = sqlite3.connect(path) + con.execute(f"PRAGMA journal_mode={journal_mode}") + con.execute( + "CREATE TABLE steps (idx integer, step_type integer NOT NULL DEFAULT 0," + " step_payload blob)" + ) + con.execute( + "CREATE TABLE trajectory_metadata_blob (id text DEFAULT 'main'," + " data blob, PRIMARY KEY (id))" + ) + for idx, (step_type, payload) in enumerate(steps): + con.execute( + "INSERT INTO steps VALUES (?, ?, ?)", + (idx, step_type, payload if isinstance(payload, bytes) else step_payload(payload)), + ) + if with_metadata: + con.execute( + "INSERT INTO trajectory_metadata_blob VALUES (?, ?)", + ("main", metadata_blob(workspace=workspace, branch=branch, + created=created, session_id=session_id)), + ) + con.commit() + if keep_open: + # Leave the writer attached, exactly like a live Antigravity editor: + # in WAL mode the committed rows still live in the -wal sidecar. + self._open = getattr(self, "_open", []) + self._open.append(con) + else: + con.close() + if mtime: + os.utime(path, (mtime, mtime)) + return path + + def close(self): + for con in getattr(self, "_open", []): + try: + con.close() + except sqlite3.Error: + pass + + +# ── tests ───────────────────────────────────────────────────────────────────── + +class TestAntigravityStepExtraction(unittest.TestCase): + def test_extracts_user_artifact_and_tool_step_types(self): + with tempfile.TemporaryDirectory() as tmp: + fx = AntigravityFixture(os.path.join(tmp, "conversations")) + path = fx.write("s1", [ + (fx.USER, "/goal Refactor the retry helper so it backs off."), + (fx.ARTIFACT, "I refactored the helper to use exponential backoff."), + (fx.TOOL, step_payload('{"toolSummary": "run_command", "x": 1}')), + (fx.USER, "That is still broken on the third attempt, please fix it."), + ], workspace=os.path.join(tmp, "proj")) + + digest = digest_antigravity_db(path) + + self.assertIsNotNone(digest) + self.assertEqual(digest.session_id, "s1") + # step 14 -> prompts, with the /goal wrapper stripped + self.assertEqual(digest.user_prompts, [ + "Refactor the retry helper so it backs off.", + "That is still broken on the third attempt, please fix it.", + ]) + # step 5 -> assistant finals + self.assertEqual(digest.assistant_finals, + ["I refactored the helper to use exponential backoff."]) + # step 33 -> tool names + self.assertEqual(digest.tools_used, ["run_command"]) + self.assertEqual(digest.n_user_turns, 2) + self.assertTrue(digest.feedback_signals) + + def test_preserves_unicode_prompts(self): + prompt = "Peux-tu corriger l'accentuation ? 変換もお願いします — emoji 🎯 too." + with tempfile.TemporaryDirectory() as tmp: + fx = AntigravityFixture(os.path.join(tmp, "conversations")) + path = fx.write("s1", [(fx.USER, prompt)], workspace=os.path.join(tmp, "p")) + digest = digest_antigravity_db(path) + + self.assertIsNotNone(digest) + self.assertEqual(digest.user_prompts, [prompt]) + + def test_tolerates_schema_drift_and_unknown_wire_types(self): + """Reshaped payloads degrade to fewer strings, never to an exception.""" + with tempfile.TemporaryDirectory() as tmp: + fx = AntigravityFixture(os.path.join(tmp, "conversations")) + drifted = ( + _varint_field(99, 7) # unknown scalar field + + _bytes_field(41, b"\x00\x01\x02\xff\xfe") # non-UTF8 blob + + _tag(12, 1) + struct.pack("