feat(ingestion): submit all sources then poll the last episode (v0.3.0) (#607)

* feat(ingestion): submit all sources then poll the last episode

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>

* chore(ingestion): increase auto wait timeout to 60s per item

Give episodes and manual graph updates more headroom before wait() raises IngestTimeoutError.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(ingestion): align wait() polling with ingestion status docs

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>

* fix(ingestion): correct thread polling and multi-thread wait semantics

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>

* chore(ingestion): add production release smoke script for v0.3.0

Document pre-release prod checks and finalize 0.3.0 changelog notes for thread polling and combine().

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: jackaldenryan

ingestion/CHANGELOG.md

@@ -4,6 +4,36 @@ All notable changes to `zep-ingest` are documented here. The project follows
 [Semantic Versioning](https://semver.org); while at `0.x` the public API may
 still change between minor versions.
 
+## 0.3.0
+
+- **Submit everything, then wait once.** Multiple files or loaders destined for
+  the same graph are submitted together. Sequential vs batch only chooses the
+  submit API (`graph.add` vs Batch API); neither waits for one file to finish
+  processing before the next is sent. If you do not need to block, submit and
+  return — `wait()` stays opt-in.
+- `wait()` aligns with [Check data ingestion status](https://help.getzep.com/check-data-ingestion-status):
+  Batch API paths poll the last batch via `batch.get`; sequential `graph.add`
+  polls the last-submitted episode; sequential `thread.add_messages` polls the
+  last message UUID per thread (regular ``add_messages`` returns message UUIDs,
+  not ``task_id``); nodes/triples poll every task id.
+  Default timeout is `wait_timeout_seconds(items_submitted)` — 60s per item,
+  minimum 120s. Pass `timeout=None` to wait without a deadline.
+  `IngestResult.from_batch_ids(...).wait()` has no item count, so auto timeout
+  does not invent a 120s cap. Do not mix Batch API and sequential `graph.add`
+  on the same graph and expect one `wait()` to cover both.
+- File one-liners and loaders accept a sequence of paths/globs in caller order
+  (`ingest_json_records(client, [issues, prs, jira], graph_id=...)`).
+- `ConcatLoader` concatenates heterogeneous loaders into one submit stream.
+- `IngestResult.combine(...)` merges poll handles for separate submits to the same
+  graph (`batch_ids`, `episode_uuids`, `task_ids`); it does not merge
+  `node_uuids` / `edge_uuids` (zip alignment). Prefer separate `wait()` calls
+  for seeding vs episode ingest.
+- Multi-thread sequential backfills poll the last message UUID **per thread**
+  (threads are independent sagas on the server). Regular `thread.add_messages`
+  returns `message_uuids` only — not `task_id`.
+- Production smoke script: `ingestion/scripts/release_smoke_prod.py` (requires
+  `ZEP_API_KEY`; run via KeyBank `zep-prod`).
+
 ## 0.2.1
 
 - Batch submission now rolls over at 10,000 items by default

ingestion/README.md

@@ -139,7 +139,13 @@ endpoint to call (HTTP 404 — an older server, a self-hosted or Community
 deployment, or a base URL that doesn't route `/batches`). Authorization and
 quota errors are raised as errors instead of quietly downgrading the run.
 Every source here ingests fine without the Batch API — pass
-`method="sequential"` to take that path deliberately.
+`method="sequential"` to take that path deliberately. Sequential does **not**
+mean "wait until this file has finished extracting before submitting the next."
+It only means each item is sent with `graph.add` (or `thread.add_messages`)
+instead of the Batch API. Submit every file into the graph first; `wait()` is
+opt-in. On the default Batch path it monitors `batch.get` on the last batch;
+on sequential fallback it polls the last-submitted episode's `processed` flag
+(see [Check data ingestion status](https://help.getzep.com/check-data-ingestion-status)).
 
 ## The pipeline
 
@@ -166,7 +172,7 @@ pipeline = Pipeline(
 )
 report = pipeline.preview()  # NO Zep API calls: inspect episodes + warnings first
 result = pipeline.run(client, graph_id="company_kb")
-result.wait(timeout=3600)  # submission returns immediately; blocking is opt-in
+result.wait()  # opt-in; batch.get tail or last-submitted episode; timeout scales with item count
 ```
 
 `preview()` shows the transformed episodes and validation warnings (including
@@ -383,7 +389,49 @@ graph):
    `source_node_uuid`/`target_node_uuid`. Extraction dedups against the existing
    graph, so known entities anchor resolution.
 5. Ingest the corpus with real `created_at` timestamps and alias
-   canonicalization, then block on the bound result with `result.wait(...)`.
+   canonicalization. Submit every source for a graph before waiting. Then, if
+   you need extraction to finish, block on the bound result with
+   `result.wait()` (or `combined.wait()` after `IngestResult.combine`).
+
+## Many files, one graph
+
+Do not wait for one file (or one graph) to finish processing before submitting
+the next. Create destinations and set ontology, then enqueue everything.
+
+```python
+from zep_ingest import ConcatLoader, ingest, ingest_json_records, ingest_nodes
+
+# Same loader, several files — submitted in this order, one wait:
+result = ingest_json_records(
+    client,
+    ["data/issues.jsonl", "data/prs.jsonl", "data/jira.jsonl"],
+    graph_id="engineering",
+)
+result.wait()  # batch.get on the last batch (default path); timeout scales with items_submitted
+
+# Mixed sources, still one submit stream:
+from zep_ingest import JsonRecordsLoader, TextFileLoader
+
+result = ingest(
+    client,
+    ConcatLoader(
+        [
+            JsonRecordsLoader(["data/issues.jsonl", "data/prs.jsonl"]),
+            TextFileLoader("data/runbooks/**/*.md"),
+        ]
+    ),
+    graph_id="engineering",
+)
+result.wait()
+
+# Already-separate calls: prefer separate wait() for seeding vs episodes.
+nodes = ingest_nodes(client, node_items, graph_id="engineering")
+nodes.wait()
+docs = ingest_json_records(client, ["data/issues.jsonl", "data/prs.jsonl"], graph_id="engineering")
+docs.wait()
+```
+
+If you are not polling, stop after submit — there is nothing else to do.
 
 ## Fact triples
 
@@ -407,7 +455,7 @@ result = ingest_fact_triples(
     ],
     graph_id="org",
 )
-result.wait(timeout=600)
+result.wait()
 # Zep assigns the fact UUID; it lands in task params as edge_uuid after completion.
 # Parallel to the submitted triples — failed tasks leave None in that slot.
 result.edge_uuids
@@ -463,7 +511,7 @@ Sequential only (the Batch API doesn't take direct nodes).
 ```python
 result = ingest_slack_export(client, "export.zip", graph_id="g1")
 result.status  # queued | processing | untracked | succeeded | partial | failed | canceled
-result.wait(timeout=3600)
+result.wait()
 result.failed_items()  # Batch API item records and/or submission AddErrors
 result.warnings  # everything the pipeline noticed
 result.raise_for_status()  # opt-in strictness
@@ -476,6 +524,20 @@ but when it raises — a timeout, or a submission the API left untracked —
 nothing was ever bound, and `batch_ids` / `task_ids` are the only handles for
 resuming or diagnosing that run.
 
+**What `wait()` polls** (aligned with
+[Check data ingestion status](https://help.getzep.com/check-data-ingestion-status)):
+
+| Submission path | Monitor via `wait()` |
+| --- | --- |
+| Batch API (default for episodes and thread backfill) | Last `batch_id` via `batch.get` |
+| Sequential `graph.add` (batch fallback) | Last-submitted episode (`episode_uuids[-1]`) |
+| Sequential `thread.add_messages` | Last message UUID per thread (`message_uuids`; no `task_id`) |
+| `ingest_nodes` / `ingest_fact_triples` | Every `task_id` (separate queue from episodes) |
+
+Do not mix Batch API and sequential `graph.add` into the same graph and expect
+one `wait()` to cover both. Seed nodes/triples first with their own `wait()`,
+then ingest episodes.
+
 Ingestion is asynchronous — a just-added fact is not instantly retrievable,
 even after `wait()`: search indexing lands a few seconds after processing.
 `search_when_ready` owns that gap so scripts don't hand-roll poll loops:
@@ -491,9 +553,9 @@ recorded as `AddError`s (indices and API messages only — never episode content
 and the run continues. `batch_ids` / `episode_uuids` / `task_ids` are the
 resume handles; `node_uuids` / `edge_uuids` record identities Zep assigned on
 `ingest_nodes` and completed `ingest_fact_triples` tasks (`None` slots mark
-failures so later successes stay zip-aligned). Task IDs are used by
-asynchronous operations such as fact triples, direct node creation, and
-sequential thread submissions, and `wait()` polls them through `client.task`.
+failures so later successes stay zip-aligned). Task IDs come from
+``ingest_nodes``, ``ingest_fact_triples``, and ``add_messages_batch`` — not from
+regular ``thread.add_messages``, which returns message UUIDs instead.
 
 If the API accepts a task-backed submission without returning a completion
 handle, the result reports `status == "untracked"` instead of claiming success.
@@ -565,5 +627,13 @@ make install   # uv sync --extra dev
 make all       # format + lint + type-check + test
 ```
 
+Before a release, run the production smoke script against a throwaway graph
+(requires `ZEP_API_KEY` and `ZEP_API_URL`):
+
+```bash
+keybank run zep-prod -- env ZEP_API_URL=https://api.getzep.com \
+  uv run python scripts/release_smoke_prod.py
+```
+
 Live integration tests run only when `ZEP_API_KEY` is set. See
 [`SETUP.md`](https://github.com/getzep/zep/blob/main/ingestion/SETUP.md) for account setup.

ingestion/examples/documents_example.py

@@ -65,7 +65,7 @@ def main() -> None:
     )
     # Submission returns immediately; blocking is opt-in. Bind the result first
     # so a wait() timeout still leaves you the ids below to resume from.
-    result.wait(timeout=600)
+    result.wait()
     print(f"Submitted {result.items_submitted} chunks via {result.method}: {result.status}")
     for warning in result.warnings:
         print(f"WARNING: {warning}")

ingestion/examples/email_example.py

@@ -49,7 +49,7 @@ def main() -> None:
     )
     # Submission returns immediately; blocking is opt-in. Bind the result first
     # so a wait() timeout still leaves you the ids below to resume from.
-    result.wait(timeout=600)
+    result.wait()
     print(f"Submitted {result.items_submitted} emails via {result.method}: {result.status}")
     for warning in result.warnings:
         print(f"WARNING: {warning}")

ingestion/examples/fact_triples_example.py

@@ -157,7 +157,7 @@ def main() -> None:
     #    already happened at FactTriple construction — before any API call.
     triples = load_triples()
     result = ingest_fact_triples(client, triples, graph_id=graph_id)
-    result.wait(timeout=600)
+    result.wait()
     result.raise_for_status()
     print(f"Molded org_chart.json into {len(triples)} fact triples: {result.status}")
 

ingestion/examples/json_records_example.py

@@ -52,7 +52,7 @@ def main() -> None:
     )
     # Submission returns immediately; blocking is opt-in. Bind the result first
     # so a wait() timeout still leaves you the ids below to resume from.
-    result.wait(timeout=600)
+    result.wait()
     print(f"Submitted {result.items_submitted} records via {result.method}: {result.status}")
     for warning in result.warnings:
         print(f"WARNING: {warning}")

ingestion/examples/slack_export_example.py

@@ -72,7 +72,7 @@ def main() -> None:
     )
     # Submission returns immediately; blocking is opt-in. Bind the result first
     # so a wait() timeout still leaves you the ids below to resume from.
-    result.wait(timeout=600)
+    result.wait()
     print(f"Submitted {result.items_submitted} episodes via {result.method}: {result.status}")
     for warning in result.warnings:
         print(f"WARNING: {warning}")

ingestion/examples/user_graph_example.py

@@ -85,7 +85,7 @@ def main() -> None:
         ),
     ]
     profile_result = ingest_fact_triples(client, profile, user_id=user_id)
-    profile_result.wait(timeout=600)
+    profile_result.wait()
     profile_result.raise_for_status()
     print(f"Seeded {len(profile)} profile facts")
 
@@ -108,7 +108,7 @@ def main() -> None:
         user_id=user_id,
         created_at="2025-06-20T00:00:00Z",  # generated source date
     )
-    docs.wait(timeout=600)
+    docs.wait()
     print(f"Ingested {docs.items_submitted} document chunks: {docs.status}")
 
     # Extraction is asynchronous; wait until facts are searchable, then pull

ingestion/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "zep-ingest"
-version = "0.2.1"
+version = "0.3.0"
 description = "Bulk data ingestion pipeline for Zep: chunk, contextualize, canonicalize, and submit unstructured and structured data"
 readme = "README.md"
 requires-python = ">=3.11"