fix(lifecycle): replay full create payload on finalize upsert

The real-Opik E2E caught an NA-trace regression: the finalize upsert
re-sent only id+output+end_time. But client.trace(id=...) builds a full
CreateTraceMessage — the omitted name/thread_id/input went as null and the
backend's last-write-wins merge clobbered the create, landing an NA trace
(name=None/thread_id=None). start_time also drifted to a fresh now().

The SDK's own Trace.update() docstring documents the fix: re-send the FULL
payload with the same id. So capture the create kwargs on TraceState
(pinning start_time) and replay them + output/end_time on both the finish
and eviction re-sends.

Also made the mock-Opik journal assertion faithful to real Opik
(last-write-wins including nulls) so the cheaper per-PR E2E now reproduces
and catches this NA regression, and added unit guards asserting the
finalize re-send carries name/thread_id/input and the create's start_time.

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

Author: JetoPistola

e2e/assert_journal.py

@@ -53,17 +53,16 @@ def main() -> None:
             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.
+    # trace-batch rows sharing one id. Merge by id, mimicking real Opik's
+    # last-write-wins — a key present in a later row overwrites earlier, INCLUDING
+    # nulls. This reproduces the OPIK-7279 first-attempt NA-trace bug: a finalize
+    # re-send that omits name/thread_id sends them as null and clobbers the
+    # create. The finalize must therefore replay the full create payload.
     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})
+        key = tid if tid is not None else id(t)
+        merged.setdefault(key, {}).update(t)
     merged_traces = list(merged.values())
 
     print(
@@ -84,15 +83,22 @@ def main() -> None:
             "upsert-only (same-id re-send), never trace.update()/end()"
         )
 
-    # 1. A root trace exists and is named (descriptive or the fallback).
+    # 1. A root trace exists and, after the finalize re-send merges in, still
+    # carries name + thread_id (not clobbered to an NA trace) AND the finished
+    # payload (output + end_time). This is the full OPIK-7279 acceptance shape.
     if not merged_traces:
         errors.append("no traces captured")
     else:
-        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("name") for t in merged_traces):
+            errors.append(
+                "no trace has a name after merge (NA trace — finalize re-send "
+                "clobbered it; it must replay the full create payload)"
+            )
+        if not any(t.get("thread_id") for t in merged_traces):
+            errors.append(
+                "no trace has a thread_id after merge (session grouping lost — "
+                "finalize re-send must replay thread_id)"
+            )
         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):

observability/opik/lifecycle.py

@@ -42,22 +42,31 @@ def start_root_trace(
         "model": model,
         "api_mode": api_mode,
     }
-    trace = client.trace(
-        name=trace_name_from_messages(messages) or "Hermes turn",
-        project_name=project_name(),
-        thread_id=session_id or None,
-        input=trace_input,
-        metadata=metadata,
-        tags=tags(),
-    )
+    # Build the create payload once and keep it: the finalize/eviction re-send
+    # (upsert via client.trace(id=...)) must replay this FULL payload, not just
+    # output+end_time. An upsert is a whole CreateTraceMessage — any omitted
+    # field is sent as null and the backend's last-write-wins merge clobbers the
+    # create (name/thread_id -> NA trace). start_time is pinned here so the
+    # re-send carries the same value instead of a fresh now() (the SDK defaults
+    # start_time to now on every client.trace() call).
+    create_kwargs: dict[str, Any] = {
+        "name": trace_name_from_messages(messages) or "Hermes turn",
+        "project_name": project_name(),
+        "thread_id": session_id or None,
+        "input": trace_input,
+        "metadata": metadata,
+        "tags": tags(),
+        "start_time": datetime.datetime.now(datetime.timezone.utc),
+    }
+    trace = client.trace(**create_kwargs)
     # 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 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.
     debug(f"started trace {trace.id} for {task_key}")
-    return TraceState(trace=trace, session_id=session_id)
+    return TraceState(trace=trace, session_id=session_id, create_kwargs=create_kwargs)
 
 
 def flush_trace_create(client: Any) -> None:
@@ -100,18 +109,24 @@ def finish_trace(task_key: str, *, output: Any = None) -> None:
         # 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.
