[OPIK-7279] [SDK] fix: convert trace lifecycle to upsert-only

QA of opik-hermes 0.1.1 surfaced the Opik SDK warning on every Hermes
message: "Calling Trace.update() shortly after creation with batching
enabled may cause data loss." That warning is the symptom of violating
the codebase-wide upsert-only rule for integrations: when an entity
finishes, re-send the same id with the finished payload rather than
mutating it via update()/end().

- lifecycle.py finish_trace: finalize via a single client.trace(id=...,
  output=..., end_time=...) re-send instead of trace.update()+trace.end().
- state.py evict_stale_locked: finalize evicted traces via the same upsert
  (id + end_time) instead of trace.end(); fail-open if no client.
- Spans were already upsert-compliant and are left untouched.

Tests: FakeOpik.trace() now models real Opik's upsert coalescing; FakeTrace
update()/end() raise so a regression fails loudly. E2E asserts the batching
warning is absent and the finalized trace carries output + end_time.

Implements OPIK-7279.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: JetoPistola

e2e/assert_journal.py

@@ -41,25 +41,62 @@ def _flatten_items(payload):
 
 def main() -> None:
     rows = _load()
-    traces, spans = [], []
+    traces, spans, updates = [], [], []
     for r in rows:
+        kind = r.get("kind")
         items = _flatten_items(r.get("payload"))
-        if r.get("kind") == "traces":
+        if kind == "traces":
             traces += items
-        elif r.get("kind") == "spans":
+        elif kind == "spans":
             spans += items
-
-    print(f"journal: {len(rows)} rows | traces={len(traces)} spans={len(spans)}")
+        elif kind == "update":
+            updates.append(r.get("payload"))
+
+    # Upsert-only lifecycle: the create and the finalize arrive as two SEPARATE
+    # trace-batch rows sharing one id (create carries name/thread/input; finalize
+    # carries output/end_time). Merge by id so the checks below see the coalesced
+    # trace, exactly as real Opik would store it.
+    merged: dict = {}
+    for t in traces:
+        tid = t.get("id")
+        if tid is None:
+            merged.setdefault(id(t), {}).update(t)
+        else:
+            slot = merged.setdefault(tid, {})
+            slot.update({k: v for k, v in t.items() if v is not None})
+    merged_traces = list(merged.values())
+
+    print(
+        f"journal: {len(rows)} rows | trace-batches={len(traces)} "
+        f"merged-traces={len(merged_traces)} spans={len(spans)} updates={len(updates)}"
+    )
 
     errors = []
 
+    # 0. Upsert-only: finalize must NOT go through trace.update() (a PATCH the
+    # mock records as an "update" row). Its presence means the plugin regressed
+    # to the forbidden post-create mutation — the source of the SDK's
+    # "Calling Trace.update() shortly after creation ... may cause data loss"
+    # warning (OPIK-7279).
+    if updates:
+        errors.append(
+            f"{len(updates)} trace.update() PATCH(es) — lifecycle must be "
+            "upsert-only (same-id re-send), never trace.update()/end()"
+        )
+
     # 1. A root trace exists and is named (descriptive or the fallback).
-    if not traces:
+    if not merged_traces:
         errors.append("no traces captured")
     else:
-        names = [t.get("name") for t in traces]
+        names = [t.get("name") for t in merged_traces]
         if not any(n for n in names):
             errors.append(f"trace(s) have no name: {names}")
+        # 1b. The finalize re-send landed: the trace carries output + end_time
+        # (acceptance criteria — the trace still finalizes correctly).
+        if not any(t.get("output") for t in merged_traces):
+            errors.append("no trace carries output (finalize re-send missing)")
+        if not any(t.get("end_time") for t in merged_traces):
+            errors.append("no trace carries end_time (trace not finalized)")
 
     # 2. At least one LLM span and one tool span, all with name+type (no NA).
     llm = [s for s in spans if s.get("type") == "llm"]
