DEV Community

Lily
Lily

Posted on Originally published at dev.to

6 Pitfalls Syncing AI Memory to Google Drive With No API: Keeping File IDs Intact

#claudecode #automation #security #python

Use the obvious "write a temp file, then os.replace()" pattern on a Google Drive for Desktop mount and the file keeps its name but loses its Drive identity. Drive sees the original file disappear and a brand-new one show up in its place, with a new file id. I switched to overwriting the same file descriptor in place. The file id now survives every push, and the sync path still needs no gws auth and no network calls.

Last time I wrote about isolating untrusted external output. This post covers the sync path behind it: memory-drive-mount-sync.py, which treats the local Google Drive for Desktop mount folder as nothing more than a place to drop files.

The problem: add a sync path without adding another API

I already have one path that puts vault memory in the cloud: memory-cloud-sync.py, built on gws (Google Workspace CLI). It's thorough: OAuth, the Drive API, Docs API batchUpdate, and CAS-style conflict detection via requiredRevisionId. The cost of that thoroughness is a heavy set of dependencies.

My Mac also already runs Google Drive for Desktop, which mounts Drive contents as local files under ~/Library/CloudStorage/GoogleDrive-<google-account>/マイドライブ/ (My Drive). Writing there should sync without any extra auth or API calls.

But the naive "write to a temp file and rename with os.replace()" approach breaks things. os.replace() swaps the inode, and Drive for Desktop ties each Drive file id to the local file object (the inode). Swap that object out with a rename and Drive sees the original file vanish and a new one appear. The file id the file carried is lost.

Design: push overwrites the same file, pull reads only a narrow set of notes

The module docstring of memory-drive-mount-sync.py lays out exactly what the script is allowed to do:

"""One bounded, single-run local sync against a Google Drive for Desktop mount.

This is deliberately separate from memory-cloud-sync.py: it never calls ``gws``,
never touches network credentials, and does nothing beyond one push (of the
locally built curated snapshot) and one bounded pull (of a handful of plain
dated notes already sitting under the mounted folder). It only ever reads and
writes: the vault's own AI/.runtime build cache, a single named file inside
the mount directory (in place), and AI/INBOX/cloud inside the vault.
"""
Enter fullscreen mode Exit fullscreen mode

It has exactly three jobs:

  • push: take the SHARED-MEMORY.txt built from the vault by memory-bridge.py build and overwrite the file of the same name in the mount, as the same file
  • pull: pick up only date-named plain-text notes in the mount (.txt/.md/.markdown, max 50 files, 256KB per file) and stage them in quarantine under the vault's AI/INBOX/cloud
  • block concurrent runs with a single-run lock

Note: With no gws and no Drive API, there's no competing-write detection like requiredRevisionId. Instead, TOCTOU (check-then-use) gaps are closed with POSIX fstat/inode matching, O_NOFOLLOW, and flock. With no auth layer, the safety comes from being meticulous at the filesystem layer.

Push: overwriting without breaking the file id

This is the core function. Instead of os.replace(), it does seek(0) → write → truncate on the existing file descriptor.

def overwrite_shared_memory(path: Path, content: bytes) -> None:
    """Overwrite an existing regular file in place, preserving its identity.

    Deliberately never os.replace(): the mount file's identity (e.g. a Google
    Drive file id behind the local inode) must be preserved rather than
    swapped for a new one.
    """
    initial = m.check_regular_non_symlink(path)
    flags = os.O_RDWR
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    fd = os.open(path, flags)
    try:
        opened = os.fstat(fd)
        if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != (initial.st_dev, initial.st_ino):
            raise MountSyncError("mount_target_changed_during_write")
        handle = os.fdopen(fd, "r+b", closefd=False)
        try:
            handle.seek(0)
            handle.write(content)
            handle.truncate()
            handle.flush()
            os.fsync(handle.fileno())
        finally:
            handle.close()
    finally:
        os.close(fd)
Enter fullscreen mode Exit fullscreen mode

check_regular_non_symlink first runs an lstat-based safety check (not a symlink, is a regular file). Then the file is open()ed, and the (st_dev, st_ino) from an immediate fstat is compared with the lstat result taken before the open. If they differ, the file was swapped between the check and the open (TOCTOU), and the write is aborted.

The push itself also has a guard before writing:

mount_path = mount_dir / "SHARED-MEMORY.txt"
# Missing / symlink / non-regular mount target: refuse to write, surface a
# non-sensitive error code, and never fall back to creating a new file
# (that would mint a new Drive file id in place of the shared one).
m.check_regular_non_symlink(mount_path)
Enter fullscreen mode Exit fullscreen mode

If the target is missing, is a symlink, or isn't a regular file, the script does not fall back to creating a new file. Creating one would make Drive issue a different file id. Content is compared with the updated: line stripped out, and if nothing else changed, the write is skipped entirely. That way a timestamp-only difference doesn't set off a Drive sync event on every run.

Pull: only date-named plain-text notes

The pull targets are explicitly narrowed:

MAX_PULL_FILES = 50
MAX_PULL_FILE_BYTES = 256 * 1024
NAME_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}")
PULL_EXTENSIONS = {".txt": "text/plain", ".md": "text/markdown", ".markdown": "text/markdown"}

def classify_pull_candidate(name: str) -> Optional[str]:
    if not NAME_DATE_RE.match(name):
        return None
    if name.upper().startswith("SHARED-MEMORY"):
        return None
    lower = name.lower()
    for suffix, mime in PULL_EXTENSIONS.items():
        if lower.endswith(suffix):
            return mime
    return None
Enter fullscreen mode Exit fullscreen mode

