Repository navigation
feat(ingestion): submit all sources then poll the last episode (v0.3.0) - #607
Merged
Merged
Conversation
Stop treating sequential ingest as wait-for-each-file. Queue every file into a graph first; wait() is opt-in, polls the tail of the per-graph queue, and scales its timeout with how many episodes were submitted. Bump zep-ingest to 0.3.0 for the release. Co-authored-by: Cursor <cursoragent@cursor.com>
There was a problem hiding this comment.
Critical — ingestion/src/zep_ingest/result.py:217: combine() does not verify that results target the same graph/user, yet tail polling marks every combined episode processed when only the final UUID completes. Combining sequential results from different graphs can therefore make wait() return while earlier episodes are still processing. Store destination identity in each result and reject mismatched combinations, or poll one tail per destination. Add a cross-destination test.
Give episodes and manual graph updates more headroom before wait() raises IngestTimeoutError. Co-authored-by: Cursor <cursoragent@cursor.com>
There was a problem hiding this comment.
- Critical —
ingestion/src/zep_ingest/result.py,combine()/_refresh_episodes(only_last=True)hunks:combine()neither verifies a shared graph destination nor preserves/verifies actual submission order. Combining results from different graphs—or callinglater.combine(earlier)—can poll an unrelated/earlier episode and then mark every episode processed, causingwait()to return while work remains. Track destination and submission ordering inIngestResultand reject incompatible combinations, or poll every combined episode instead of applying the tail optimization.
Poll thread backfills via the last message UUID per request, document batch vs episode vs task handles, and warn against mixing batch and sequential graph.add on one graph. Co-authored-by: Cursor <cursoragent@cursor.com>
There was a problem hiding this comment.
Critical
ingestion/src/zep_ingest/result.py,_refresh_episodes()/combine()hunks:combine()does not verify that results target the same graph/user, yet tail polling marks every combined episode processed when onlyepisode_uuids[-1]completes. Combining sequential submissions from different destinations can therefore makewait()returnsucceededwhile earlier ingestion is still processing. Store and validate the destination when combining, or retain separate completion tails per submission/destination and poll each one.
No other actionable findings.
Regular thread.add_messages returns message_uuids only (verified against production). Remove dead task_id fallback, poll one tail per thread, and stop falsely marking all threads done when one saga finishes. Co-authored-by: Cursor <cursoragent@cursor.com>
There was a problem hiding this comment.
Critical
ingestion/src/zep_ingest/result.py,IngestResult.combine()hunk (_single_queue_episode_pollaggregation): Combining multiple results incorrectly preserves tail-only polling whenever each input individually represents one queue. For example, combining two sequential single-thread ingests for different threads leaves the flagTrue;wait()polls only the second thread’s UUID and marks every UUID processed when it completes, even if the first thread is still processing. The same false-success can occur when combining sequential graph ingests for different graph destinations because destination identity is not retained or validated. Track each result’s queue/destination identity and enable tail-only polling only when all episode handles demonstrably share that queue; otherwise set_single_queue_episode_poll=Falseand poll every tail independently. Add tests combining separate single-thread results and results from different graph destinations.
Document pre-release prod checks and finalize 0.3.0 changelog notes for thread polling and combine(). Co-authored-by: Cursor <cursoragent@cursor.com>
There was a problem hiding this comment.
Critical
ingestion/src/zep_ingest/result.py,IngestResult.combine()hunk: combining independently queued episode results preserves_single_queue_episode_poll=True. This is unsafe for results from different threads or graphs:wait()polls only the final UUID, then marks every combined UUID processed, potentially returningsucceededwhile earlier queues are still running. Set the flag to false when combining separate episode-bearing results, or retain queue boundaries and poll one tail per queue. Add a test combining two single-thread results where only the second tail is processed.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
graph.addvs the Batch API — not “finish extracting this file before sending the next.”wait()is opt-in and aligned with Check data ingestion status:batch_idviabatch.getgraph.add: poll last-submitted episodethread.add_messages: poll last message UUID per thread (message_uuidsonly — regularadd_messagesdoes not returntask_id)task_idmax(120s, 60s × items_submitted). Passtimeout=Nonefor unbounded wait.SourcePathssequences,ConcatLoader, andIngestResult.combine()for mixed/multi-call submits.combine()merges poll handles only (notnode_uuids/edge_uuids). Prefer separatewait()for seeding vs episodes.Test plan
cd ingestion && uv sync --extra dev && uv run pytest tests/ -m "not integration"(678 passed)uv run ruff check . && uv run ruff format --check . && uv run mypy src/keybank run zep-prod -- env ZEP_API_URL=https://api.getzep.com uv run python scripts/release_smoke_prod.py(7/7)main→ publisheszep-ingest0.3.0 (zep-ingest-v0.3.0)