@@ -83,7 +120,8 @@ def main() -> None:
         sys.exit(1)
 
     print("\n=== E2E PASSED ===")
-    print(f"  trace: {traces[0].get('name')!r}")
+    named = next((t.get("name") for t in merged_traces if t.get("name")), None)
+    print(f"  trace: {named!r} (finalized via upsert, no trace.update PATCH)")
     print(f"  llm spans: {len(llm)} | tool spans: {len(tool)} | NA: 0")
 
 

e2e/assert_real_opik.py

@@ -58,15 +58,20 @@ def main() -> None:
     # correctly-typed LLM + tool spans with no NA spans.
     errors = []
     # The trace name and thread_id are set only at creation. If the create
-    # message coalesces with the finalize update()/end() in one batch window
-    # (a fast turn), the trace lands with name=None/thread_id=None — an "NA"
-    # trace. The plugin flushes the create to prevent this; assert it held.
+    # message coalesces with the finalize re-send in one batch window (a fast
+    # turn), the trace lands with name=None/thread_id=None — an "NA" trace. The
+    # plugin flushes the create to prevent this; assert it held.
     if not t.get("name"):
         errors.append("trace has no name (NA trace — create/finalize batching race)")
     if not t.get("thread_id"):
         errors.append("trace has no thread_id (session grouping lost to the race)")
     if not t.get("end_time"):
         errors.append("trace not finalized (end_time is null)")
+    # The finalize re-send (upsert: same id + output + end_time) must have
+    # coalesced onto the trace — proves the upsert-only lifecycle landed the
+    # finished payload, not just an open trace (OPIK-7279).
+    if not t.get("output"):
+        errors.append("trace has no output (finalize re-send did not land)")
     if not llm:
         errors.append("no llm spans")
     if not tool:

e2e/run_e2e.sh

@@ -88,14 +88,20 @@ if command -v timeout >/dev/null 2>&1; then TIMEOUT="timeout 180"; else TIMEOUT=
 # No runtime install (the isolated network has no internet). Do NOT run
 # `hermes plugins enable` (it prompts and would hang with no TTY). stdin from
 # /dev/null so any stray prompt gets EOF instead of blocking.
+# Capture the FULL Hermes output (not just the tail) so we can assert the Opik
+# SDK batching warning is absent — the upsert-only lifecycle must not trip it.
+HERMES_LOG="$WORK/hermes.log"
 $TIMEOUT docker run --rm --name e2e-hermes --network "$NET" \
   -e HERMES_UID=0 -e HERMES_GID=0 \
   -v "$HERMES_HOME:/opt/data" \
   "$E2E_IMAGE" \
   sh -c '
     hermes chat -q "Compute 2 to the power 10 and report the number." \
-      --provider openai-api --model gpt-5 2>&1 | tail -20
-  ' < /dev/null || echo "(hermes turn exited non-zero / timed out; assertion judges from the journal)"
+      --provider openai-api --model gpt-5 2>&1
+  ' < /dev/null > "$HERMES_LOG" 2>&1 \
+  || echo "(hermes turn exited non-zero / timed out; assertion judges from the journal)"
+tail -20 "$HERMES_LOG" || true
+cp "$HERMES_LOG" /tmp/opik-e2e-hermes.log 2>/dev/null || true
 
 # Give the SDK background flush a moment to POST to mock-opik.
 sleep 3
@@ -104,5 +110,16 @@ sleep 3
 cp "$JOURNAL_DIR/opik-journal.jsonl" /tmp/opik-e2e-journal.jsonl 2>/dev/null || true
 echo "==> journal saved to /tmp/opik-e2e-journal.jsonl ($(wc -l < "$JOURNAL_DIR/opik-journal.jsonl" 2>/dev/null || echo 0) rows)"
 