+        # payload 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.
+        #
+        # Replay the FULL create payload (name/thread_id/input/start_time/...)
+        # plus output+end_time — NOT just output+end_time. An upsert is a whole
+        # CreateTraceMessage; omitted fields go as null and the backend's
+        # last-write-wins merge would clobber the create, landing an NA trace
+        # (name=None/thread_id=None). See TraceState.create_kwargs.
         final_output = merge_trace_output(output, state)
         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),
+            **state.create_kwargs,
         )
     except Exception as exc:  # pragma: no cover - fail-open
         debug(f"finish trace failed: {exc}")

observability/opik/state.py

@@ -70,6 +70,14 @@ class PendingGeneration:
 class TraceState:
     trace: Any
     session_id: str = ""
+    # The exact kwargs the trace was CREATED with (name, project_name,
+    # thread_id, input, metadata, tags, start_time). The finalize/eviction
+    # re-send replays them verbatim + output/end_time: an upsert via
+    # client.trace(id=...) is a full CreateTraceMessage, so any field omitted
+    # goes as null and the backend's last-write-wins merge would clobber the
+    # create (name/thread_id -> NA trace). Re-sending the full payload keeps
+    # them. See lifecycle.finish_trace.
+    create_kwargs: Dict[str, Any] = field(default_factory=dict)
     generations: Dict[str, PendingGeneration] = field(default_factory=dict)
     tools: Dict[str, PendingTool] = field(default_factory=dict)
     pending_tools_by_name: Dict[str, list] = field(default_factory=dict)
@@ -141,9 +149,14 @@ def evict_stale_locked() -> None:
         if client is None:
             continue
         try:
+            # Replay the full create payload + end_time (not a bare id+end_time):
+            # an upsert is a whole CreateTraceMessage, so omitting name/thread_id
+            # sends them as null and clobbers the create -> NA trace. See
+            # TraceState.create_kwargs and lifecycle.finish_trace.
             client.trace(
                 id=state.trace.id,
                 end_time=datetime.datetime.now(datetime.timezone.utc),
+                **state.create_kwargs,
             )
         except Exception as exc:  # pragma: no cover - fail-open
             debug(f"evict stale trace failed: {exc}")

tests/test_keying_eviction.py

@@ -48,4 +48,8 @@ def test_lru_eviction_bounds_state_and_finalizes_evicted_trace(plugin):
     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)
+    # The eviction re-send replays the full create payload (name/thread_id), so
+    # the evicted trace isn't clobbered to an NA trace — same fix as finish.
+    assert all(t.finalize_kwargs.get("name") for t in finalized)
+    assert all(t.finalize_kwargs.get("thread_id") == "s" for t in finalized)
     assert not any(e[0] in ("trace.update", "trace.end") for e in plugin._fake.events)

tests/test_lifecycle.py

@@ -141,6 +141,28 @@ def test_post_llm_call_finalizes_and_flushes_trace(plugin):
     )
 
 
+def test_finalize_resend_replays_full_create_payload(plugin):
+    # Regression (OPIK-7279 first attempt): the finalize upsert must replay the
+    # FULL create payload (name/thread_id/input/start_time), not just
+    # output+end_time. An upsert is a whole CreateTraceMessage — a partial
+    # re-send sends the omitted fields as null and the backend's last-write-wins
+    # merge clobbers the create, landing an NA trace (name=None/thread_id=None).
+    # Verified against real Opik: a partial re-send nulled name+thread_id.
+    _run_turn_with_tool(plugin, finalize=True)
+    fk = plugin._fake.traces[0].finalize_kwargs
+    assert fk.get("name"), "finalize re-send must carry name (else NA trace)"
+    assert fk.get("thread_id") == "s", "finalize re-send must carry thread_id"
+    assert fk.get("input") is not None, "finalize re-send must carry input"
+    assert fk.get("start_time") is not None, (
+        "finalize re-send must carry the create's start_time, not a fresh now()"
+    )
+    # And it must be the SAME start_time the trace was created with.
+    ck = plugin._fake.traces[0].create_kwargs
+    assert fk.get("start_time") == ck.get("start_time"), (
+        "finalize start_time must match the create's, not drift to now()"
+    )
+
+
 def test_finalize_is_upsert_not_update_or_end(plugin):
     # Upsert-only: finalize must be a same-id client.trace(...) re-send, never
     # trace.update()/trace.end() (FakeTrace raises on those). The re-send targets