fix(ingest): persist staged content pointers#67
Conversation
📝 WalkthroughWalkthroughThe changes add transactional staged chunk persistence, stricter key-value namespace normalization, flexible Composio proxy response decoding, and opt-in early termination for incremental sync pages containing only already-synced items. ChangesTransactional chunk persistence
Namespace sanitization
Composio synchronization
Estimated code review effort: 3 (Moderate) | ~25 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 227f9b90aa
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| let chunks_written = upsert_chunks(config, &chunks)?; | ||
| let chunks_written = with_connection(config, |connection| { | ||
| let transaction = connection.unchecked_transaction()?; | ||
| let count = upsert_staged_chunks_tx(&transaction, &staged)?; |
There was a problem hiding this comment.
When this switches ingest to upsert_staged_chunks_tx, the SQLite content column is no longer the full chunk; that helper stores only the first 500 chars as a preview. L0 sealing still hydrates leaves via hydrate_leaf_inputs in src/memory/tree/hydrate.rs, which calls get_chunk and feeds chunk.content directly to the summarizer, so any ingested document/chat chunk longer than 500 chars is summarized from a truncated preview; plain ingest_email also has no raw refs for read_chunk_body to recover the skipped email body. Please either hydrate seal inputs from the content store/raw archive or avoid staged previews until those consumers are switched.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2b68447978
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| let payload = if raw.get("successful").is_some() { | ||
| raw | ||
| } else { | ||
| raw.get("data").cloned().unwrap_or(raw) |
There was a problem hiding this comment.
When the proxy uses the backend envelope, this branch replaces the entire response with raw["data"]. If that envelope carries costUsd or markdownFormatted beside data (the flat proxied path already records costUsd, and SyncState::record_action only sees response.cost_usd), every wrapped successful call is returned with the default 0.0 cost and the sync audit/budget under-reports provider spend. Preserve the outer billing fields when unwrapping the provider payload.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
src/memory/store/kv.rs (1)
96-99: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDoc comment doesn't mention the empty→
GLOBAL_NAMESPACEfallback.The function-level doc only describes character collapsing; it doesn't mention that an empty (post-trim) namespace now resolves to
GLOBAL_NAMESPACE. This is non-obvious behavior worth documenting given the coding guidelines' emphasis on documenting non-obvious behavior.As per coding guidelines,
src/**/*.rsshould "Document public APIs, module contracts, and non-obvious behavior thoroughly, preferring module-level docs and item docs."📝 Suggested doc update
- /// Normalise a namespace into a stable storage key: lowercase is *not* - /// applied (callers may rely on case), but whitespace and path-hostile - /// characters collapse to `_` so `"team alpha/#1"` and `"team_alpha/_1"` - /// address the same bucket. + /// Normalise a namespace into a stable storage key: lowercase is *not* + /// applied (callers may rely on case), but whitespace and path-hostile + /// characters collapse to `_` so `"team alpha/#1"` and `"team_alpha/_1"` + /// address the same bucket. An empty or all-whitespace namespace falls + /// back to [`GLOBAL_NAMESPACE`](crate::memory::types::GLOBAL_NAMESPACE).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memory/store/kv.rs` around lines 96 - 99, Update the function-level documentation for the namespace normalization function to state that an empty namespace after trimming resolves to GLOBAL_NAMESPACE. Keep the existing character-collapsing and case-preservation behavior documented.Source: Coding guidelines
src/memory/sync/composio/client.rs (2)
232-240: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a doc comment explaining the envelope-vs-flat heuristic.
The branching logic in
decode_proxy_response(detecting shape by presence of top-level"successful") is non-obvious and worth documenting inline, especially since it's the crux of the new dual-shape support.As per path instructions, "Document public APIs, module contracts, and non-obvious behavior thoroughly, preferring module-level docs and item docs."
📝 Suggested doc comment
+/// Normalizes proxied Composio responses into `ExecuteResponse`. +/// +/// The proxy backend may return either: +/// - a flat shape where `successful`/`data`/`error` are at the top level, or +/// - an enveloped shape (e.g. `{ "success": bool, "data": { ... } }`) where the +/// actual response is nested under `data`. +/// +/// Presence of a top-level `successful` key is used to distinguish the two. fn decode_proxy_response(raw: serde_json::Value) -> anyhow::Result<ExecuteResponse> {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memory/sync/composio/client.rs` around lines 232 - 240, Add a concise doc comment to decode_proxy_response documenting that it accepts both flat responses and data-wrapped envelopes, using the presence of the top-level "successful" field to distinguish the shapes before deserialization. Keep the existing decoding behavior unchanged.Source: Path instructions
224-240: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDistinguish transport/JSON-parse failures from schema-mismatch failures.
Line 227 (JSON body parse failure) and line 239 (schema deserialize failure) share the identical error string
"Composio proxy response decode failed: {error}". Differentiating these would speed up debugging when a proxy backend returns valid-but-unexpected JSON shapes versus malformed responses.🔍 Suggested tweak
- serde_json::from_value(payload) - .map_err(|error| anyhow::anyhow!("Composio proxy response decode failed: {error}")) + serde_json::from_value(payload) + .map_err(|error| anyhow::anyhow!("Composio proxy response shape mismatch: {error}"))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memory/sync/composio/client.rs` around lines 224 - 240, Differentiate the error contexts in the response handling: update the response.json() error mapping in the caller to identify malformed JSON or body parsing failures, while keeping decode_proxy_response’s mapping specific to schema deserialization failures. Preserve the existing error details and successful decoding behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/memory/store/kv.rs`:
- Around line 101-112: Update the namespace sanitization flow used by
KvStore::open so existing SQLite kv_namespace keys remain readable after the
whitelist change. Add a migration/backfill or dual-read fallback from the legacy
sanitized namespace to the new value, preserving access for namespaces
containing previously supported punctuation or non-ASCII characters while
continuing to write using the new sanitizer.
In `@src/memory/sync/composio/client.rs`:
- Around line 273-305: Move the inline #[cfg(test)] mod tests containing
proxied_backend_envelope_decodes_provider_response and
flat_proxy_response_remains_supported from client.rs into the sibling
client_tests.rs module. Import the necessary client symbols there and preserve
both test cases and their assertions unchanged.
---
Nitpick comments:
In `@src/memory/store/kv.rs`:
- Around line 96-99: Update the function-level documentation for the namespace
normalization function to state that an empty namespace after trimming resolves
to GLOBAL_NAMESPACE. Keep the existing character-collapsing and
case-preservation behavior documented.
In `@src/memory/sync/composio/client.rs`:
- Around line 232-240: Add a concise doc comment to decode_proxy_response
documenting that it accepts both flat responses and data-wrapped envelopes,
using the presence of the top-level "successful" field to distinguish the shapes
before deserialization. Keep the existing decoding behavior unchanged.
- Around line 224-240: Differentiate the error contexts in the response
handling: update the response.json() error mapping in the caller to identify
malformed JSON or body parsing failures, while keeping decode_proxy_response’s
mapping specific to schema deserialization failures. Preserve the existing error
details and successful decoding behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 972603bd-fa97-48a9-a4b4-8a1139445b37
📒 Files selected for processing (4)
src/memory/ingest/pipeline.rssrc/memory/ingest/pipeline_tests.rssrc/memory/store/kv.rssrc/memory/sync/composio/client.rs
| let trimmed = namespace.trim(); | ||
| if trimmed.is_empty() { | ||
| return crate::memory::types::GLOBAL_NAMESPACE.to_string(); | ||
| } | ||
| trimmed | ||
| .chars() | ||
| .map(|c| match c { | ||
| c if c.is_whitespace() => '_', | ||
| '#' => '_', | ||
| other => other, | ||
| c if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '/') => c, | ||
| _ => '_', | ||
| }) | ||
| .collect() | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether kv_namespace rows persist across app restarts/deployments
# and whether any migration/backfill exists for previously-sanitized namespace values.
rg -n "kv_namespace" --type=rust -C3
rg -n "migration|backfill" src/memory/store -C3Repository: tinyhumansai/tinycortex
Length of output: 161
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the relevant files and inspect namespace handling.
git ls-files 'src/memory/store/*' 'src/memory/types*' | sed -n '1,200p'
printf '\n--- kv.rs outline ---\n'
ast-grep outline src/memory/store/kv.rs --view expanded || true
printf '\n--- kv.rs relevant lines ---\n'
nl -ba src/memory/store/kv.rs | sed -n '1,240p'
printf '\n--- search for namespace persistence / storage backend ---\n'
rg -n "struct .*Namespace|kv_namespace|sanitize_namespace|get_namespace|set_namespace|delete_namespace|list_namespace|records_namespace|GLOBAL_NAMESPACE" src -C 3Repository: tinyhumansai/tinycortex
Length of output: 2895
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- src/memory/store/kv.rs (selected lines) ---'
sed -n '1,380p' src/memory/store/kv.rs
printf '\n%s\n' '--- src/memory/store/kv_tests.rs ---'
sed -n '1,260p' src/memory/store/kv_tests.rs
printf '\n%s\n' '--- src/memory/store/mod.rs ---'
sed -n '1,220p' src/memory/store/mod.rs
printf '\n%s\n' '--- src/memory/store/store.rs ---'
sed -n '1,220p' src/memory/store/store.rsRepository: tinyhumansai/tinycortex
Length of output: 26062
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Look for any compatibility layer or migration that rewrites kv_namespace rows.
rg -n "sanitize_namespace|kv_namespace|backfill|migration|compat|legacy namespace|namespace.*old" src -C 3
# Check whether the KV store is documented as durable / file-backed.
rg -n "open\\(|open_in_memory|SQLite-backed|persist|durable|db_path" src/memory/store/kv.rs src/memory/store/mod.rs src/memory/store/store.rs -C 3Repository: tinyhumansai/tinycortex
Length of output: 50379
Namespace sanitization needs a migration path.
KvStore::open(db_path) persists kv_namespace in SQLite, so this whitelist change remaps any previously stored namespace containing preserved punctuation or non-ASCII characters to a different storage key. Rows written under the old sanitizer will become unreachable from the same namespace input, and there’s no backfill or dual-read fallback here.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/memory/store/kv.rs` around lines 101 - 112, Update the namespace
sanitization flow used by KvStore::open so existing SQLite kv_namespace keys
remain readable after the whitelist change. Add a migration/backfill or
dual-read fallback from the legacy sanitized namespace to the new value,
preserving access for namespaces containing previously supported punctuation or
non-ASCII characters while continuing to write using the new sanitizer.
|
|
||
| #[cfg(test)] | ||
| mod tests { | ||
| use super::*; | ||
|
|
||
| #[test] | ||
| fn proxied_backend_envelope_decodes_provider_response() { | ||
| let response = decode_proxy_response(serde_json::json!({ | ||
| "success": true, | ||
| "data": { | ||
| "successful": true, | ||
| "data": {"messages": [{"messageId": "message-1"}]}, | ||
| "error": null | ||
| } | ||
| })) | ||
| .unwrap(); | ||
|
|
||
| assert!(response.successful); | ||
| assert_eq!(response.data["messages"][0]["messageId"], "message-1"); | ||
| } | ||
|
|
||
| #[test] | ||
| fn flat_proxy_response_remains_supported() { | ||
| let response = decode_proxy_response(serde_json::json!({ | ||
| "successful": true, | ||
| "data": {"items": [1]} | ||
| })) | ||
| .unwrap(); | ||
|
|
||
| assert!(response.successful); | ||
| assert_eq!(response.data["items"], serde_json::json!([1])); | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Move tests to client_tests.rs.
New tests are added inline in client.rs via #[cfg(test)] mod tests, but the guideline requires unit tests to live in a per-file sibling module (e.g., client_tests.rs), not mixed into implementation files.
As per path instructions, "Keep tests in per-file <name>_tests.rs siblings, such as store.rs and store_tests.rs, rather than mixing tests into implementation files."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/memory/sync/composio/client.rs` around lines 273 - 305, Move the inline
#[cfg(test)] mod tests containing
proxied_backend_envelope_decodes_provider_response and
flat_proxy_response_remains_supported from client.rs into the sibling
client_tests.rs module. Import the necessary client symbols there and preserve
both test cases and their assertions unchanged.
Source: Path instructions
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5dd9b8d890
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| c if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '/') => c, | ||
| _ => '_', |
There was a problem hiding this comment.
If a workspace already has KV rows in namespaces containing characters that the old sanitizer preserved, such as project.v1 or skill:gmail, this new allowlist rewrites future lookups to project_v1 / skill_gmail while the existing rows remain stored under the original namespace string. Because KvStore::open only installs the schema and does not migrate kv_namespace.namespace, those persisted records become invisible to get_namespace, list_namespace, and records_namespace after upgrade; either preserve backward-compatible lookup or migrate existing namespace keys when changing the normalization.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 671e78a014
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| } | ||
|
|
||
| page_token = fetched.next; | ||
| if source.stop_on_empty_pending() && !saw_unsynced_item { |
There was a problem hiding this comment.
When Gmail has more results than one capped run can process, more_pending prevents cursor advancement, leaving the next tick to start from the first page again with those items already in synced_ids. If that first page is now all deduped but still has a nextPageToken, this new break exits before following the token, so older unsynced Gmail pages remain permanently unreachable after hitting max_items or max_pages on a prior run.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/memory/sync/composio/orchestrator.rs (1)
277-298: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftDon’t stop paging on pages with unreadable items
Pages containing only items without a dedup key are treated the same as fully deduplicated pages. In Gmail,dedup_key()can returnNonefor messages missingid/messageId, sostop_on_empty_pending()can end paging before a later page with valid new messages. Track “unprocessable” items separately from “already synced” ones.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memory/sync/composio/orchestrator.rs` around lines 277 - 298, Update the paging state in the orchestrator loop around saw_unsynced_item and source.dedup_key so unreadable items with no dedup key are tracked separately from already-synced items. Ensure stop_on_empty_pending() does not treat a page containing only unprocessable items as empty, while preserving existing handling for fully deduplicated pages.
🧹 Nitpick comments (2)
src/memory/sync/composio/orchestrator.rs (2)
68-70: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMissing docs on new public trait method.
stop_on_empty_pending()is a new public contract method onIncrementalSourcewith non-obvious semantics (it changes pagination termination behavior) but has no doc comment explaining when implementers should override it or what "empty pending" means.📝 Suggested doc
+ /// Returns `true` if the source can safely stop paging once a fetched + /// page contains no items that are new/unsynced (i.e. every item was + /// either already synced or had no dedup key). Only enable this for + /// sources whose pages are ordered such that an "empty pending" page + /// guarantees no further unsynced items remain (e.g. reverse-chronological + /// feeds using a persisted cursor). fn stop_on_empty_pending(&self) -> bool { false }As per coding guidelines, "Document public APIs, module contracts, and non-obvious behavior thoroughly, preferring module-level docs and item docs."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memory/sync/composio/orchestrator.rs` around lines 68 - 70, Document the public IncrementalSource::stop_on_empty_pending method with an item-level doc comment explaining that “empty pending” means no pending items remain and that returning true terminates pagination in that case; state when implementers should override the default false behavior.Source: Coding guidelines
384-392: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNo test coverage for the new early-stop branch.
This new control-flow path (early break on
stop_on_empty_pending()) has no accompanying test in this diff. Given the subtlety flagged above around dedup-key semantics, a targeted unit test (e.g., a fakeIncrementalSourcewhose page has items with no dedup key vs. items already synced) would help lock in intended behavior.As per coding guidelines, tests should live in
<name>_tests.rssiblings such asorchestrator_tests.rs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memory/sync/composio/orchestrator.rs` around lines 384 - 392, Add targeted unit coverage in the orchestrator test sibling for the early-break branch in the sync flow containing stop_on_empty_pending(). Use a fake IncrementalSource to verify pages with items lacking dedup keys and pages containing only already-synced items trigger the intended stop behavior, while preserving continued synchronization when an unsynced item is observed.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/memory/sync/composio/orchestrator.rs`:
- Around line 277-298: Update the paging state in the orchestrator loop around
saw_unsynced_item and source.dedup_key so unreadable items with no dedup key are
tracked separately from already-synced items. Ensure stop_on_empty_pending()
does not treat a page containing only unprocessable items as empty, while
preserving existing handling for fully deduplicated pages.
---
Nitpick comments:
In `@src/memory/sync/composio/orchestrator.rs`:
- Around line 68-70: Document the public
IncrementalSource::stop_on_empty_pending method with an item-level doc comment
explaining that “empty pending” means no pending items remain and that returning
true terminates pagination in that case; state when implementers should override
the default false behavior.
- Around line 384-392: Add targeted unit coverage in the orchestrator test
sibling for the early-break branch in the sync flow containing
stop_on_empty_pending(). Use a fake IncrementalSource to verify pages with items
lacking dedup keys and pages containing only already-synced items trigger the
intended stop behavior, while preserving continued synchronization when an
unsynced item is observed.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: ebbf963b-03ec-4b9f-bc07-311f150e6aa9
📒 Files selected for processing (2)
src/memory/sync/composio/gmail.rssrc/memory/sync/composio/orchestrator.rs
1 participant
Summary
Validation
cargo test --lib ingest_chat_writes_chunks_and_enqueues_extract_jobs --quietcargo clippy --all-targets -- -D warningsSummary by CodeRabbit