feat(ingestion): add hardened zep-ingest pipeline#562
Merged
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
New standalone package owning everything upstream of the Zep API:
loaders (Slack exports, text/Markdown, .eml email, JSONL/CSV/JSON
records), transforms (chunking, LLM contextualization, alias
canonicalization, JSON normalization, always-on episode-limit guard),
Batch API + sequential submitters with auto fallback, thread-message
backfills, fact triples, and preview()/warnings for temporal
correctness — plus examples, a starter ontology, test and release
workflows.
Includes post-review fixes: batch.process failures are retried and
recorded instead of crashing the run, JsonNormalizer no longer emits
empty {} episodes, non-string timestamps fail validation with a named
error, loader warnings surface through the pipeline, and the release
workflow skips (not fails) on non-ingestion tags.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
jackaldenryan
force-pushed
the
create-ingestion-pipeline
branch
from
July 20, 2026 20:25
332f0af to
0aa2ae3
Compare
…n gaps Enterprise Grid organization exports name the user roster org_users.json rather than users.json, so SlackExportLoader silently resolved nothing and ingested every author/@mention as a raw U0... ID. Read org_users.json as a fallback, and surface the previously-silent degradation instead of hiding it: a warning when no roster file is present, a count of user IDs referenced in messages but absent from the roster (deactivated/bot/Slack Connect users), and a ConfigurationError when the input is not a Slack export at all. Warnings flow through preview()/result.warnings via a new loader warnings list. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- Create canonical nodes through the public client.graph.add_nodes SDK method instead of a private-transport POST; the package now uses only the public SDK surface. Require zep-cloud>=3.25.0 (where add_nodes/AddNodeItem are public). - Map NodeItem.uuid to the SDK's uuid_ field (serialized to the wire "uuid" key) and omit unset fields, avoiding a silent identity drop and null clobbers on upsert. - README: default entity/edge types apply to user graphs only, so named (standalone) graphs must declare every type they rely on; also document ingest_nodes / NodeItem direct node seeding. - Remove the no-op _capped_metadata helper (episode metadata is validated at construction) and its vestigial warnings parameter. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The method="auto" dispatch probed batch availability with a bare client.batch.create() call. A transient 429/5xx on that one call propagated straight out of run(), crashing the ingestion after the full load+validate pass had already run. Wrap the probe in call_with_retries, matching every other batch call: transient errors are retried, a genuine plan-gating status still falls back to sequential, and a non-gating error that survives retries is re-raised rather than silently downgraded to sequential. Adds tests for transient-then-recover and persistent-error-raises. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The pre-defined conversational pipelines no longer emit `message` episodes. The Slack, transcript, and email loaders now produce `text` episodes; each already embeds its speaker/author context inline — `Sender (Slack #channel, time): …`, `Speaker: …`, `Email from … to …` — so no attribution is lost. `message` remains a valid `data_type` for directly constructed episodes and is still split line-aware by LimitGuard. Also commits the previously-uncommitted restriction of the direct dataclass paths (`ingest_fact_triples`, `ingest_nodes`, `ingest_thread_messages`) to JSON input (JSONL or JSON array); a `.csv` path is now rejected with a clear error because those schemas carry list/mapping fields CSV cannot express. `ingest_json_records` still accepts CSV for flat tabular data. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
mehulp93
marked this pull request as ready for review
July 24, 2026 16:35
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
-
Critical —
ingestion/src/zep_ingest/threads.py:130: Existing global thread IDs are accepted without verifying ownership. If the ID belongs to another user, subsequent messages are written into that user’s thread, causing cross-user data leakage. Fetch the existing thread and confirm itsuser_id; otherwise reject the ingest. -
Critical —
ingestion/src/zep_ingest/loaders/slack.py:64: Channel names fromchannels.jsonare joined directly to the export directory. A name such as../../privatecan make ingestion read local JSON files outside the export and submit their contents to Zep. Resolve paths and enforce containment within the export root; also reject traversal components and escaping symlinks.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
-
Critical —
.github/workflows/release-ingestion.yml:80: The build job checks outinputs.refindependently after testing. Because dispatch accepts mutable refs, the branch/tag can move between jobs, allowing untested code—or a different package version—to be built and published. Resolve the input to a commit SHA once, validate it as an immutable tag/SHA, and pass that exact SHA to every job; also verify the version in the build job. -
Critical —
ingestion/src/zep_ingest/submitters/sequential.py:44:call_with_retriesretries non-idempotent POST operations after 5xx responses. If the server committedgraph.add,thread.add_messages, oradd_fact_triplebut the response was lost, the retry duplicates data and can corrupt ingestion timelines. Supply stable idempotency identifiers where supported, or avoid retrying ambiguous failures for non-idempotent calls. Add a test covering “request committed, response failed.”
There was a problem hiding this comment.
-
Critical —
ingestion/src/zep_ingest/loaders/email.py:47: Valid RFC email dates using the-0000timezone produce a naivedatetimefromparsedate_to_datetime(). Its.isoformat()value is then rejected byEpisode, aborting the entire ingestion. Treat naive parsed dates as UTC/unknown-offset or leavecreated_atunset with a warning. Add a test coveringDate: ... -0000. -
Warning —
ingestion/src/zep_ingest/loaders/slack.py:105:groupingis not validated. Any typo, such asgrouping="messages", silently selects thread grouping because every value other than"message"follows that branch. Reject values outside"thread"and"message"during initialization and test invalid input. -
Warning —
ingestion/src/zep_ingest/submitters/sequential.py:62:Retry-Aftersupports both delay-seconds and an HTTP date, but the parser only accepts numeric values. A standards-compliant date therefore falls back to a short exponential delay, potentially exhausting retries before the instructed retry time. Parse HTTP dates and clamp negative delays to zero; add coverage for both date and malformed headers.
There was a problem hiding this comment.
Critical
- [ingestion/src/zep_ingest/threads.py:143]
client.thread.get(..., limit=1)uses an unsupported SDK argument. Existing repository usage passeslastn, so an existing-thread collision raisesTypeErrorinstead of verifying ownership, breaking reruns. Replacelimit=1withlastn=1and assert the exact call in tests.
There was a problem hiding this comment.
-
Critical —
.github/workflows/release-ingestion.yml:31: dispatch refs matchingzep-ingest-v…are assumed to be immutable tags, but the workflow never verifies that the ref resolves underrefs/tags/. A mutable branch with that name could therefore be published to PyPI. After checkout, require non-SHA refs to resolve torefs/tags/$REFand verify that tag’s commit equalsHEAD. -
Critical —
ingestion/src/zep_ingest/loaders/json_records.py:35:_parse_timestamp()silently assigns UTC to timezone-naive input. Backfills containing local timestamps will consequently receive incorrect instants, corrupting validity ordering. Treat naive timestamps as unparseable/missing, consistent with the package’s other timestamp validation, or require an explicit caller-supplied timezone. -
Warning —
ingestion/src/zep_ingest/transforms/canonicalizer.py:79: withstrict=False, aliases differing only by case can map to different canonical names._lowered_aliasesthen silently keeps the last mapping, so mixed-case occurrences are rewritten unpredictably. Detect case-folded collisions during construction and raiseConfigurationError.
There was a problem hiding this comment.
- Critical — .github/workflows/release-ingestion.yml, checkout hunk: The workflow checks out
${{ github.sha }}, which for areleaseevent represents the repository’s default-branch commit, not necessarily the released tag. The subsequent comparison againstTAG_SHAtherefore rejects valid releases whenever the tag is not exactly at the current default-branch head, preventing publication. Check out or fetch${{ github.event.release.tag_name }}directly, resolve its commit, and archive that resolved SHA; do not compare it withgithub.shafor this event type.
There was a problem hiding this comment.
Critical
- ingestion/src/zep_ingest/submitters/sequential.py,
_add_episode()hunk —graph.add()responses without a UUID are appended toepisode_uuidsasNone. Subsequentrefresh()calls querygraph.episode.get(uuid_=None), sowait()fails or never completes instead of reporting the documenteduntrackedstate. This case is handled correctly for nodes, triples, and thread messages but lacks equivalent handling and a regression test here. Checkgetattr(zep_episode, "uuid_", None); append a valid UUID, otherwise incrementuntracked_items, and add a sequential-submitter test for a response without a UUID.
… source_description Every episode now records its provenance as structured, queryable metadata instead of a single free-text source_description: - source_type on every episode: document, slack, transcript, email, json_record - plus source-specific keys: document/email/json_record file_name, Slack channel + thread_ts, transcript meeting, email subject Removes Episode.source_description and its Batch/graph.add mappings; the loaders, chunker, contextualizer, limit guard, and validated-replay no longer thread it through. Tests updated to assert the new metadata schema. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
A defect review of today's commits surfaced a set of real bugs and a batch of docs that had drifted from the code. Fixes, grouped by theme: Reliability - Retry transient failures on batch.create in the thread path and on episode rollovers, matching the episode probe. A 429 previously aborted a whole backfill after the user and threads had already been created. - A rollover that cannot open its next batch now stops the run and records the reason instead of raising. The batches already submitted are still processing and their ids are the only handle on them; raising discarded them. - Reject a row that omits a required column with a ConfigurationError naming the field and the row, not a raw TypeError. Required fields are now derived from the dataclass, so threads, fact triples, and nodes are all covered and cannot drift again. Correctness - Loader-stamped provenance can no longer be overwritten by a record field of the same name; a collision is dropped with a warning and the record still reaches the graph intact. An over-budget metadata_fields list is rejected up front rather than crashing partway through a file. - Report Zep API failures as the API reported them, passing the server's status and message through unchanged. The package no longer substitutes its own redacted string, which hid the reason for quota and authorization failures. - Narrow the batch-to-sequential fallback to a missing endpoint (404). A refused key or an exhausted quota would refuse graph.add just as readily, so falling back only hid the real error behind a slow run. API - Remove the wait/poll_interval/timeout parameters. Waiting is now always result.wait(...), which binds the result before the call that can fail; the old shape discarded the ids needed to resume whenever wait() raised. - Stop leaking importlib's version() as zep_ingest.version, which shadowed the obvious accessor and returned a function. - Pin the full public surface with an exact-equality assertion. Docs - The Batch API is not enterprise-only and is not gated on a plan; remove that claim throughout. - Drop the JSON "normalization" the normalizer removal made obsolete, give the quickstarts the ontology step they told readers was required, and correct a metadata key that no code emits. - Correct the timestamp claim: three of the five one-liners leave created_at unset unless the caller opts in. Co-Authored-By: Claude <noreply@anthropic.com>
Each of these silently produced a wrong result rather than failing, so none were visible from the test suite or the type checker. - raise_for_status() ignored a canceled ingest. "canceled" is terminal, so wait() returned normally and the strictness check then stayed silent on a run that never finished. The unsuccessful set is now derived from the terminal status sets, so a status added later cannot slip past the same way. "untracked" stays excluded: no completion handle means unknown, not failed. - JSON field mappings read the dict they were writing into, so an earlier mapping could feed a later one — id_field="sku" with name_field="id" gave name the SKU. Mapping sources now read the untouched record, making the result independent of mapping order. created_at_field reads the record too; the metadata lift still reads the emitted episode, so a lifted value always matches the body under the same key. - An alias that was also another canonical name was accepted and then silently ignored, because canonical protection wins in _resolve. Such a map is self-contradictory and is now rejected at construction, alongside the existing alias-collision validation. - Splitting overwrote a caller's "chunk" or "part" metadata with the internal sequence marker; the guard that looked like it prevented this was a key budget test, not value protection. The marker is diagnostic and the caller's value is domain data, so a collision now omits the marker and warns, reusing the omit-and-warn path already there for a full key budget. - Node UUID uniqueness compared text, so two spellings of one UUID passed the guard and were submitted together — and with batching they could land in different requests, where nothing else could catch them. UUIDs are now canonicalized at construction, so the value compared is the value sent, and the error names the offending UUIDs. - WebVTT voice end tags leaked into transcript text: <v Alice>Hello</v> produced "Alice: Hello</v>", corrupting every line of any transcript using the closing form, including the wrapped-cue case. Co-Authored-By: Claude <noreply@anthropic.com>
…identifiers _is_vtt_identifier scanned forward past blank lines for a timing line, and treated the current line as a cue identifier if it found one. But a cue's last payload line is also followed — after the blank line separating cues — by the next cue's timing line, so it matched too and was discarded. Cue identifiers are optional in WebVTT and routinely omitted, and in a file without them every payload line matched: a three-cue transcript loaded as a single turn containing only the last line, with no warning. Existing tests all used identifiers, so nothing caught it. A cue identifier is now recognized only where the spec puts one: directly above the timing line with no blank between, and opening its own block. The second half matters independently — without it, a payload line left flush against the next timing line is dropped the same way. Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
Critical
.github/workflows/test-ingestion.yml:41— The matrix installs each requested Python version withuv python install, butuv syncis not told to use it. On GitHub runners, uv may select the preinstalled system Python, so all matrix entries can test the same interpreter while appearing to cover 3.11–3.13. The release workflow repeats this pattern. Pin the interpreter (uv python pin "${{ matrix.python-version }}") or pass--pythontouv sync/uv run.
Warning
ingestion/src/zep_ingest/verify.py:68andingestion/src/zep_ingest/result.py:201— Both polling loops sleep for the fullpoll_intervalwithout considering the remaining timeout. For example,timeout=1, poll_interval=3600can block for roughly an hour despite the one-second timeout. Sleep formin(poll_interval, remaining_time)and check the deadline before sleeping.
…on defects
Slack exports index conversations across four files — channels.json,
groups.json, dms.json, mpims.json. The loader read only channels.json, so
private channels and DMs were skipped with no warning: a successful run and a
graph missing most of the export.
All four are now read, and conversation_types= selects any subset. It defaults
to public channels only, which preserves the effective behavior and keeps
private conversations from flowing into a graph unnoticed; anything present but
unselected is reported with counts and the parameter to pass. channels= filters
by name within the selected types, and naming a conversation whose type is not
selected is an error rather than a silently empty run. DMs are labeled by their
resolved members rather than the opaque id or mpdm- slug Slack names the folder
with, since raw ids degrade extraction.
Also:
- An all-caps speaker turn ("ALICE: ...") is shaped exactly like a KEY: value
header, so the header scanner consumed the turns as metadata and left nothing
to ingest. A KEY: value line is now metadata only if it has no value — which
a turn cannot have — or if a word of its key names transcript metadata. The
asymmetry is deliberate: misreading a turn as a header can empty a file,
while misreading a header as a turn costs one line.
- messages_per_call had no upper bound, so a value above the API's 30-per-call
limit failed as an HTTP 400 after the user and threads were created. Bound
and default now share one constant.
- Metadata and attributes rejected arrays, but the API accepts a scalar or an
array of scalars for both — a value like {"teams": ["sales", "support"]} was
refused before it was ever sent. One shared rule now mirrors the API,
including that empty arrays and null elements are not accepted.
- Correct the security note that promised AddError omits raw API response
bodies. It carries them deliberately, so the failure reads as the API
reported it; the note now says so, and that nothing is added from submitted
episodes.
Co-Authored-By: Claude <noreply@anthropic.com>
…y default A Slack export wrapped in a single folder was only unwrapped if that folder held users.json or channels.json. An Enterprise Grid export names its roster org_users.json, and an export without public channels has no channels.json, so such an archive stayed wrapped: every index lookup missed, the loader concluded there were no indexes, and fell back to listing directories — where the wrapper itself became a single pseudo-channel whose "day files" were the nested message files of every conversation inside it. The result was DM content ingested under the default conversation_types of public channels only, labelled public_channel, with no skipped-type warning, because nothing knew there was anything to skip. Reproduced, and fixed on three independent fronts: - The wrapper is recognized by any export marker — all four conversation indexes and both roster spellings — not just two of them. - Day files are a conversation folder's direct children. Returning nested paths is what let one folder serve another conversation's messages, and it also pulled in the subfolders newer exports place inside a conversation. - An index-less export no longer types every folder as a public channel. DM and group-DM folder names are unambiguous, so they are typed as such and excluded by the default; the remaining folders are ingested with an explicit warning that their type could not be determined, rather than silently assumed. Also: - Metadata and attribute keys must be non-blank strings. A non-string key was accepted client-side and rejected by the API at submission — on the batch path, after batch.create had already opened a draft. - Reject NaN and Infinity in JSON records. Python reads and writes them, but they are not valid JSON, so the episode body was unparseable and a lifted value was silently sent as null. The record is now refused with the file, record number and field path named, and the emit itself is strict. - Correct the resume example: status reads cached state, so it needs refresh() first. The comment claimed it refreshed on demand. Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
- Critical —
ingestion/src/zep_ingest/submitters/sequential.py:34-43:_retry_after_seconds()accepts negative orNaNnumericRetry-Aftervalues. These reachtime.sleep()and raiseValueError, aborting ingestion instead of retrying or returning anAddError. Treat non-finite or negative values as invalid (or clamp to zero), and add tests for malformed numeric headers.
…e run _retry_after_seconds returned float(value) for the numeric form of the header without checking range or finiteness, so "Retry-After: -1" or "nan" reached time.sleep and raised an uncaught ValueError. That broke call_with_retries' documented contract of returning (None, last_error): a single malformed header aborted the whole import on the first retry instead of retrying the request or recording an AddError. A non-finite value carries no delay to honor, so treat it as a missing header and fall back to the exponential backoff that None already selects. Clamp a negative value to zero, matching what the RFC-date branch has always done for an already-elapsed date. MAX_RETRY_WAIT_SECONDS already bounded the upper side; only the lower side and NaN were unguarded. Co-Authored-By: Claude <noreply@anthropic.com>
…json pieces valid Three small defects left over from the previous round, none a data leak. - A Slack export wrapped in a single folder was unwrapped when read from a zip but not when read from disk, so the same export ingested nothing once extracted: the wrapper read as one conversation whose only files were the index files. The directory reader now re-roots the same way, declining to follow a symlink out of the export. - SlackExportLoader reset its unresolved-user set at the start of load() but appended to the previous pass's warning list, so a second load() reported every warning twice. Both are per-pass state now. - Splitting a json episode re-serialized whatever json.loads accepted, and Python accepts NaN and Infinity in both directions. An oversized body carrying one was split into pieces still labelled json that no strict parser accepts. The split now parses strictly and declines, which routes the body through the existing text fallback and its warning — pieces are labelled for what they are rather than claiming a structure they do not have. Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
Warning
- ingestion/src/zep_ingest/_validation.py:106 —
require_nonnegative_number()acceptsNaNand positive infinity because both bypassvalue < 0. Public parameters such astimeout,poll_interval, andmin_intervalcan consequently cause infinite polling or atime.sleep()runtime error. Reject non-finite values withmath.isfinite(value)and add tests coveringnanandinf.
Two gaps with one root cause: a comparison that is False for NaN and +inf, so neither value was ever caught. - require_nonnegative_number guarded with `value < 0`, which is False for both NaN and +inf. Six call sites share it, so poll_interval=nan reached time.sleep as a ValueError and poll_interval=inf as an OverflowError. Worse, timeout=nan or inf silently stopped being a timeout: `elapsed >= nan` and `elapsed >= inf` are never true, so IngestResult.wait() and search_when_ready() polled forever instead of giving up. The validator now requires a finite value, raising the same ConfigurationError it already raised for a negative one. The finiteness check is scoped to float, since math.isfinite() raises OverflowError for an int too large to convert to one — a validator must not become a source of the error type it exists to prevent. (-inf was already refused as negative.) - c232f84 rejected NaN/Infinity inside ingest_json_records, but every other path still accepted them: is_scalar_or_scalar_array returned True for a non-finite float and for an array containing one, so check_scalar_map emitted nothing, Episode.validate() passed, and node and triple attributes — which arrive via _io.load_rows, not the json_records loader — reached the wire serialized as a bare {"score": NaN}, which no strict JSON parser accepts. The check now lives in the shared validator so every caller benefits, and it names the field and the offending value instead of reusing the generic "must only contain scalar values (string, number, ...)" text, which would be misleading about why a number was refused. A nested or otherwise malformed value still gets that generic message, and no value reports both. Co-Authored-By: Claude <noreply@anthropic.com>
…e paths - include_bots=False dropped a message only when it carried a bot_id, but an incoming-webhook post often has just subtype "bot_message", so that traffic reached the graph despite the filter. Both markers are honored now, and include_bots=True still lets them through. - Duplicate timestamps within a conversation are deduped, which is right for exports merged from several dumps, but the drop was silent. It is now reported with a count, like every other thing this package removes. - The Slack zip was never closed. load() is a generator, so a caller that stops early — preview(limit=...) — abandoned it mid-iteration and left the archive open until collection. Closing happens in a finally. - A .json file holding newline-delimited objects is a routine export shape and the shared row reader already accepts it, but the records loader raised. It now falls back to JSONL when the format was auto-detected; an explicit format="json" still fails, since that is the caller stating the shape. - An empty row file parsed as zero JSONL rows, so the run reported success having submitted nothing. It is refused by name. An explicit [] still means ingesting nothing deliberately. Not changed: the report of thread grouping emitting episodes out of chronological order. Threads are anchored at thread start, which is the order conversations began; a reply's ts is always later than its thread_ts, so the scenario described cannot occur. Co-Authored-By: Claude <noreply@anthropic.com>
…ad of failing the export Ordering and created_at both read ts as a float, but nothing checked it was one: a single non-numeric ts anywhere in an export raised an uncaught ValueError and lost the entire run. thread_ts has the same exposure, since it keys thread grouping. Both are validated when the message is parsed. A message carrying either is skipped and counted, and the count is reported in warnings — the same handling as a duplicate timestamp, and consistent with the package never dropping content in silence. Co-Authored-By: Claude <noreply@anthropic.com>
…st a non-finite one
The finiteness check added to require_nonnegative_number was scoped to
`isinstance(value, float)` to keep math.isfinite() from raising OverflowError
on an int too large to convert to a float. But that scoping avoided the raise
by *accepting* the value, and such an int is exactly as unusable as +inf is:
- SequentialSubmitter(min_interval=10**400) constructed, then submit() raised
OverflowError: timestamp too large to convert to C _PyTime_t from time.sleep.
- search_when_ready(timeout=10**400) was accepted, then raised OverflowError:
int too large to convert to float building its deadline.
- IngestResult.wait(timeout=10**400) never timed out at all, since
monotonic() - start >= 10**400 is never true — the timeout silently stopped
being a timeout, which is the worst of the three.
float("inf") was correctly refused throughout, so the guard was letting through
the one input that produced the failures it exists to prevent. Catching the
OverflowError and treating the value as non-finite fixes all three and lets the
float-only special case go away.
The previous behavior had a test asserting it, whose comment argued the
finiteness check "must not turn a validator into a source of the very error
type it exists to prevent" — true, but inverted for this input: the validator
did not raise, it shipped the OverflowError to time.sleep twenty lines later.
That test now asserts the rejection, and a finite-int case pins that the guard
is finiteness rather than "not an int".
Deliberately not changed: the metadata guards still reject non-finite floats
only. JSON integers are arbitrary-precision, so 10**400 in a metadata value
serializes and reparses exactly and is refused by nothing, whereas NaN and
Infinity have no JSON form at all. The two guards answer different questions —
"can this be written as JSON" versus "can the C clock hold it" — so the
asymmetry is spelled out in both places to keep it from being unified into a
bug later.
Also adds the metadata coverage that was missing: [[nan]] must report the
generic nesting error rather than the non-finite one, since _first_non_finite
only looks one level down and its branch must not shadow a shape error for a
value it cannot see; and [nan, {"a": 1}], which is both non-finite and
otherwise invalid, is named once by the more specific message. One assertion
tightened from "inf" to "is inf;", which -inf no longer satisfies, so a
wrong-sign report fails.
Co-Authored-By: Claude <noreply@anthropic.com>
…tered out Episode metadata carried thread_ts only when a thread grouped more than one message. But a thread can reduce to a single message when its parent is skipped as join or bot noise, and that lone reply still belongs to the thread — it just lost the sibling that made the count exceed one. Message count was standing in for "is this a thread", and the message's own thread_ts answers that directly. Episodes that are part of a thread are now uniformly filterable by it; a genuine standalone message still carries none. Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
Critical
ingestion/src/zep_ingest/submitters/batch.py:199— Abatch.addread timeout or dropped response is recorded as a definite failed page, and submission continues without incrementingitems_in_batch. The server may already have accepted those items, causing later retries to duplicate data and potentially allowing a batch to exceed the 50,000-item limit. Treat ambiguous transport failures as indeterminate: stop submission, preserve the partial result/batch ID, and require reconciliation rather than reporting the page as safely failed. The equivalent thread-batch path iningestion/src/zep_ingest/threads.pyshould receive the same handling.
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 2 potential issues.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit fad72c2. Configure here.
…ng early load() is a generator, and its roster, duplicate-timestamp and invalid-timestamp counts were appended after the yield loop. A consumer that stops early — preview(limit=...) — abandons the generator at a yield, so Python unwinds it with GeneratorExit: the finally clause ran and closed the archive, but nothing after the try block ever did. A preview could sample a message with a corrupt timestamp, skip it, and report nothing. The tallies move into the same finally as the close. On an abandoned generator they describe only the messages actually consumed, which is what a partial preview should say. Co-Authored-By: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
Critical
ingestion/src/zep_ingest/submitters/batch.py:142-160andingestion/src/zep_ingest/threads.py:304-319— Abatch.addtransport error such as a read timeout is treated as proof that no items were added. The server may have accepted the page before the response was lost, yet the code neither incrementsitems_in_batchnor stops using that batch. Subsequent pages can therefore exceed the 50,000-item limit, anditems_submitted/add_errorscan misreport successfully accepted data. For ambiguous transport failures, stop adding to that batch and preserve it for reconciliation, or conservatively count the page toward capacity and explicitly report its outcome as unknown. Apply the same handling to episode and thread-message batches.
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
4 participants
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.
Adds the standalone
zep-ingestPython package with lazy loaders, transforms, preview/validation, batch and sequential submission, retries, and status tracking. Supports Slack exports, documents, email, transcripts, JSON/CSV records, user threads, direct nodes, and fact triples, with safer preflight validation, splitting, metadata limits, and asynchronous task handling. Includes typed packaging metadata, extensive examples and fixtures, and documentation for setup, ontology design, temporal correctness, and recovery. Adds CI across Python 3.11–3.13 and dependency resolutions, release automation, and 341 offline tests with 94% coverage.Note
Medium Risk
Large greenfield surface area and PyPI publish automation, but changes are additive with client-side validation and gated releases; runtime risk is mainly mis-ingestion or API misuse by callers, not core repo auth.
Overview
Adds a new
ingestion/tree shippingzep-ingest(0.1.0): a lazyLoader → Transforms → LimitGuard → Submitterpipeline withpreview()(no API calls), one-liners for Slack exports, documents, email, transcripts, JSON/CSV records, user thread backfill,ingest_fact_triples, andingest_nodes, plus optional LLM contextualization and batch/sequential submission withIngestResulttracking.Automation:
test-ingestion.ymlruns ruff, mypy, pytest (and push-only live integration tests) oningestion/**across Python 3.11–3.13 and dependency resolutions;release-ingestion.ymlpublishes to PyPI only onzep-ingest-v*tags after archiving the release commit, version checks, and build;release-integrations.ymlnow skips those tags so integration releases do not double-fire.Repo docs list
ingestion/in the root README and document the release tag scheme and PyPI trusted-publishing setup in.github/workflows/README.md. The change set also includes runnable examples, sample data, and package docs (README, SETUP, CHANGELOG).Reviewed by Cursor Bugbot for commit 6b2f8e9. Bugbot is set up for automated code reviews on this repo. Configure here.