From fcba2121abd2ea79b5d04da20db1516ba92662ad Mon Sep 17 00:00:00 2001
From: jackaldenryan <jackaldenryan@gmail.com>
Date: Wed, 12 Aug 2026 15:50:27 -0500
Subject: [PATCH 1/4] fix(ingestion): align zep-ingest with server-assigned
 node/fact UUIDs

Stop accepting client node/fact UUIDs (API rejects/ignores them), surface
Zep-assigned identities on IngestResult in zip-safe order, and bump to 0.2.0.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 ingestion/CHANGELOG.md              |  20 +++-
 ingestion/README.md                 |  50 +++++---
 ingestion/pyproject.toml            |   4 +-
 ingestion/src/zep_ingest/_io.py     |  16 ++-
 ingestion/src/zep_ingest/nodes.py   | 102 ++++++++--------
 ingestion/src/zep_ingest/result.py  |  69 ++++++++++-
 ingestion/src/zep_ingest/triples.py |  22 ++--
 ingestion/tests/conftest.py         |   8 +-
 ingestion/tests/test_nodes.py       | 176 +++++++++++++++++++---------
 ingestion/tests/test_result.py      | 122 +++++++++++++++++++
 ingestion/tests/test_triples.py     |  26 ++++
 11 files changed, 475 insertions(+), 140 deletions(-)

diff --git a/ingestion/CHANGELOG.md b/ingestion/CHANGELOG.md
index 94f3301c..84cce685 100644
--- a/ingestion/CHANGELOG.md
+++ b/ingestion/CHANGELOG.md
@@ -4,7 +4,25 @@ 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.1.0 (unreleased)
+## 0.2.0 (unreleased)
+
+**Breaking:** Zep assigns node and fact UUIDs server-side. Matches the API
+change that rejects caller-supplied node identity and ignores caller-supplied
+fact identity.
+
+- `NodeItem` no longer accepts a client `uuid`, and `ingest_nodes` no longer
+  has `require_uuids`. Zep assigns node identities and returns them on
+  `IngestResult.node_uuids` (parallel to the submitted nodes, with `None` for
+  failed batches; also recovered from completed task params when resuming).
+- `FactTriple` no longer accepts `fact_uuid`; after `wait()`/`refresh()`,
+  assigned fact identities land on `IngestResult.edge_uuids` from task params.
+- `source_node_uuid` / `target_node_uuid` remain caller-supplied pins to
+  existing nodes.
+- JSON row files that still include `uuid` / `fact_uuid` raise a clear
+  ConfigurationError naming the retired field.
+- Requires `zep-cloud>=3.27.0`.
+
+## 0.1.0
 
 First release — everything upstream of the Zep API for getting unstructured and
 structured data into Context Graphs correctly.
diff --git a/ingestion/README.md b/ingestion/README.md
index 8d36ecfa..2c56a0d1 100644
--- a/ingestion/README.md
+++ b/ingestion/README.md
@@ -374,10 +374,14 @@ graph):
 2. Set the ontology before any data flows: `client.graph.set_ontology(...)`,
    scoped with `graph_ids=`/`user_ids=` (or project-wide by omitting both). It is
    not retroactive, and the ingestion package never sets it for you.
-3. Optionally connect fact triples to existing canonical entities by pinning
-   endpoints with `source_node_uuid`/`target_node_uuid`. Extraction dedups
-   against the existing graph, so known entities anchor resolution.
-4. Ingest the corpus with real `created_at` timestamps and alias
+3. Optionally seed canonical entities with `ingest_nodes` and keep
+   `result.node_uuids` (Zep assigns them; do not supply a client UUID). Use
+   `client.graph.node.update` with those UUIDs for later edits — `add_nodes`
+   always creates new nodes.
+4. Optionally connect fact triples to those entities by pinning endpoints with
+   `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(...)`.
 
 ## Fact triples
@@ -385,7 +389,7 @@ graph):
 ```python
 from zep_ingest import FactTriple, ingest_fact_triples
 
-ingest_fact_triples(
+result = ingest_fact_triples(
     client,
     [
         FactTriple(
@@ -395,17 +399,22 @@ ingest_fact_triples(
             source_node_labels=["Person"],  # ties the node to a declared type
             target_node_name="GTM analytics",
             target_node_labels=["Project"],
+            # Optional: pin endpoints to UUIDs from ingest_nodes / a prior read.
+            # source_node_uuid=..., target_node_uuid=...,
             valid_at="2024-06-15T00:00:00Z",
         ),
     ],
     graph_id="org",
 )
+result.wait(timeout=600)
+# Zep assigns the fact UUID; it lands in task params as edge_uuid after completion.
+result.edge_uuids
 ```
 
 Triples skip extraction entirely, so nodes they create are **untyped unless
 you label them** — pass `source_node_labels`/`target_node_labels` (one
 declared entity type each) or the declared ontology never touches a
