Paste this into Claude Code, Codex, or another local agent with filesystem and network access:
Perform a read-only audit of Grok Build codebase collection and retention on
this machine. Do the investigation with shell/Python; do not just give me
instructions. Do not launch Grok, send it a prompt, upload/download archives,
invoke /privacy, delete anything, or change configuration/account state. Never
print tokens, session IDs, or storage object IDs. Do not create or delete even
temporary files; process everything in memory or through stdin/stdout.
Inspect ~/.grok/logs/unified*.jsonl plus rotated logs beside it. Find
`repo_state.upload.start` and `repo_state.upload.enqueued`. Correlate events by
`sid`, `ctx.turn_number`, and phase; derive an enqueued event's phase from its
`gcs_path` basename (`before_codebase.tar.gz` or `after_codebase.tar.gz`). Map
each archive to the start event's `ctx.repo_path`.
For every repository, count collection attempts and queued archives and show
first/last timestamps. Then perform a metadata-only existence check: read the
newest cached `key` from ~/.grok/auth.json without displaying it and POST the
exact queued paths as `{"paths":[...]}` batches to:
https://cli-chat-proxy.grok.com/v1/storage/batch_exists
with `Authorization: Bearer <token>`. Never call download or follow signed URLs.
Print the UTC check time and this table:
STATUS | PRESENT/ABSENT/UNKNOWN | ATTEMPTS | FIRST/LAST | REPOSITORY
Classify COLLECTED-ONLY when collection started without a matching queue event;
QUEUED when an archive was enqueued; REMOTE-PRESENT when any exact ID is in
`exists`; NOT-LISTED only when all its IDs are in `missing`; otherwise UNKNOWN.
Finish with totals and caveats: queueing alone does not prove server retention;
REMOTE-PRESENT proves only that this API still recognizes the object;
NOT-LISTED does not prove deletion from backups, replicas, derived data, or
other stores; missing logs can cause false negatives. Warn that repository
paths are sensitive before I share the output.
| #!/usr/bin/env python3 | |
| """Find repositories collected by Grok Build. | |
| Usage: | |
| python3 grok-upload-audit.py # read local Grok logs only | |
| python3 grok-upload-audit.py --remote # also ask xAI if archive IDs exist | |
| The remote check sends only logged archive IDs to xAI. It never uploads or | |
| downloads source code and never prints credentials or archive IDs. | |
| Labels: | |
| COLLECTED-ONLY Grok's collector started; no matching queue event was found. | |
| QUEUED An archive reached Grok's persistent upload queue. | |
| REMOTE-PRESENT xAI's API still recognizes at least one exact archive ID. | |
| NOT-LISTED xAI reports the IDs absent. This does not prove deletion from | |
| backups, replicas, derived data, or other storage systems. | |
| """ | |
| import json | |
| import os | |
| import sys | |
| import urllib.error | |
| import urllib.request | |
| from collections import defaultdict | |
| from pathlib import Path | |
| REMOTE = "--remote" in sys.argv | |
| LOG = Path(os.getenv("GROK_LOG", "~/.grok/logs/unified.jsonl")).expanduser() | |
| AUTH = Path(os.getenv("GROK_AUTH", "~/.grok/auth.json")).expanduser() | |
| API = "https://cli-chat-proxy.grok.com/v1/storage/batch_exists" | |
| if "--help" in sys.argv or "-h" in sys.argv: | |
| raise SystemExit(doc) | |
| if not LOG.is_file(): | |
| raise SystemExit(f"Grok log not found: {LOG}") | |
| starts = {} | |
| repos = defaultdict(lambda: {"attempts": 0, "archives": [], "last": ""}) | |
| for line in LOG.open(encoding="utf-8", errors="replace"): | |
| try: | |
| event = json.loads(line) | |
| except json.JSONDecodeError: | |
| continue | |
| ctx = event.get("ctx") or {} | |
| msg, sid = event.get("msg"), event.get("sid") | |
| turn, timestamp = ctx.get("turn_number"), event.get("ts") or "" | |
| if msg == "repo_state.upload.start" and isinstance(ctx.get("repo_path"), str): | |
| repo = ctx["repo_path"] | |
| starts[sid, turn, ctx.get("phase")] = repo | |
| repos[repo]["attempts"] += 1 | |
| repos[repo]["last"] = max(repos[repo]["last"], timestamp) | |
| elif msg == "repo_state.upload.enqueued" and isinstance(ctx.get("gcs_path"), str): | |
| archive = ctx["gcs_path"] | |
| name = archive.rsplit("/", 1)[-1] | |
| if name not in {"before_codebase.tar.gz", "after_codebase.tar.gz"}: | |
| continue | |
| phase = name.removesuffix(".tar.gz") | |
| repo = starts.get((sid, turn, phase), "<unknown: earlier log missing>") | |
| repos[repo]["archives"].append(archive) | |
| repos[repo]["last"] = max(repos[repo]["last"], timestamp) | |
| paths = list(dict.fromkeys(p for row in repos.values() for p in row["archives"])) | |
| present, absent = set(), set() | |
| if REMOTE and paths: | |
| auth = json.loads(AUTH.read_text(encoding="utf-8")) | |
| entries = [v for v in auth.values() if isinstance(v, dict) and v.get("key")] | |
| if not entries: | |
| raise SystemExit("No cached token found; run grok login first.") | |
| token = max(entries, key=lambda v: v.get("expires_at", ""))["key"] | |
| for offset in range(0, len(paths), 100): | |
| body = json.dumps({"paths": paths[offset : offset + 100]}).encode() | |
| request = urllib.request.Request( | |
| API, | |
| data=body, | |
| method="POST", | |
| headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, | |
| ) | |
| try: | |
| with urllib.request.urlopen(request, timeout=20) as response: | |
| result = json.load(response) | |
| except urllib.error.HTTPError as error: | |
| raise SystemExit(f"Storage check failed with HTTP {error.code}.") from None | |
| except urllib.error.URLError as error: | |
| raise SystemExit(f"Storage check failed: {error.reason}") from None | |
| present.update(result.get("exists", [])) | |
| absent.update(result.get("missing", [])) | |
| home = str(Path.home()) | |
| print("STATUS\tARCHIVES\tATTEMPTS\tLAST EVENT\tREPOSITORY") | |
| for repo, row in sorted(repos.items()): | |
| archives = row["archives"] | |
| if not archives: | |
| status, summary = "COLLECTED-ONLY", "0" | |
| elif not REMOTE: | |
| status, summary = "QUEUED", str(len(archives)) | |
| else: | |
| yes = sum(p in present for p in archives) | |
| no = sum(p in absent for p in archives) | |
| unknown = len(archives) - yes - no | |
| status = "REMOTE-PRESENT" if yes else "NOT-LISTED" if no == len(archives) else "UNKNOWN" | |
| summary = f"{yes} present/{no} absent/{unknown} unknown" | |
| shown = "~" + repo[len(home) :] if repo.startswith(home + "/") else repo | |
| print(f"{status}\t{summary}\t{row['attempts']}\t{row['last'] or '-'}\t{shown}") | |
| print("\nReview output before sharing: repository paths may be private.") | |
| print("NOT-LISTED is not proof of deletion from backups, replicas, or derived data.") |