+# Assert the "may cause data loss" batching warning is absent from Hermes'
+# output — its presence means the plugin regressed to trace.update()/end()
+# shortly after create (OPIK-7279). Matched loosely so wording drift in the SDK
+# still catches it.
+echo "==> asserting no Opik batching warning in Hermes output"
+if grep -Ei "may cause data loss|Calling Trace\.update\(\) shortly after creation" "$HERMES_LOG"; then
+  echo "=== E2E FAILED: Opik batching warning present (lifecycle not upsert-only) ==="
+  exit 1
+fi
+echo "  (no batching warning — lifecycle is upsert-only)"
+
 echo "==> asserting journal"
 MOCK_OPIK_JOURNAL="$JOURNAL_DIR/opik-journal.jsonl" python3 "$REPO_ROOT/e2e/assert_journal.py"

e2e/run_e2e_real_opik.sh

@@ -103,11 +103,21 @@ docker build -q --build-arg HERMES_IMAGE="$HERMES_IMAGE" \
 
 echo "==> running one Hermes turn (Opik is REAL)"
 if command -v timeout >/dev/null 2>&1; then TIMEOUT="timeout 180"; else TIMEOUT=""; fi
+HERMES_LOG="$WORK/hermes.log"
 $TIMEOUT docker run --rm --name e2e-real-hermes --network "$OPIK_NET" \
   -e HERMES_UID=0 -e HERMES_GID=0 -v "$HERMES_HOME:/opt/data" \
   opik-hermes-e2e:local \
-  sh -c 'hermes chat -q "Compute 2 to the power 10 and report the number." --provider openai-api --model gpt-5 2>&1 | tail -15' \
-  < /dev/null || echo "(hermes turn non-zero/timeout; assertion judges from Opik)"
+  sh -c 'hermes chat -q "Compute 2 to the power 10 and report the number." --provider openai-api --model gpt-5 2>&1' \
+  < /dev/null > "$HERMES_LOG" 2>&1 || echo "(hermes turn non-zero/timeout; assertion judges from Opik)"
+tail -15 "$HERMES_LOG" || true
+
+# Upsert-only lifecycle must not trip the SDK batching warning (OPIK-7279).
+echo "==> asserting no Opik batching warning in Hermes output"
+if grep -Ei "may cause data loss|Calling Trace\.update\(\) shortly after creation" "$HERMES_LOG"; then
+  echo "=== E2E FAILED: Opik batching warning present (lifecycle not upsert-only) ==="
+  exit 1
+fi
+echo "  (no batching warning — lifecycle is upsert-only)"
 
 echo "==> letting the SDK flush, then querying the REAL Opik API"
 sleep 5

observability/opik/hooks.py