Only names starting with YYYY-MM-DD are picked up, and the push target itself (anything starting with SHARED-MEMORY) is explicitly excluded. Without that exclusion, the next pull would pick up the file you just pushed as an "unknown external note," creating a loop.

for entry in entries:
    info = entry.stat(follow_symlinks=False)
    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
        counts["skipped"] += 1
        continue
    ...
    candidates.append((entry, mime))
    if len(candidates) >= MAX_PULL_FILES:
        break
Enter fullscreen mode Exit fullscreen mode

Symlinks and non-regular files are rejected here, and the candidate list stops at 50. Even with thousands of files in the folder, scandir keeps the scan cheap, and at most 50 files are actually read.

Reusing secret screening and symlink safety checks

Each note picked up by pull goes straight into m.stage_note(), which already exists in memory-cloud-sync.py. The hyphen in the filename rules out import memory-cloud-sync, so the module is loaded by path with importlib.

_CLOUD_SYNC_PATH = Path(__file__).with_name("memory-cloud-sync.py")
_spec = importlib.util.spec_from_file_location("memory_cloud_sync_for_mount_sync", _CLOUD_SYNC_PATH)
if _spec is None or _spec.loader is None:  # pragma: no cover - defensive only
    raise ImportError("memory-cloud-sync.py could not be loaded")
m = importlib.util.module_from_spec(_spec)
sys.modules[_spec.name] = m
_spec.loader.exec_module(m)
Enter fullscreen mode Exit fullscreen mode

This lets the script reuse without duplicating a single line of code:

  • m.secure_read: reads with an fstat match check before and after open
  • m.has_secret / m.is_unsafe: a regex bank covering AWS keys, GitHub PATs, sk--style API keys, JWT shapes, Slack tokens, Bearer headers, emails, phone numbers, and medical/whistleblowing terms
  • m.stage_note: saves safe notes to AI/INBOX/cloud/<sha256>.md with provenance, and writes unsafe ones to quarantine/ as JSON that leaves out the body

Split that logic across two places and eventually you fix one copy and forget the other, so keeping it in one place is deliberate.

A single-run lock to prevent concurrent runs

launchd starts the script every 5 minutes, so if the previous run hasn't finished, the new one has to skip immediately.

lock_path = vault / "AI" / ".runtime" / "memory-drive-mount-sync.lock"
m.ensure_directory_no_symlinks(lock_path.parent)
lock_fd = acquire_lock_fd(lock_path)
...
try:
    fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
    os.close(lock_fd)
    print("locked")
    return 0
Enter fullscreen mode Exit fullscreen mode

acquire_lock_fd refuses to proceed if the lock file itself is a symlink, then opens it with O_NOFOLLOW. If the non-blocking flock can't be acquired, the script prints "locked" and exits cleanly with exit 0 rather than raising an error. The key point: a concurrent start is expected, not a failure.

An unusable mount (Drive app not running, not mounted, etc.) is handled separately too:

def check_mount_dir(path: Path) -> None:
    try:
        info = os.stat(path)
    except OSError as exc:
        raise MountUnavailableError() from exc
    if not stat.S_ISDIR(info.st_mode):
        raise MountUnavailableError()
    if not os.access(path, os.R_OK | os.X_OK):
        raise MountUnavailableError()
Enter fullscreen mode Exit fullscreen mode

This returns mount_unavailable with exit 3, a different code from the lock case ("locked", exit 0) and from general errors (exit 2). That lets you tell three cases apart mechanically in the logs: "still running," "mount is gone," and "broke mid-run."

launchd config: every 5 minutes, low priority 

<key>StartInterval</key>
<integer>300</integer>
<key>LowPriorityIO</key>
<true/>
<key>Nice</key>
<integer>12</integer>
<key>ThrottleInterval</key>
<integer>60</integer>
Enter fullscreen mode Exit fullscreen mode

It starts every 5 minutes (StartInterval=300), and LowPriorityIO plus Nice 12 lower its disk I/O and CPU priority. Since the sync uses no network, the only real costs are local disk I/O and the CPU time for memory-bridge.py build, so a priority that stays out of the way of foreground work is enough. ThrottleInterval=60 is launchd's minimum restart interval, so even a crash loop can't restart more often than once a minute.

Pitfalls I hit

  • Overwriting via rename with os.replace() breaks the Drive file id → switched to seek(0) + write + truncate on the same fd
  • TOCTOU: the file gets swapped between check and open → compare (st_dev, st_ino) from lstat and fstat, and abort on mismatch
  • The next pull picks up the file you just pushed → classify_pull_candidate explicitly excludes names starting with SHARED-MEMORY
  • A diff in the updated: line alone triggers a write every run → strip the updated: line before checking whether content changed
  • Hyphens in the filename make import impossible → load by path with importlib.util.spec_from_file_location
  • Write failures when the mount is unavailable get buried among "errors" → split mount_unavailable out into its own exit code

Wrap-up

  • The Google Drive for Desktop local mount can serve as a plain file drop for syncing, with no auth layer added
  • Drive tracks file ids by inode, so never use os.replace(). Overwrite through the same fd to keep the file object intact
  • With no auth layer, POSIX-level care (fstat/inode matching, O_NOFOLLOW, flock) provides the safety instead
  • Pull is narrowed by date-based name, extension, file count, and size, and secret screening plus symlink safety checks are reused from the existing implementation so there's only one copy to maintain
  • Concurrent runs, an unavailable mount, and general errors each get their own exit code, so logs can be triaged mechanically

Next time, I plan to write about extending this single-run lock and staging quarantine approach to the other sync paths.

Have you run into file identity problems when writing to a synced folder like Drive, Dropbox, or iCloud? How did you work around them?

Written by **Lily* — I ship iOS apps and automate my content stack with Claude Code.
Follow along: Portfolio · X · GitHub*

Top comments (0)