feat(sync): write-through Brain.correct() → /api/v1/ingest (day 3 of #194)

[Gradata]

Day 3 of #194 — wire Brain.correct() to write-through cloud sync.

What

Every Brain.correct(draft, final) now enqueues into local sync_queue + a background daemon thread POSTs each row to api.gradata.ai/api/v1/ingest.

Why

Removes the dependency on session-end hooks (which barely fire for long-running agents on hermes --continue) and the cron sync band-aid. See /home/olive/gradata-office-hours-memo.md for the full architectural rationale.

Env vars

• GRADATA_API_KEY required to start worker
• GRADATA_DISABLE_WRITE_THROUGH=1: opt-out
• GRADATA_CLOUD_INGEST_URL (default https://api.gradata.ai/api/v1/ingest)
• GRADATA_SYNC_TICK_SEC (default 30)

Failure modes

• Cloud unreachable: row stays pending, retried next tick
• 422 permanent: poison row marked synced+failed
• 429: backoff, bail batch
• No API key: worker doesn't start, local correct() unaffected
• enqueue exception: caught, local correct() always succeeds

Tests

38 passed (test_sync_queue × 9, test_sync_worker × 9, test_brain_write_through × 5, test_brain × 15 regression).

Series

• feat(sync): sync_queue primitives (day 1 of #194) #195 (day 1, draft): SDK sync_queue primitives
• Gradata/gradata-cloud#58 (day 2, merged): cloud /ingest receiver
• This PR (day 3): SDK wire-up

[coderabbitai]

Caution

Review failed

The pull request is closed.

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro

Run ID: a08a8bf9-1013-463a-be64-954a778f046e

📥 Commits

Reviewing files that changed from the base of the PR and between 0561c82 and 30d7dfe.

📒 Files selected for processing (9)

• Gradata/src/gradata/_migrations/__init__.py
• Gradata/src/gradata/_sync_queue.py
• Gradata/src/gradata/_sync_worker.py
• Gradata/src/gradata/brain.py
• Gradata/src/gradata/onboard.py
• Gradata/tests/test_brain_write_through.py
• Gradata/tests/test_daemon_sync.py
• Gradata/tests/test_sync_queue.py
• Gradata/tests/test_sync_worker.py

---

📝 Walkthrough

• Implements write-through cloud sync for Brain.correct(): each correction is enqueued to a local SQLite queue and posted to cloud ingest endpoint (default https://api.gradata.ai/api/v1/ingest) by a background daemon thread
• Worker starts automatically when API key is resolvable (GRADATA_API_KEY env var or ~/.gradata/key) and can be disabled via GRADATA_DISABLE_WRITE_THROUGH=1
• Configurable sync behavior via environment variables: GRADATA_API_KEY, GRADATA_DISABLE_WRITE_THROUGH, GRADATA_CLOUD_INGEST_URL, GRADATA_SYNC_TICK_SEC (default 30 seconds)
• Differentiates failure modes: transient errors (5xx, network) retry next tick; 422 marked as poison/failed to prevent retries; 429 triggers backoff and batch abort; local correct() always succeeds even if cloud sync fails
• Adds new sync_queue table to SQLite schema with columns for payload, kind, enqueue/sync timestamps, retry tracking, and partial index for pending rows
• New modules: _sync_queue.py (enqueue/CRUD primitives), _sync_worker.py (daemon with tick/backoff/poison handling), updated brain.py with worker lifecycle management
• No breaking changes to existing Brain public API; Brain.close() now gracefully drains pending sync queue
• Comprehensive test coverage: 38 tests for queue primitives, worker HTTP handling (2xx, 422, 429, 5xx, network errors), and Brain write-through scenarios

Walkthrough

This PR implements write-through cloud sync for Brain corrections. A new sync_queue SQLite table buffers corrections with timestamps and retry metadata. The _sync_queue module provides enqueue, peek, and mark operations. A background SyncWorker daemon drains the queue by POSTing to a cloud ingest endpoint and handles HTTP statuses (2xx marks synced, 429 aborts batch, 422 marks permanent, 5xx/network retries). Brain integrates the worker during construction (if API key available and write-through enabled), enqueues corrections after local processing, and drains queued items on close.

Changes

Write-through cloud sync

Layer / File(s)	Summary
Schema and migration setup src/gradata/_migrations/__init__.py, src/gradata/onboard.py	Adds sync_queue table with payload, kind, enqueue/sync timestamps, retry attempts, and error columns; includes partial index on synced_at IS NULL for pending row lookups.
Sync queue primitives and tests src/gradata/_sync_queue.py, tests/test_sync_queue.py	Implements enqueue (JSON serialization, row id return), enqueue_correction (convenience wrapper), peek_pending (FIFO with fallback JSON decode), mark_synced (timestamp update), and mark_failed (increment attempts, truncate error); unit tests cover FIFO ordering, state transitions, constraint validation, and error truncation.
Sync worker daemon and tests src/gradata/_sync_worker.py, tests/test_sync_worker.py	Implements SyncWorker as daemon thread that ticks repeatedly, fetching up to 50 pending rows and POSTing each to cloud ingest with bearer auth; differentiates status handling (2xx syncs, 429 batch-aborts, 422 permanent, 5xx/network transient); tests cover happy path, failure modes, retries across ticks, lifecycle (stop/drain), and at-least-once semantics when DB-write fails.
Brain write-through wiring and integration tests src/gradata/brain.py, tests/test_brain_write_through.py	Brain resolves API key from env/file, starts worker on init if available and not disabled, enqueues corrections after local processing via _write_through_enqueue (builds cloud payload, computes event_id, manages fresh connection), stops worker on close with drain; integration tests verify worker start conditions, enqueue behavior, cloud error resilience, and local path correctness.
Daemon sync test update tests/test_daemon_sync.py	Updates empty-brain sync test to expect heartbeat POST and non-fatal cloud failures per PR #198 behavior.

Sequence Diagram(s)

Write-through sync workflow from correction to cloud delivery:

sequenceDiagram
    participant User
    participant Brain
    participant SyncWorker as SyncWorker<br/>(daemon)
    participant Queue as sync_queue<br/>(SQLite)
    participant Cloud
    
    User->>Brain: brain.correct(draft, final)
    Brain->>Brain: local correction logic
    Brain->>Queue: enqueue_correction(brain_id, correction, event_id)
    Queue-->>Brain: row_id
    Note over Brain,Queue: enqueue best-effort, errors logged
    
    Note over SyncWorker: background tick loop
    SyncWorker->>Queue: peek_pending(limit=50)
    Queue-->>SyncWorker: list of pending rows
    
    loop for each row
        SyncWorker->>Cloud: POST /api/v1/ingest<br/>payload + bearer token
        alt 2xx success
            Cloud-->>SyncWorker: 200/201
            SyncWorker->>Queue: mark_synced(id)
        else 429 rate limit
            Cloud-->>SyncWorker: 429
            SyncWorker->>Queue: mark_failed(id, error)
            SyncWorker->>SyncWorker: abort tick, sleep
        else 422 permanent
            Cloud-->>SyncWorker: 422
            SyncWorker->>Queue: mark_failed then mark_synced
        else 5xx/network
            Cloud-->>SyncWorker: 500 or timeout
            SyncWorker->>Queue: mark_failed(id, error)
            Note over SyncWorker,Queue: continues batch, retries next tick
        end
    end
    
    User->>Brain: brain.close()
    Brain->>SyncWorker: stop(drain=True)
    SyncWorker->>SyncWorker: final synchronous tick
    SyncWorker->>Brain: join thread

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related issues

• Write-through sync in brain.correct() — eliminate hook+session dependency #194: This PR directly implements the write-through sync feature (sync_queue table, primitives, SyncWorker daemon, and Brain integration) specified in that issue.

Suggested labels

feature

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/brain-correct-write-through-day3

Warning

There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure.

🔧 OpenGrep (1.20.0)

OpenGrep fatal error (exit code 2):
┌──────────────┐
│ Opengrep CLI │
└──────────────┘

�[32m✔�[39m �[1mOpengrep OSS�[0m
�[32m✔�[39m Basic security coverage for first-party code vulnerabilities.

�[1m Loading rules from local config...�[0m
[00.25][ERROR]: Error: exception Glob.Lexer.Syntax_error("malformed glob pattern: missing ']'")
Raised at Glob__Lexer.syntax_error in file "libs/glob/Lexer.mll", line 8, characters 2-26
Called from Glob__Lexer.__ocaml_lex_token_rec in file "libs/glob/Lexer.mll", line 29, characters 26-53
Cal

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[greptile-apps]

Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.