@@ -207,7 +207,8 @@ def on_post_llm_call(
     #     (agent/turn_finalizer.py). It carries `assistant_response` but no
     #     api_call_count, so it never matches a PendingGeneration. This is the
     #     only reliable end-of-turn signal — finalize the root trace here,
-    #     otherwise the trace never .end()s and never surfaces as completed.
+    #     otherwise the trace never gets its finalize re-send (output +
+    #     end_time) and never surfaces as completed.
     if pending is None:
         if state is not None and assistant_response is not None:
             finish_trace(task_key, output={"content": safe_value(assistant_response)})

observability/opik/lifecycle.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import datetime
 from typing import Any
 
 from .client import get_client
@@ -51,7 +52,7 @@ def start_root_trace(
     )
     # NOTE: the caller flushes this create (via flush_trace_create) AFTER
     # releasing the state lock. name/thread_id/input are set only at creation, so
-    # the create must not coalesce with the turn's later update()+end() in one
+    # the create must not coalesce with the turn's later finalize re-send in one
     # batch window (a fast turn) or the trace lands NA (name=None/thread=None/
     # input=null) — the trace-level twin of the span NA-bug. flush() blocks on
     # the network, so it is deliberately kept out of the lock.
@@ -97,14 +98,21 @@ def finish_trace(task_key: str, *, output: Any = None) -> None:
         # for calls that never received a post (interrupted turn). They have no
         # span yet — an in-flight call with no response isn't a meaningful span,
         # so we simply drop them rather than emit a partial span.
+        #
+        # Upsert-only finalize: re-send the SAME trace id with the finished
+        # payload (output + end_time) instead of trace.update()/trace.end(). The
+        # SDK's batching layer coalesces this with the create into one final row.
+        # An update() shortly after create trips the "may cause data loss"
+        # warning; the upsert is the mandated pattern and avoids it. name/
+        # thread_id/input were flushed with the create, so they are not re-sent.
         final_output = merge_trace_output(output, state)
-        if final_output is not None:
-            state.trace.update(
-                output=final_output
-                if isinstance(final_output, dict)
-                else {"content": final_output}
-            )
-        state.trace.end()
+        client.trace(
+            id=state.trace.id,
+            output=final_output
+            if final_output is None or isinstance(final_output, dict)
+            else {"content": final_output},
+            end_time=datetime.datetime.now(datetime.timezone.utc),
+        )
     except Exception as exc:  # pragma: no cover - fail-open
         debug(f"finish trace failed: {exc}")
     finally:

observability/opik/state.py

@@ -14,6 +14,7 @@
 from dataclasses import dataclass, field
 from typing import Any, Callable, Dict, Optional, Tuple
 
+from .client import get_client
 from .config import debug
 
 # Hard cap on live trace state. Each turn keys the store by a unique turn_id,
@@ -126,15 +127,23 @@ def evict_stale_locked() -> None:
     entry. Bounds the leak from turns that never reach ``finish_trace``
     (interrupted / tool-only final step / empty final content), whose unique
     per-turn key would otherwise linger forever. The evicted entry's trace is
-    ended so it is not left dangling on the Opik side.
+    finalized via an upsert re-send (same id + end_time) so it is not left
+    dangling on the Opik side — never trace.end(), which is the forbidden
+    post-create-mutation that trips the batching "may cause data loss" warning.
     """
     over = len(store) - (MAX_TRACE_STATE - 1)
     if over <= 0:
         return
+    client = get_client()
     stale = sorted(store.items(), key=lambda kv: kv[1].last_updated_at)[:over]
     for key, state in stale:
         store.pop(key, None)
+        if client is None:
+            continue
         try:
-            state.trace.end()
+            client.trace(
+                id=state.trace.id,
+                end_time=datetime.datetime.now(datetime.timezone.utc),
+            )
         except Exception as exc:  # pragma: no cover - fail-open
             debug(f"evict stale trace failed: {exc}")

tests/conftest.py

@@ -65,7 +65,12 @@ def __init__(self, events: list, trace_id: str, kwargs: dict):
         self._events = events
         self.id = trace_id
         self.create_kwargs = kwargs
-        self.ended = False
+        # A finalize re-send (upsert: same id, output+end_time) merges here.
+        # `finalized` flips when the finish/evict re-send arrives, replacing the
+        # old .end() bool. `updates` collects each finalize re-send's kwargs so
+        # tests can assert the finished payload without a forbidden trace.update.
+        self.finalized = False
+        self.finalize_kwargs: dict = {}
         self.updates: list[dict] = []
         self.spans: list[FakeSpan] = []
 
@@ -75,27 +80,46 @@ def span(self, **kwargs: Any) -> FakeSpan:
         self.spans.append(s)
         return s
 
-    def update(self, **kwargs: Any) -> None:
+    def upsert(self, **kwargs: Any) -> None:
+        """Record a same-id re-send from client.trace(id=...) as a finalize."""
+        self.finalized = True
+        self.finalize_kwargs.update(kwargs)
         self.updates.append(kwargs)
-        self._events.append(("trace.update", sorted(kwargs)))
+        self._events.append(("trace.upsert", sorted(kwargs)))
 
-    def end(self, **kwargs: Any) -> None:
-        self.ended = True
-        self._events.append(("trace.end",))
+    # Kept only so tests can prove the plugin NO LONGER calls these. The
+    # upsert-only lifecycle must never touch trace.update()/trace.end().
+    def update(self, **kwargs: Any) -> None:  # pragma: no cover - must not be called
+        raise AssertionError("trace.update() is forbidden; use an upsert re-send")
+
+    def end(self, **kwargs: Any) -> None:  # pragma: no cover - must not be called
+        raise AssertionError("trace.end() is forbidden; use an upsert re-send")
 
 
 class FakeOpik:
-    """Minimal stand-in for opik.Opik that records the lifecycle."""
+    """Minimal stand-in for opik.Opik that records the lifecycle.
+
+    ``trace()`` models real Opik's upsert coalescing: a call carrying an ``id``
+    that matches an existing trace merges into it (a finalize re-send) instead
+    of minting a new row. A call with no ``id`` (or an unknown one) is a create.
+    """
 
     def __init__(self, **_: Any):
         self.events: list = []
         self.traces: list[FakeTrace] = []
         self.flushed = 0
 
     def trace(self, **kwargs: Any) -> FakeTrace:
-        tid = f"trace-{len(self.traces) + 1}"
+        tid = kwargs.get("id")
+        if tid is not None:
+            existing = next((t for t in self.traces if t.id == tid), None)
+            if existing is not None:
+                self.events.append(("client.trace", kwargs.get("name")))
+                existing.upsert(**kwargs)
+                return existing
+        new_id = tid or f"trace-{len(self.traces) + 1}"
         self.events.append(("client.trace", kwargs.get("name")))
-        t = FakeTrace(self.events, tid, kwargs)
+        t = FakeTrace(self.events, new_id, kwargs)
         self.traces.append(t)
         return t
 

tests/test_extra_hooks.py

@@ -132,9 +132,9 @@ def test_session_end_finalizes_open_traces_and_flushes(plugin):
         session_id="sX",
         turn_id="T1",
     )
-    assert not plugin._fake.traces[0].ended
+    assert not plugin._fake.traces[0].finalized
     plugin.on_session_end(session_id="sX")
-    assert plugin._fake.traces[0].ended
+    assert plugin._fake.traces[0].finalized
     assert plugin._fake.flushed >= 1
 
 
@@ -152,9 +152,9 @@ def test_session_end_finalizes_task_keyed_traces(plugin):
     )
     key = plugin.keys.trace_key("task-42", "sess-y", turn_id="T1")
     assert key.startswith("task:"), "precondition: turn is task-keyed"
-    assert not plugin._fake.traces[0].ended
+    assert not plugin._fake.traces[0].finalized
     plugin.on_session_end(session_id="sess-y")
-    assert plugin._fake.traces[0].ended
+    assert plugin._fake.traces[0].finalized
     assert plugin._fake.flushed >= 1
 
 

tests/test_keying_eviction.py

@@ -29,7 +29,7 @@ def test_concurrent_turns_dont_collide(plugin):
     assert a != b
 
 
-def test_lru_eviction_bounds_state_and_ends_evicted_trace(plugin):
+def test_lru_eviction_bounds_state_and_finalizes_evicted_trace(plugin):
     cap = plugin.state.MAX_TRACE_STATE
     # Open cap+5 distinct turns; each opens a root trace via pre_llm_request.
     for i in range(cap + 5):
@@ -43,6 +43,9 @@ def test_lru_eviction_bounds_state_and_ends_evicted_trace(plugin):
         )
     # State never exceeds the cap...
     assert len(plugin.state.store) <= cap
-    # ...and evicted traces were ended (not left dangling on the Opik side).
-    ended = [t for t in plugin._fake.traces if t.ended]
-    assert len(ended) >= 5
+    # ...and evicted traces were finalized via an upsert re-send (end_time,
+    # same id), not the forbidden trace.end() — not left dangling on Opik.
+    finalized = [t for t in plugin._fake.traces if t.finalized]
+    assert len(finalized) >= 5
+    assert all(t.finalize_kwargs.get("end_time") is not None for t in finalized)
+    assert not any(e[0] in ("trace.update", "trace.end") for e in plugin._fake.events)