|
8 | 8 |
|
9 | 9 | from __future__ import annotations |
10 | 10 |
|
| 11 | +import datetime |
11 | 12 | from typing import Any |
12 | 13 |
|
13 | 14 | from .client import get_client |
@@ -41,22 +42,31 @@ def start_root_trace( |
41 | 42 | "model": model, |
42 | 43 | "api_mode": api_mode, |
43 | 44 | } |
44 | | - trace = client.trace( |
45 | | - name=trace_name_from_messages(messages) or "Hermes turn", |
46 | | - project_name=project_name(), |
47 | | - thread_id=session_id or None, |
48 | | - input=trace_input, |
49 | | - metadata=metadata, |
50 | | - tags=tags(), |
51 | | - ) |
| 45 | + # Build the create payload once and keep it: the finalize/eviction re-send |
| 46 | + # (upsert via client.trace(id=...)) must replay this FULL payload, not just |
| 47 | + # output+end_time. An upsert is a whole CreateTraceMessage — any omitted |
| 48 | + # field is sent as null and the backend's last-write-wins merge clobbers the |
| 49 | + # create (name/thread_id -> NA trace). start_time is pinned here so the |
| 50 | + # re-send carries the same value instead of a fresh now() (the SDK defaults |
| 51 | + # start_time to now on every client.trace() call). |
| 52 | + create_kwargs: dict[str, Any] = { |
| 53 | + "name": trace_name_from_messages(messages) or "Hermes turn", |
| 54 | + "project_name": project_name(), |
| 55 | + "thread_id": session_id or None, |
| 56 | + "input": trace_input, |
| 57 | + "metadata": metadata, |
| 58 | + "tags": tags(), |
| 59 | + "start_time": datetime.datetime.now(datetime.timezone.utc), |
| 60 | + } |
| 61 | + trace = client.trace(**create_kwargs) |
52 | 62 | # NOTE: the caller flushes this create (via flush_trace_create) AFTER |
53 | 63 | # releasing the state lock. name/thread_id/input are set only at creation, so |
54 | | - # the create must not coalesce with the turn's later update()+end() in one |
| 64 | + # the create must not coalesce with the turn's later finalize re-send in one |
55 | 65 | # batch window (a fast turn) or the trace lands NA (name=None/thread=None/ |
56 | 66 | # input=null) — the trace-level twin of the span NA-bug. flush() blocks on |
57 | 67 | # the network, so it is deliberately kept out of the lock. |
58 | 68 | debug(f"started trace {trace.id} for {task_key}") |
59 | | - return TraceState(trace=trace, session_id=session_id) |
| 69 | + return TraceState(trace=trace, session_id=session_id, create_kwargs=create_kwargs) |
60 | 70 |
|
61 | 71 |
|
62 | 72 | def flush_trace_create(client: Any) -> None: |
@@ -97,14 +107,27 @@ def finish_trace(task_key: str, *, output: Any = None) -> None: |
97 | 107 | # for calls that never received a post (interrupted turn). They have no |
98 | 108 | # span yet — an in-flight call with no response isn't a meaningful span, |
99 | 109 | # so we simply drop them rather than emit a partial span. |
| 110 | + # |
| 111 | + # Upsert-only finalize: re-send the SAME trace id with the finished |
| 112 | + # payload instead of trace.update()/trace.end(). The SDK's batching layer |
| 113 | + # coalesces this with the create into one final row. An update() shortly |
| 114 | + # after create trips the "may cause data loss" warning; the upsert is the |
| 115 | + # mandated pattern and avoids it. |
| 116 | + # |
| 117 | + # Replay the FULL create payload (name/thread_id/input/start_time/...) |
| 118 | + # plus output+end_time — NOT just output+end_time. An upsert is a whole |
| 119 | + # CreateTraceMessage; omitted fields go as null and the backend's |
| 120 | + # last-write-wins merge would clobber the create, landing an NA trace |
| 121 | + # (name=None/thread_id=None). See TraceState.create_kwargs. |
100 | 122 | final_output = merge_trace_output(output, state) |
101 | | - if final_output is not None: |
102 | | - state.trace.update( |
103 | | - output=final_output |
104 | | - if isinstance(final_output, dict) |
105 | | - else {"content": final_output} |
106 | | - ) |
107 | | - state.trace.end() |
| 123 | + client.trace( |
| 124 | + id=state.trace.id, |
| 125 | + output=final_output |
| 126 | + if final_output is None or isinstance(final_output, dict) |
| 127 | + else {"content": final_output}, |
| 128 | + end_time=datetime.datetime.now(datetime.timezone.utc), |
| 129 | + **state.create_kwargs, |
| 130 | + ) |
108 | 131 | except Exception as exc: # pragma: no cover - fail-open |
109 | 132 | debug(f"finish trace failed: {exc}") |
110 | 133 | finally: |
|
0 commit comments