-triples-only graph.
+triples-only graph. Do not supply a `fact_uuid`; Zep owns fact identity.
 
 Every documented limit (fact ≤250 chars, names ≤50, summaries ≤500,
 SCREAMING_SNAKE_CASE `fact_name`, string attribute and metadata keys whose
@@ -424,21 +433,27 @@ without relationships — `ingest_nodes` adds them directly via
 ```python
 from zep_ingest import NodeItem, ingest_nodes
 
-ingest_nodes(
+result = ingest_nodes(
     client,
     [
-        NodeItem(name="Ana Azimova", label="Person", uuid="…"),
-        NodeItem(name="GTM analytics", label="Project", uuid="…"),
+        NodeItem(name="Ana Azimova", label="Person"),
+        NodeItem(name="GTM analytics", label="Project"),
     ],
     graph_id="org",
 )
+# Zep assigns each node's UUID — keep them for updates and fact-triple pinning.
+result.node_uuids
 ```
 
-Pass a persisted UUIDv4 per node (required by default): it is the node's only
-identity/dedup key, so a re-run upserts instead of duplicating. Up to 100 nodes
-per request, every documented limit (name ≤50, summary ≤500, label ≤100, ≤10
-attributes, ≤10 metadata keys, each value a scalar or an array of scalars)
-validated client-side at construction.
+Zep assigns node UUIDs; do not supply them (a client `uuid` is rejected).
+`result.node_uuids` is parallel to the submitted list — each success carries
+the assigned UUID, and a failed batch leaves `None` in those slots so a later
+success cannot shift under `zip`. Direct `add_nodes` calls create new nodes
+each time (no name or UUID upsert) — use `client.graph.node.update` with a
+returned UUID to rename, replace a summary, edit attributes, or clear an
+entity type. Up to 100 nodes per request, every documented limit (name ≤50,
+summary ≤500, label ≤100, ≤10 attributes, ≤10 metadata keys, each value a
+scalar or an array of scalars) validated client-side at construction.
 Sequential only (the Batch API doesn't take direct nodes).
 
 ## Monitoring a run
@@ -472,9 +487,10 @@ response = search_when_ready(client, "who runs the pilot?", graph_id="g1")
 Partial failures never crash a run: pages/episodes that keep failing are
 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. 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`.
+resume handles; `node_uuids` / `edge_uuids` record identities Zep assigned on
+`ingest_nodes` and completed `ingest_fact_triples` tasks. 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`.
 
 If the API accepts a task-backed submission without returning a completion
 handle, the result reports `status == "untracked"` instead of claiming success.
diff --git a/ingestion/pyproject.toml b/ingestion/pyproject.toml
index cd44fe44..3b470faa 100644
--- a/ingestion/pyproject.toml
+++ b/ingestion/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "zep-ingest"
-version = "0.1.0"
+version = "0.2.0"
 description = "Bulk data ingestion pipeline for Zep: chunk, contextualize, canonicalize, and submit unstructured and structured data"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -18,7 +18,7 @@ classifiers = [
     "Typing :: Typed",
 ]
 dependencies = [
-    "zep-cloud>=3.25.0",
+    "zep-cloud>=3.27.0",
     # Transport errors are classified for retry; zep-cloud's own floor.
     "httpx>=0.21.2",
 ]
diff --git a/ingestion/src/zep_ingest/_io.py b/ingestion/src/zep_ingest/_io.py
index ef9d3ea9..22668121 100644
--- a/ingestion/src/zep_ingest/_io.py
+++ b/ingestion/src/zep_ingest/_io.py
@@ -10,6 +10,7 @@
 """
 
 import json
+from collections.abc import Mapping
 from dataclasses import MISSING, fields
 from pathlib import Path
 from typing import Any
@@ -50,7 +51,12 @@ def load_rows(path: Path) -> list[dict[str, Any]]:
     raise ConfigurationError(f"{path.name} must contain JSON objects, not {type(parsed).__name__}.")
 
 
-def rows_to_fields(rows: list[dict[str, Any]], row_type: type) -> list[dict[str, Any]]:
+def rows_to_fields(
+    rows: list[dict[str, Any]],
+    row_type: type,
+    *,
+    retired_fields: Mapping[str, str] | None = None,
+) -> list[dict[str, Any]]:
     """Validate row shapes against a dataclass and retain supplied fields exactly.
 
     Unknown columns are rejected because silently dropping a misspelled public
@@ -59,6 +65,11 @@ def rows_to_fields(rows: list[dict[str, Any]], row_type: type) -> list[dict[str,
     ``name`` or ``created_at`` names the field and the row rather than surfacing
     as a bare TypeError from the dataclass constructor. Both the allowed and the
     required sets are read off the dataclass, so neither can drift from it.
+
+    ``retired_fields`` maps former public field names to an actionable error
+    message (for example server-owned identity fields that must not be supplied).
+    Those are checked before the generic unknown-field path so the message can
+    name the replacement rather than listing expected columns.
     """
     spec = fields(row_type)
     allowed = frozenset(field.name for field in spec)
@@ -67,10 +78,13 @@ def rows_to_fields(rows: list[dict[str, Any]], row_type: type) -> list[dict[str,
         for field in spec
         if field.default is MISSING and field.default_factory is MISSING
     ]
+    retired = dict(retired_fields or {})
     validated: list[dict[str, Any]] = []
     for index, row in enumerate(rows):
         if not isinstance(row, dict):
             raise ConfigurationError(f"Row {index} must be a JSON object, got {type(row).__name__}")
+        for name in sorted(set(row) & retired.keys()):
+            raise ConfigurationError(f"Row {index}: {retired[name]}")
         unknown = sorted(set(row) - allowed)
         if unknown:
             raise ConfigurationError(
diff --git a/ingestion/src/zep_ingest/nodes.py b/ingestion/src/zep_ingest/nodes.py
index 2dbeea68..a891e783 100644
--- a/ingestion/src/zep_ingest/nodes.py
+++ b/ingestion/src/zep_ingest/nodes.py
@@ -1,7 +1,5 @@
 """Batch node seeding for canonical entities, independent of episode extraction."""
 
-import uuid as uuid_module
-from collections import Counter
 from collections.abc import Iterable
 from dataclasses import dataclass
 from pathlib import Path
@@ -34,9 +32,9 @@
 class NodeItem:
     """One canonical entity node, validated against the batch-node API limits.
 
-    A supplied ``uuid`` is stored in its canonical spelling, so one UUID
-    written two ways — differing case, braces, no hyphens — stays one identity
-    both in the uniqueness check and on the wire.
+    Zep assigns each node's UUID. Capture the returned values from
+    ``IngestResult.node_uuids`` (parallel to the submitted nodes; ``None`` where
+    a batch failed) if you need them for updates or fact-triple pinning.
     """
 
     name: str
@@ -44,7 +42,6 @@ class NodeItem:
     summary: str | None = None
     attributes: dict[str, Any] | None = None
     metadata: dict[str, Any] | None = None
-    uuid: str | None = None
     created_at: str | None = None
 
     def __post_init__(self) -> None:
@@ -55,32 +52,16 @@ def __post_init__(self) -> None:
         check_scalar_map("attributes", self.attributes, errors, max_keys=MAX_ATTRIBUTE_KEYS)
         check_scalar_map("metadata", self.metadata, errors, max_keys=MAX_ATTRIBUTE_KEYS)
         check_timestamp("created_at", self.created_at, errors)
-        if self.uuid is not None:
-            try:
-                parsed = uuid_module.UUID(str(self.uuid))
-                if parsed.version != 4:
-                    errors.append(f"uuid must be UUIDv4 (got version {parsed.version})")
-                # A UUID is a 128-bit value, not the text it was typed as; keep
-                # the canonical spelling so identity comparisons and the
-                # submitted value cannot disagree with each other.
-                self.uuid = str(parsed)
-            except (ValueError, AttributeError, TypeError):
-                errors.append(f"uuid is not a valid UUID: {self.uuid!r}")
         if errors:
             raise ConfigurationError(f"Invalid node {str(self.name)[:40]!r}: " + "; ".join(errors))
 
     def to_add_node_item(self) -> AddNodeItem:
         """Build the SDK request model, omitting unset fields.
 
-        Our ``uuid`` maps to the SDK's ``uuid_`` field, which the client
-        serializes to the wire ``uuid`` key. Only fields that are actually set
-        are passed, so an unset field is omitted from the request rather than
-        sent as ``null`` (which matters for upserts, where a null could clobber
-        an existing value).
+        Only fields that are actually set are passed, so an unset field is
+        omitted from the request rather than sent as ``null``.
         """
         fields: dict[str, Any] = {"name": self.name}
-        if self.uuid is not None:
-            fields["uuid_"] = self.uuid
         if self.label is not None:
             fields["label"] = self.label
         if self.summary is not None:
@@ -94,11 +75,41 @@ def to_add_node_item(self) -> AddNodeItem:
         return AddNodeItem(**fields)
 
 
+_RETIRED_NODE_FIELDS = {
+    "uuid": (
+        "uuid cannot be supplied: Zep assigns node UUIDs "
+        "(use client.graph.node.update with a UUID from IngestResult.node_uuids "
+        "to update an existing node)"
+    ),
+}
+
+
 def _load_nodes(path: Path) -> list[NodeItem]:
-    rows = rows_to_fields(load_rows(path), NodeItem)
+    rows = rows_to_fields(load_rows(path), NodeItem, retired_fields=_RETIRED_NODE_FIELDS)
     return [NodeItem(**row) for row in rows]
 
 
+def _assigned_node_uuids(response: Any, *, expected: int) -> list[str | None]:
+    """Extract Zep-assigned node UUIDs from an ``add_nodes`` response.
+
+    The returned list is always ``expected`` long and aligned with the request
+    batch: a missing entry is ``None`` so callers can zip against the submitted
+    nodes without shifting later identities forward over a gap.
+    """
+    nodes = getattr(response, "nodes", None) or []
+    uuids: list[str | None] = []
+    for index in range(expected):
+        if index >= len(nodes):
+            uuids.append(None)
+            continue
+        node = nodes[index]
+        node_uuid = getattr(node, "uuid_", None)
+        if node_uuid is None and isinstance(node, dict):
+            node_uuid = node.get("uuid") or node.get("uuid_")
+        uuids.append(str(node_uuid) if node_uuid else None)
+    return uuids
+
+
 def ingest_nodes(
     client: Zep,
     nodes: Iterable[NodeItem] | str | Path,
@@ -107,43 +118,23 @@ def ingest_nodes(
     user_id: str | None = None,
     batch_size: int = MAX_NODES_PER_REQUEST,
     max_retries: int = 5,
-    require_uuids: bool = True,
 ) -> IngestResult:
-    """Create/upsert canonical nodes via ``client.graph.add_nodes``.
-
-    UUID is the only safe idempotency key. By default every node must have a
-    persisted UUIDv4; callers must explicitly opt out of that protection.
-    UUID uniqueness is checked by value, not by spelling, before the first
-    network call.
+    """Create canonical nodes via ``client.graph.add_nodes``.
 
-    Submission is asynchronous; bind the result, then wait on it, so the resume
-    handles survive a timeout::
+    Zep assigns each node's UUID. ``result.node_uuids`` is parallel to the
+    submitted node list: successes carry the assigned UUID, and a failed batch
+    (or a missing response entry) leaves ``None`` in those slots so a later
+    success cannot shift forward under ``zip``. Submission is asynchronous; bind
+    the result, then wait on it, so the resume handles survive a timeout::
 
         result = ingest_nodes(client, nodes, graph_id="g1")
         result.wait(timeout=600)
+        # result.node_uuids[i] matches the i-th submitted node (or None)
     """
     destination = Destination(graph_id=graph_id, user_id=user_id)
     if not 1 <= batch_size <= MAX_NODES_PER_REQUEST:
         raise ConfigurationError(f"batch_size must be 1..{MAX_NODES_PER_REQUEST}, got {batch_size}")
     materialized = _load_nodes(Path(nodes)) if isinstance(nodes, str | Path) else list(nodes)
-    missing = [node.name for node in materialized if node.uuid is None]
-    if missing and require_uuids:
-        sample = ", ".join(repr(name) for name in missing[:3])
-        raise ConfigurationError(
-            f"{len(missing)} node(s) have no persisted UUIDv4 ({sample}). "
-            "Set require_uuids=False only for an intentionally non-idempotent ingest."
-        )
-    # NodeItem canonicalizes on construction, so this compares identities, not
-    # spellings: one UUID written two ways is one node, and pinning it twice
-    # would silently overwrite the first node with the second.
-    counts = Counter(node.uuid for node in materialized if node.uuid is not None)
-    duplicates = [node_uuid for node_uuid, count in counts.items() if count > 1]
-    if duplicates:
-        sample = ", ".join(repr(node_uuid) for node_uuid in duplicates[:3])
-        raise ConfigurationError(
-            f"{len(duplicates)} node UUID(s) appear on more than one node ({sample}). "
-            "UUIDs are compared as values, not text; give each node its own UUIDv4."
-        )
 
     scope = (
         {"graph_id": destination.graph_id}
@@ -151,8 +142,6 @@ def ingest_nodes(
         else {"user_id": destination.user_id}
     )
     result = IngestResult(method="sequential", client=client)
-    if missing:
-        result.warnings.append(f"{len(missing)} node(s) have no UUID and may duplicate on a rerun.")
     for start in range(0, len(materialized), batch_size):
         batch = materialized[start : start + batch_size]
         items = [node.to_add_node_item() for node in batch]
@@ -160,6 +149,9 @@ def ingest_nodes(
             lambda: client.graph.add_nodes(nodes=items, **scope),  # noqa: B023
             max_retries=max_retries,
         )
+        # Always extend node_uuids by batch length so indices stay aligned with
+        # the submitted list across partial failures.
+        result._node_uuids_from_submit = True
         if error is not None:
             result.add_errors.append(
                 AddError(
@@ -168,8 +160,10 @@ def ingest_nodes(
                     error=format_api_error("graph.add_nodes", error),
                 )
             )
+            result.node_uuids.extend([None] * len(batch))
             continue
         result.items_submitted += len(batch)
+        result.node_uuids.extend(_assigned_node_uuids(response, expected=len(batch)))
         task_id = getattr(response, "task_id", None)
         if task_id and str(task_id) not in result.task_ids:
             result.task_ids.append(str(task_id))
diff --git a/ingestion/src/zep_ingest/result.py b/ingestion/src/zep_ingest/result.py
index 5fa1e6c8..74bd2755 100644
--- a/ingestion/src/zep_ingest/result.py
+++ b/ingestion/src/zep_ingest/result.py
@@ -3,7 +3,7 @@
 import time
 from collections.abc import Sequence
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 from zep_ingest._validation import require_int_range, require_nonnegative_number
 from zep_ingest.exceptions import IngestFailedError, IngestTimeoutError, IngestUntrackedError
@@ -50,6 +50,28 @@ def _normalize_task_status(status: str | None) -> str:
     return status
 
 
+def _identity_from_task_params(params: Any) -> tuple[list[str], list[str]]:
+    """Pull server-assigned identities out of completed-task params.
+
+    After ``add_nodes`` succeeds the worker merges ``node_uuids`` into params;
+    after ``add_fact_triple`` succeeds it merges ``edge_uuid``. Pending tasks
+    do not carry these keys yet.
+    """
+    if not isinstance(params, dict):
+        return [], []
+    node_uuids: list[str] = []
+    raw_nodes = params.get("node_uuids")
+    if isinstance(raw_nodes, list):
+        node_uuids = [str(item) for item in raw_nodes if item]
+    elif isinstance(raw_nodes, str) and raw_nodes:
+        node_uuids = [raw_nodes]
+    edge_uuids: list[str] = []
+    edge_uuid = params.get("edge_uuid")
+    if edge_uuid:
+        edge_uuids = [str(edge_uuid)]
+    return node_uuids, edge_uuids
+
+
 @dataclass(slots=True)
 class AddError:
     """A submission failure: where it happened, and what the API said about it.
@@ -68,9 +90,16 @@ class IngestResult:
     """Outcome of an ingestion run.
 
     Stateless by design: everything recoverable comes from Batch API statuses or
-    episode/task processing flags; ``batch_ids``/``episode_uuids``/``task_ids`` are the resume
-    handles a caller can persist. ``untracked_items`` records accepted writes for
-    which the API returned no completion handle.
+    episode/task processing flags; ``batch_ids``/``episode_uuids``/``task_ids`` are the
+    resume handles a caller can persist. ``node_uuids`` is parallel to the
+    ``ingest_nodes`` input (assigned UUID or ``None`` for a failed/missing slot)
+    so ``zip`` cannot pin a later success to an earlier failure; when resuming
+    from task IDs only, UUIDs are recovered from completed task params in
+    ``task_ids`` order. ``edge_uuids`` records fact identities from
+    ``add_fact_triple`` task params, as a contiguous prefix of ``task_ids`` so
+    out-of-order completion cannot scramble zip order against the submitted
+    triples. ``untracked_items`` records accepted writes for which the API
+    returned no completion handle.
     """
 
     method: Literal["batch", "sequential"]
@@ -78,6 +107,8 @@ class IngestResult:
     batch_ids: list[str] = field(default_factory=list)
     episode_uuids: list[str] = field(default_factory=list)
     task_ids: list[str] = field(default_factory=list)
+    node_uuids: list[str | None] = field(default_factory=list)
+    edge_uuids: list[str] = field(default_factory=list)
     untracked_items: int = 0
     add_errors: list[AddError] = field(default_factory=list)
     warnings: list[str] = field(default_factory=list)
@@ -87,6 +118,12 @@ class IngestResult:
     )
     _processed_uuids: set[str] = field(default_factory=set, repr=False, compare=False)
     _task_statuses: dict[str, str] = field(default_factory=dict, repr=False, compare=False)
+    # Cached task.params by task_id so identities can be rebuilt in task_ids order
+    # even when a later task becomes terminal before an earlier one.
+    _task_params: dict[str, Any] = field(default_factory=dict, repr=False, compare=False)
+    # True when node_uuids was filled from the add_nodes response (submission order).
+    # Task-param recovery must not overwrite or reorder that list.
+    _node_uuids_from_submit: bool = field(default=False, repr=False, compare=False)
 
     @classmethod
     def from_batch_ids(cls, client: "Zep", batch_ids: "Sequence[str]") -> "IngestResult":
@@ -131,6 +168,30 @@ def refresh(self) -> None:
                     continue
                 task = self.client.task.get(task_id)
                 self._task_statuses[task_id] = _normalize_task_status(task.status)
+                self._task_params[task_id] = getattr(task, "params", None)
+            self._sync_identities_from_task_params()
+
+    def _param_identity_prefix(self, *, kind: Literal["node", "edge"]) -> list[str]:
+        """Identities from cached task params as a contiguous ``task_ids`` prefix.
+
+        Stops at the first task that does not yet expose the identity key, so a
+        later task finishing first cannot surface its UUID ahead of an earlier
+        submission (which would break zip-against-inputs).
+        """
+        collected: list[str] = []
+        for task_id in self.task_ids:
+            nodes, edges = _identity_from_task_params(self._task_params.get(task_id))
+            values = nodes if kind == "node" else edges
+            if not values:
+                break
+            collected.extend(values)
+        return collected
+
+    def _sync_identities_from_task_params(self) -> None:
+        """Rebuild param-sourced identities in ``task_ids`` order after each poll."""
+        self.edge_uuids = self._param_identity_prefix(kind="edge")
+        if not self._node_uuids_from_submit:
+            self.node_uuids = self._param_identity_prefix(kind="node")
 
     @property
     def status(self) -> str:
diff --git a/ingestion/src/zep_ingest/triples.py b/ingestion/src/zep_ingest/triples.py
index 46cbc6cc..2ba849d3 100644
--- a/ingestion/src/zep_ingest/triples.py
+++ b/ingestion/src/zep_ingest/triples.py
@@ -41,9 +41,11 @@ class FactTriple:
     """One fact edge between two named nodes, validated against the API limits.
 
     ``source_node_uuid``/``target_node_uuid`` pin an endpoint to an existing
-    node by identity instead of name resolution — pass them for endpoints with
-    known UUIDs so a re-run cannot resolve a slightly different name to a new
-    node.
+    node by identity instead of name resolution — pass UUIDs from
+    ``IngestResult.node_uuids`` (or another prior read) so a re-run cannot
+    resolve a slightly different name to a new node. Zep assigns the fact's own
+    UUID; it is returned as ``edge_uuid`` in the task params once the task
+    completes and is collected on ``IngestResult.edge_uuids`` after ``wait()``.
     """
 
     fact: str
@@ -56,7 +58,6 @@ class FactTriple:
     target_node_labels: list[str] | None = None
     source_node_uuid: str | None = None
     target_node_uuid: str | None = None
-    fact_uuid: str | None = None
     valid_at: str | None = None
     invalid_at: str | None = None
     created_at: str | None = None
@@ -95,7 +96,7 @@ def __post_init__(self) -> None:
                 )
             elif labels:
                 check_len(f"{field}[0]", labels[0], 100, errors)
-        for field in ("source_node_uuid", "target_node_uuid", "fact_uuid"):
+        for field in ("source_node_uuid", "target_node_uuid"):
             value = getattr(self, field)
             if value is not None:
                 try:
@@ -126,7 +127,6 @@ def to_api_kwargs(self, destination: Destination) -> dict[str, Any]:
             "target_node_labels",
             "source_node_uuid",
             "target_node_uuid",
-            "fact_uuid",
             "valid_at",
             "invalid_at",
             "created_at",
@@ -144,8 +144,16 @@ def to_api_kwargs(self, destination: Destination) -> dict[str, Any]:
         return kwargs
 
 
+_RETIRED_TRIPLE_FIELDS = {
+    "fact_uuid": (
+        "fact_uuid cannot be supplied: Zep assigns fact UUIDs "
+        "(returned as edge_uuid in task params once the task completes)"
+    ),
+}
+
+
 def _load_triples(path: Path) -> list[FactTriple]:
-    rows = rows_to_fields(load_rows(path), FactTriple)
+    rows = rows_to_fields(load_rows(path), FactTriple, retired_fields=_RETIRED_TRIPLE_FIELDS)
     return [FactTriple(**row) for row in rows]
 
 
diff --git a/ingestion/tests/conftest.py b/ingestion/tests/conftest.py
index 836f1d55..524b993b 100644
--- a/ingestion/tests/conftest.py
+++ b/ingestion/tests/conftest.py
@@ -8,6 +8,7 @@
 from zep_cloud.types.add_nodes_response import AddNodesResponse
 from zep_cloud.types.add_thread_messages_response import AddThreadMessagesResponse
 from zep_cloud.types.add_triple_response import AddTripleResponse
+from zep_cloud.types.added_node import AddedNode
 from zep_cloud.types.batch_item_detail import BatchItemDetail
 from zep_cloud.types.batch_item_list_response import BatchItemListResponse
 from zep_cloud.types.batch_progress import BatchProgress
@@ -60,7 +61,12 @@ def mock_zep() -> MagicMock:
     client.graph.get = MagicMock()
     client.graph.set_ontology = MagicMock()
     client.graph.add_fact_triple = MagicMock(return_value=AddTripleResponse(task_id="task-1"))
-    client.graph.add_nodes = MagicMock(return_value=AddNodesResponse(task_id="task-1"))
+    client.graph.add_nodes = MagicMock(
+        return_value=AddNodesResponse(
+            task_id="task-1",
+            nodes=[AddedNode(name="node", uuid_="11111111-1111-4111-8111-111111111111")],
+        )
+    )
     client.graph.node = MagicMock()
     client.graph.edge = MagicMock()
     client.user = MagicMock()
diff --git a/ingestion/tests/test_nodes.py b/ingestion/tests/test_nodes.py
index f320cf69..53c1e0ca 100644
--- a/ingestion/tests/test_nodes.py
+++ b/ingestion/tests/test_nodes.py
@@ -1,16 +1,14 @@
 """Tests for direct canonical-node ingestion."""
 
-import uuid
-
 import pytest
+from zep_cloud.core.api_error import ApiError
 from zep_cloud.types.add_node_item import AddNodeItem
 from zep_cloud.types.add_nodes_response import AddNodesResponse
+from zep_cloud.types.added_node import AddedNode
 
 from zep_ingest.exceptions import ConfigurationError
 from zep_ingest.nodes import NodeItem, ingest_nodes
 
-CANONICAL_UUID = "f6b6bcbe-6b64-4d3f-9f9e-8f6a6f9f0f47"
-
 
 @pytest.mark.parametrize(
     ("field", "value"),
@@ -22,45 +20,17 @@ def test_non_string_node_fields_raise_configuration_error(field, value):
         NodeItem(**kwargs)
 
 
-@pytest.mark.parametrize(
-    "spelling",
-    [
-        "F6B6BCBE-6B64-4D3F-9F9E-8F6A6F9F0F47",
-        "{f6b6bcbe-6b64-4d3f-9f9e-8f6a6f9f0f47}",
-        "f6b6bcbe6b644d3f9f9e8f6a6f9f0f47",
-    ],
-)
-def test_node_uuid_spellings_are_canonicalized(spelling):
-    node = NodeItem(name="Avery Brown", uuid=spelling)
-
-    # One UUID value, one text: what we dedup on is what we submit.
-    assert node.uuid == CANONICAL_UUID
-    assert node.to_add_node_item().uuid_ == CANONICAL_UUID
-
-
-def test_case_different_uuid_spellings_are_rejected_as_duplicates(mock_zep):
-    node_uuid = str(uuid.uuid4())
-    nodes = [
-        NodeItem(name="Avery Brown", uuid=node_uuid),
-        NodeItem(name="Avery B.", uuid=node_uuid.upper()),
-    ]
-
-    # Both spellings are one node, so the second would silently overwrite the
-    # first — the error must name the UUID to be actionable on a large plan.
-    with pytest.raises(ConfigurationError, match=node_uuid):
-        ingest_nodes(mock_zep, nodes, graph_id="g1")
-
-    mock_zep.graph.add_nodes.assert_not_called()
-
-
-def test_node_task_id_is_tracked_as_task(mock_zep):
-    mock_zep.graph.add_nodes.return_value = AddNodesResponse(task_id="node-task-1")
-    node_uuid = str(uuid.uuid4())
-    node = NodeItem(name="Avery Brown", uuid=node_uuid)
+def test_node_task_id_is_tracked_and_assigned_uuids_are_recorded(mock_zep):
+    mock_zep.graph.add_nodes.return_value = AddNodesResponse(
+        task_id="node-task-1",
+        nodes=[AddedNode(name="Avery Brown", uuid_="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa")],
+    )
+    node = NodeItem(name="Avery Brown")
 
     result = ingest_nodes(mock_zep, [node], graph_id="g1")
 
     assert result.task_ids == ["node-task-1"]
+    assert result.node_uuids == ["aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"]
     assert result.batch_ids == []
     assert result.status == "queued"
     # Submitted through the typed SDK method — not a raw transport.
@@ -68,24 +38,126 @@ def test_node_task_id_is_tracked_as_task(mock_zep):
     assert kwargs["graph_id"] == "g1"
     (item,) = kwargs["nodes"]
     assert isinstance(item, AddNodeItem)
-    # Must populate the SDK's uuid_ field (the client serializes it to the wire
-    # "uuid" key); passing the "uuid" alias instead would leave identity unset.
-    assert item.uuid_ == node_uuid
+    assert item.name == "Avery Brown"
+    assert "uuid_" not in item.model_fields_set
+
+
+def test_node_uuids_preserve_batch_submission_order(mock_zep):
+    mock_zep.graph.add_nodes.side_effect = [
+        AddNodesResponse(
+            task_id="task-a",
+            nodes=[
+                AddedNode(name="Avery Brown", uuid_="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"),
+                AddedNode(name="Blake Carter", uuid_="bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"),
+            ],
+        ),
+        AddNodesResponse(
+            task_id="task-b",
+            nodes=[AddedNode(name="Casey Diaz", uuid_="cccccccc-cccc-4ccc-8ccc-cccccccccccc")],
+        ),
+    ]
+    nodes = [
+        NodeItem(name="Avery Brown"),
+        NodeItem(name="Blake Carter"),
+        NodeItem(name="Casey Diaz"),
+    ]
+
+    result = ingest_nodes(mock_zep, nodes, graph_id="g1", batch_size=2)
+
+    assert result.node_uuids == [
+        "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
+        "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
+        "cccccccc-cccc-4ccc-8ccc-cccccccccccc",
+    ]
+    assert result.task_ids == ["task-a", "task-b"]
+    assert result.items_submitted == 3
+
+
+def test_node_uuids_keep_none_gaps_when_earlier_batch_fails(mock_zep):
+    mock_zep.graph.add_nodes.side_effect = [
+        ApiError(status_code=500, body="boom"),
+        AddNodesResponse(
+            task_id="task-b",
+            nodes=[
+                AddedNode(name="Casey Diaz", uuid_="cccccccc-cccc-4ccc-8ccc-cccccccccccc"),
+                AddedNode(name="Drew Ellis", uuid_="dddddddd-dddd-4ddd-8ddd-dddddddddddd"),
+            ],
+        ),
+    ]
+    nodes = [
+        NodeItem(name="Avery Brown"),
+        NodeItem(name="Blake Carter"),
+        NodeItem(name="Casey Diaz"),
+        NodeItem(name="Drew Ellis"),
+    ]
+
+    result = ingest_nodes(mock_zep, nodes, graph_id="g1", batch_size=2)
+
+    assert result.node_uuids == [
+        None,
+        None,
+        "cccccccc-cccc-4ccc-8ccc-cccccccccccc",
+        "dddddddd-dddd-4ddd-8ddd-dddddddddddd",
+    ]
+    assert [node.name for node, _ in zip(nodes, result.node_uuids, strict=True)] == [
+        "Avery Brown",
+        "Blake Carter",
+        "Casey Diaz",
+        "Drew Ellis",
+    ]
+    assert result.items_submitted == 2
+    assert len(result.add_errors) == 1
+    assert result.add_errors[0].index == 0
+    assert result.task_ids == ["task-b"]
+
+
+def test_node_uuids_pad_none_when_response_omits_an_entry(mock_zep):
+    mock_zep.graph.add_nodes.return_value = AddNodesResponse(
+        task_id="task-a",
+        nodes=[AddedNode(name="Avery Brown", uuid_="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa")],
+    )
+    nodes = [NodeItem(name="Avery Brown"), NodeItem(name="Blake Carter")]
+
+    result = ingest_nodes(mock_zep, nodes, graph_id="g1")
+
+    assert result.node_uuids == ["aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", None]
+    assert result.items_submitted == 2
+
+
+def test_client_supplied_uuid_field_is_rejected():
+    with pytest.raises(TypeError, match="uuid"):
+        NodeItem(name="Avery Brown", uuid="f6b6bcbe-6b64-4d3f-9f9e-8f6a6f9f0f47")  # type: ignore[call-arg]
+
+
+def test_json_row_with_uuid_is_rejected_before_any_api_call(mock_zep, tmp_path):
+    path = tmp_path / "nodes.jsonl"
+    path.write_text(
+        '{"name": "Avery Brown", "uuid": "f6b6bcbe-6b64-4d3f-9f9e-8f6a6f9f0f47"}\n',
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ConfigurationError, match="uuid cannot be supplied"):
+        ingest_nodes(mock_zep, path, graph_id="g1")
+
+    mock_zep.graph.add_nodes.assert_not_called()
 
 
 def test_node_submission_without_task_id_is_untracked(mock_zep):
-    mock_zep.graph.add_nodes.return_value = AddNodesResponse()
-    node = NodeItem(name="Avery Brown", uuid=str(uuid.uuid4()))
+    mock_zep.graph.add_nodes.return_value = AddNodesResponse(
+        nodes=[AddedNode(name="Avery Brown", uuid_="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa")]
+    )
+    node = NodeItem(name="Avery Brown")
 
     result = ingest_nodes(mock_zep, [node], graph_id="g1")
 
     assert result.items_submitted == 1
     assert result.untracked_items == 1
+    assert result.node_uuids == ["aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"]
     assert result.status == "untracked"
 
 
 def test_node_wait_polls_until_terminal(mock_zep):
-    node = NodeItem(name="Avery Brown", uuid=str(uuid.uuid4()))
+    node = NodeItem(name="Avery Brown")
 
     result = ingest_nodes(mock_zep, [node], graph_id="g1")
     result.wait(poll_interval=0)
@@ -142,8 +214,8 @@ def test_blank_node_map_keys_raise(field):
 
 def test_non_string_map_key_fails_before_any_api_call(mock_zep):
     def plan():
-        yield NodeItem(name="Avery Brown", uuid=str(uuid.uuid4()))
-        yield NodeItem(name="Blake Carter", uuid=str(uuid.uuid4()), metadata={1: "sales"})
+        yield NodeItem(name="Avery Brown")
+        yield NodeItem(name="Blake Carter", metadata={1: "sales"})
 
     # The API takes JSON object keys, which are strings. A non-string one fails
     # while the plan is still being materialized — nothing reaches the wire, so
@@ -161,8 +233,8 @@ def plan():
 )
 def test_non_finite_map_value_fails_before_any_api_call(mock_zep, field, value):
     def plan():
-        yield NodeItem(name="Avery Brown", uuid=str(uuid.uuid4()))
-        yield NodeItem(name="Blake Carter", uuid=str(uuid.uuid4()), **{field: {"score": value}})
+        yield NodeItem(name="Avery Brown")
+        yield NodeItem(name="Blake Carter", **{field: {"score": value}})
 
     with pytest.raises(ConfigurationError, match="not valid JSON"):
         ingest_nodes(mock_zep, plan(), graph_id="g1")
@@ -176,8 +248,7 @@ def test_non_finite_attribute_from_a_json_file_fails_before_any_api_call(mock_ze
     # failure must name the file's field, not surface as a serialization error.
     path = tmp_path / "nodes.jsonl"
     path.write_text(
-        '{"name": "Avery Brown", "uuid": "' + CANONICAL_UUID + '", '
-        '"attributes": {"score": NaN, "ratios": [1.0, Infinity]}}\n',
+        '{"name": "Avery Brown", "attributes": {"score": NaN, "ratios": [1.0, Infinity]}}\n',
         encoding="utf-8",
     )
 
@@ -191,7 +262,7 @@ def test_non_finite_attribute_from_a_json_file_fails_before_any_api_call(mock_ze
 
 def test_finite_float_attributes_are_accepted(mock_zep):
     # the guard is finiteness, not magnitude
-    node = NodeItem(name="Avery Brown", uuid=CANONICAL_UUID, attributes={"score": 1e308})
+    node = NodeItem(name="Avery Brown", attributes={"score": 1e308})
     ingest_nodes(mock_zep, [node], graph_id="g1")
 
     assert mock_zep.graph.add_nodes.call_args.kwargs["nodes"][0].attributes == {"score": 1e308}
@@ -200,7 +271,6 @@ def test_finite_float_attributes_are_accepted(mock_zep):
 def test_empty_node_maps_are_sent_to_clear_existing_values():
     node = NodeItem(
         name="Avery Brown",
-        uuid=str(uuid.uuid4()),
         attributes={},
         metadata={},
     )
diff --git a/ingestion/tests/test_result.py b/ingestion/tests/test_result.py
index ab59c53a..a1beaf54 100644
--- a/ingestion/tests/test_result.py
+++ b/ingestion/tests/test_result.py
@@ -229,6 +229,128 @@ def test_task_ids_are_polled_until_succeeded(self, mock_zep):
         assert result.status == "succeeded"
         assert mock_zep.task.get.call_count == 2
 
+    def test_refresh_collects_edge_uuid_from_completed_task_params(self, mock_zep):
+        from zep_cloud.types.get_task_response import GetTaskResponse
+
+        result = IngestResult.from_task_ids(mock_zep, ["t1", "t2"])
+        mock_zep.task.get.side_effect = [
+            GetTaskResponse(
+                task_id="t1",
+                status="succeeded",
+                params={"edge_uuid": "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"},
+            ),
+            GetTaskResponse(task_id="t2", status="processing"),
+            GetTaskResponse(
+                task_id="t2",
+                status="succeeded",
+                params={"edge_uuid": "ffffffff-ffff-4fff-8fff-ffffffffffff"},
+            ),
+        ]
+
+        result.refresh()
+        assert result.edge_uuids == ["eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"]
+        assert result.status == "processing"
+
+        result.refresh()
+        assert result.edge_uuids == [
+            "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee",
+            "ffffffff-ffff-4fff-8fff-ffffffffffff",
+        ]
+        assert result.status == "succeeded"
+
+    def test_edge_uuids_preserve_task_ids_order_when_later_task_finishes_first(self, mock_zep):
+        from zep_cloud.types.get_task_response import GetTaskResponse
+
+        result = IngestResult.from_task_ids(mock_zep, ["t1", "t2"])
+        mock_zep.task.get.side_effect = [
+            # First refresh: earlier task still running, later task already done.
+            GetTaskResponse(task_id="t1", status="processing"),
+            GetTaskResponse(
+                task_id="t2",
+                status="succeeded",
+                params={"edge_uuid": "ffffffff-ffff-4fff-8fff-ffffffffffff"},
+            ),
+            # Second refresh: earlier task completes. t2 is terminal and skipped.
+            GetTaskResponse(
+                task_id="t1",
+                status="succeeded",
+                params={"edge_uuid": "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"},
+            ),
+        ]
+
+        result.refresh()
+        # Contiguous prefix only: do not surface t2's UUID ahead of t1.
+        assert result.edge_uuids == []
+        assert result.status == "processing"
+
+        result.refresh()
+        assert result.edge_uuids == [
+            "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee",
+            "ffffffff-ffff-4fff-8fff-ffffffffffff",
+        ]
+        assert result.status == "succeeded"
+
+    def test_refresh_keeps_submit_time_node_uuids_when_task_params_arrive(self, mock_zep):
+        from zep_cloud.types.get_task_response import GetTaskResponse
+
+        # ingest_nodes already recorded response UUIDs in submission order; task
+        # params must not extend or reorder that list on refresh.
+        result = IngestResult(
+            method="sequential",
+            task_ids=["t1"],
+            node_uuids=["aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"],
+            client=mock_zep,
+        )
+        result._node_uuids_from_submit = True
+        mock_zep.task.get.return_value = GetTaskResponse(
+            task_id="t1",
+            status="succeeded",
+            params={
+                "node_uuids": [
+                    "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
+                    "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
+                ]
+            },
+        )
+
+        result.refresh()
+        result.refresh()
+
+        assert result.node_uuids == ["aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"]
+
+    def test_node_uuids_from_task_params_preserve_task_ids_order(self, mock_zep):
+        from zep_cloud.types.get_task_response import GetTaskResponse
+
+        result = IngestResult.from_task_ids(mock_zep, ["t1", "t2"])
+        mock_zep.task.get.side_effect = [
+            GetTaskResponse(task_id="t1", status="processing"),
+            GetTaskResponse(
+                task_id="t2",
+                status="succeeded",
+                params={"node_uuids": ["bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"]},
+            ),
+            GetTaskResponse(
+                task_id="t1",
+                status="succeeded",
+                params={
+                    "node_uuids": [
+                        "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
+                        "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaab",
+                    ]
+                },
+            ),
+        ]
+
+        result.refresh()
+        assert result.node_uuids == []
+
+        result.refresh()
+        assert result.node_uuids == [
+            "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
+            "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaab",
+            "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",
+        ]
+
     def test_failed_task_makes_result_failed(self, mock_zep):
         from zep_cloud.types.get_task_response import GetTaskResponse
 
diff --git a/ingestion/tests/test_triples.py b/ingestion/tests/test_triples.py
index 72f740ee..58a19427 100644
--- a/ingestion/tests/test_triples.py
+++ b/ingestion/tests/test_triples.py
@@ -308,7 +308,33 @@ def test_valid_uuids_accepted_and_mapped(self):
         )
         assert kwargs["source_node_uuid"] == source
         assert kwargs["target_node_uuid"] == target
+        assert "fact_uuid" not in kwargs
 
     def test_invalid_uuid_raises_naming_the_field(self):
         with pytest.raises(ConfigurationError, match="source_node_uuid"):
             triple(source_node_uuid="nope")
+
+    def test_client_supplied_fact_uuid_field_is_rejected(self):
+        with pytest.raises(TypeError, match="fact_uuid"):
+            triple(fact_uuid="f6b6bcbe-6b64-4d3f-9f9e-8f6a6f9f0f47")  # type: ignore[call-arg]
+
+    def test_json_row_with_fact_uuid_is_rejected_before_any_api_call(self, mock_zep, tmp_path):
+        path = tmp_path / "triples.jsonl"
+        path.write_text(
+            json.dumps(
+                {
+                    "fact": "Avery Brown met Blake Carter",
+                    "fact_name": "MET",
+                    "source_node_name": "Avery Brown",
+                    "target_node_name": "Blake Carter",
+                    "fact_uuid": "f6b6bcbe-6b64-4d3f-9f9e-8f6a6f9f0f47",
+                }
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+
+        with pytest.raises(ConfigurationError, match="fact_uuid cannot be supplied"):
+            ingest_fact_triples(mock_zep, path, graph_id="g1")
+
+        mock_zep.graph.add_fact_triple.assert_not_called()

From 0292d2a873902d8e6c11b9a1b40f7ecb25c02673 Mon Sep 17 00:00:00 2001
From: jackaldenryan <jackaldenryan@gmail.com>
Date: Wed, 12 Aug 2026 15:55:06 -0500
Subject: [PATCH 2/4] fix(ingestion): keep later UUIDs after earlier task
 failures

Stop treating terminal tasks without identities like in-flight ones so
edge_uuids/node_uuids still collect later successes with None gaps.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 ingestion/CHANGELOG.md              |  4 +++-
 ingestion/README.md                 |  4 +++-
 ingestion/src/zep_ingest/result.py  | 28 ++++++++++++++--------
 ingestion/src/zep_ingest/triples.py |  4 +++-
 ingestion/tests/test_result.py      | 36 +++++++++++++++++++++++++++++
 5 files changed, 63 insertions(+), 13 deletions(-)

diff --git a/ingestion/CHANGELOG.md b/ingestion/CHANGELOG.md
index 84cce685..f129e2f3 100644
--- a/ingestion/CHANGELOG.md
+++ b/ingestion/CHANGELOG.md
@@ -15,7 +15,9 @@ fact identity.
   `IngestResult.node_uuids` (parallel to the submitted nodes, with `None` for
   failed batches; also recovered from completed task params when resuming).
 - `FactTriple` no longer accepts `fact_uuid`; after `wait()`/`refresh()`,
-  assigned fact identities land on `IngestResult.edge_uuids` from task params.
+  assigned fact identities land on `IngestResult.edge_uuids` from task params
+  (parallel to submitted triples, with `None` for a terminal task that
+  assigned none).
 - `source_node_uuid` / `target_node_uuid` remain caller-supplied pins to
   existing nodes.
 - JSON row files that still include `uuid` / `fact_uuid` raise a clear
diff --git a/ingestion/README.md b/ingestion/README.md
index 2c56a0d1..807ac15d 100644
--- a/ingestion/README.md
+++ b/ingestion/README.md
@@ -408,6 +408,7 @@ result = ingest_fact_triples(
 )
 result.wait(timeout=600)
 # 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
 ```
 
@@ -488,7 +489,8 @@ Partial failures never crash a run: pages/episodes that keep failing are
 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. Task IDs are used by
+`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`.
 
diff --git a/ingestion/src/zep_ingest/result.py b/ingestion/src/zep_ingest/result.py
index 74bd2755..8ccbbb4e 100644
--- a/ingestion/src/zep_ingest/result.py
+++ b/ingestion/src/zep_ingest/result.py
@@ -96,7 +96,8 @@ class IngestResult:
     so ``zip`` cannot pin a later success to an earlier failure; when resuming
     from task IDs only, UUIDs are recovered from completed task params in
     ``task_ids`` order. ``edge_uuids`` records fact identities from
-    ``add_fact_triple`` task params, as a contiguous prefix of ``task_ids`` so
+    ``add_fact_triple`` task params in ``task_ids`` order (``None`` when a
+    terminal task assigned none), stopping only before still-in-flight tasks so
     out-of-order completion cannot scramble zip order against the submitted
     triples. ``untracked_items`` records accepted writes for which the API
     returned no completion handle.
@@ -108,7 +109,7 @@ class IngestResult:
     episode_uuids: list[str] = field(default_factory=list)
     task_ids: list[str] = field(default_factory=list)
     node_uuids: list[str | None] = field(default_factory=list)
-    edge_uuids: list[str] = field(default_factory=list)
+    edge_uuids: list[str | None] = field(default_factory=list)
     untracked_items: int = 0
     add_errors: list[AddError] = field(default_factory=list)
     warnings: list[str] = field(default_factory=list)
@@ -171,20 +172,27 @@ def refresh(self) -> None:
                 self._task_params[task_id] = getattr(task, "params", None)
             self._sync_identities_from_task_params()
 
-    def _param_identity_prefix(self, *, kind: Literal["node", "edge"]) -> list[str]:
-        """Identities from cached task params as a contiguous ``task_ids`` prefix.
+    def _param_identity_prefix(self, *, kind: Literal["node", "edge"]) -> list[str | None]:
+        """Identities from cached task params in ``task_ids`` order.
 
-        Stops at the first task that does not yet expose the identity key, so a
-        later task finishing first cannot surface its UUID ahead of an earlier
-        submission (which would break zip-against-inputs).
+        Stops only at the first still-in-flight task, so a later finish cannot
+        surface its UUID ahead of an earlier submission. A terminal task with no
+        identity (failed/canceled) leaves a ``None`` gap so later successes are
+        still collected and stay zip-aligned with ``task_ids``.
         """
-        collected: list[str] = []
+        collected: list[str | None] = []
         for task_id in self.task_ids:
             nodes, edges = _identity_from_task_params(self._task_params.get(task_id))
             values = nodes if kind == "node" else edges
-            if not values:
+            if values:
+                collected.extend(values)
+                continue
+            status = self._task_statuses.get(task_id)
+            if status not in _TERMINAL_TASK_STATUSES:
+                # Not finished (or not polled yet) — hide later successes for now.
                 break
-            collected.extend(values)
+            # Terminal without an assigned identity: keep a slot for this task.
+            collected.append(None)
         return collected
 
     def _sync_identities_from_task_params(self) -> None:
diff --git a/ingestion/src/zep_ingest/triples.py b/ingestion/src/zep_ingest/triples.py
index 2ba849d3..1a6074d1 100644
--- a/ingestion/src/zep_ingest/triples.py
+++ b/ingestion/src/zep_ingest/triples.py
@@ -45,7 +45,9 @@ class FactTriple:
     ``IngestResult.node_uuids`` (or another prior read) so a re-run cannot
     resolve a slightly different name to a new node. Zep assigns the fact's own
     UUID; it is returned as ``edge_uuid`` in the task params once the task
-    completes and is collected on ``IngestResult.edge_uuids`` after ``wait()``.
+    completes and is collected on ``IngestResult.edge_uuids`` after ``wait()``
+    (parallel to the submitted triples, with ``None`` for a terminal task that
+    assigned none).
     """
 
     fact: str
diff --git a/ingestion/tests/test_result.py b/ingestion/tests/test_result.py
index a1beaf54..9d07bb7c 100644
--- a/ingestion/tests/test_result.py
+++ b/ingestion/tests/test_result.py
@@ -290,6 +290,42 @@ def test_edge_uuids_preserve_task_ids_order_when_later_task_finishes_first(self,
         ]
         assert result.status == "succeeded"
 
+    def test_edge_uuids_keep_later_success_after_earlier_terminal_failure(self, mock_zep):
+        from zep_cloud.types.get_task_response import GetTaskResponse
+
+        result = IngestResult.from_task_ids(mock_zep, ["t1", "t2"])
+        mock_zep.task.get.side_effect = [
+            GetTaskResponse(task_id="t1", status="failed", params={}),
+            GetTaskResponse(
+                task_id="t2",
+                status="succeeded",
+                params={"edge_uuid": "ffffffff-ffff-4fff-8fff-ffffffffffff"},
+            ),
+        ]
+
+        result.refresh()
+
+        assert result.edge_uuids == [None, "ffffffff-ffff-4fff-8fff-ffffffffffff"]
+        assert result.status == "failed"
+
+    def test_node_uuids_keep_later_success_after_earlier_terminal_failure(self, mock_zep):
+        from zep_cloud.types.get_task_response import GetTaskResponse
+
+        result = IngestResult.from_task_ids(mock_zep, ["t1", "t2"])
+        mock_zep.task.get.side_effect = [
+            GetTaskResponse(task_id="t1", status="canceled", params={}),
+            GetTaskResponse(
+                task_id="t2",
+                status="succeeded",
+                params={"node_uuids": ["bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"]},
+            ),
+        ]
+
+        result.refresh()
+
+        assert result.node_uuids == [None, "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"]
+        assert result.status == "canceled"
+
     def test_refresh_keeps_submit_time_node_uuids_when_task_params_arrive(self, mock_zep):
         from zep_cloud.types.get_task_response import GetTaskResponse
 

From ae937ef3cd6f4a22380451b6c282feac862a3e64 Mon Sep 17 00:00:00 2001
From: jackaldenryan <jackaldenryan@gmail.com>
Date: Wed, 12 Aug 2026 15:55:53 -0500
Subject: [PATCH 3/4] docs(ingestion): mark 0.2.0 as released in changelog

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 ingestion/CHANGELOG.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/ingestion/CHANGELOG.md b/ingestion/CHANGELOG.md
index f129e2f3..1e811e3b 100644
--- a/ingestion/CHANGELOG.md
+++ b/ingestion/CHANGELOG.md
@@ -4,7 +4,7 @@ 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.2.0 (unreleased)
+## 0.2.0
 
 **Breaking:** Zep assigns node and fact UUIDs server-side. Matches the API
 change that rejects caller-supplied node identity and ignores caller-supplied

From 8341e00a9781d818f487109e595248d174628a51 Mon Sep 17 00:00:00 2001
From: jackaldenryan <jackaldenryan@gmail.com>
Date: Wed, 12 Aug 2026 16:00:55 -0500
Subject: [PATCH 4/4] test(ingestion): stop pinning package version in smoke
 tests

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 ingestion/tests/test_basic.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/ingestion/tests/test_basic.py b/ingestion/tests/test_basic.py
index 5af95cad..b5e43a46 100644
--- a/ingestion/tests/test_basic.py
+++ b/ingestion/tests/test_basic.py
@@ -1,12 +1,8 @@
-"""Package-level smoke tests: imports, __all__, version."""
+"""Package-level smoke tests: imports and __all__."""
 
 import zep_ingest
 
 
-def test_version():
-    assert zep_ingest.__version__ == "0.1.0"
-
-
 def test_public_api_exports():
     """__all__ is pinned exactly, not as a subset: widening the public surface
     is as much a deliberate, reviewable decision as narrowing it, and everything