vanishingradient commited on
Commit
8b908da
·
verified ·
1 Parent(s): 721e233

Fix Langfuse: clean roots + explicit child parents

Browse files
Files changed (1) hide show
  1. src/deepagents_rl/train/swe_env.py +44 -7
src/deepagents_rl/train/swe_env.py CHANGED
@@ -194,6 +194,7 @@ class DeepAgentSWEEnv:
194
  self._root_cm: Any = None
195
  self._root: Any = None
196
  self._trace_id: str = ""
 
197
 
198
  def _langfuse_enabled(self) -> bool:
199
  if os.getenv("DARL_LANGFUSE", "1").strip() in ("0", "false", "no"):
@@ -219,6 +220,17 @@ class DeepAgentSWEEnv:
219
  print(f"[langfuse] init failed: {e}")
220
  return None
221
 
 
 
 
 
 
 
 
 
 
 
 
222
  def _end_trace(self) -> None:
223
  if self._root_cm is not None:
224
  try:
@@ -227,6 +239,8 @@ class DeepAgentSWEEnv:
227
  pass
228
  self._root_cm = None
229
  self._root = None
 
 
230
  if self._lf is not None:
231
  try:
232
  self._lf.flush()
@@ -249,7 +263,8 @@ class DeepAgentSWEEnv:
249
  input=payload,
250
  )
251
  lf = self._lf
252
- if lf is None or self._root is None:
 
253
  try:
254
  out = method(self, *args, **kwargs)
255
  except Exception as e: # noqa: BLE001
@@ -265,12 +280,14 @@ class DeepAgentSWEEnv:
265
  with lf.start_as_current_observation(
266
  name=name,
267
  as_type="tool",
 
268
  input=payload,
269
  metadata={
270
  "rollout_id": self._rollout_id,
271
  "instance_id": self._task.get("instance_id"),
272
  "harness": "deepagents-swe-grpo",
273
  "tool_index": n,
 
274
  },
275
  ) as span:
276
  try:
@@ -348,12 +365,13 @@ class DeepAgentSWEEnv:
348
  _vlog("env.reset.langfuse", enabled=lf is not None)
349
  if lf is not None:
350
  try:
351
- # Fresh root trace per rollout (avoid nesting everything under one parent).
352
- self._trace_id = lf.create_trace_id()
 
 
353
  self._root_cm = lf.start_as_current_observation(
354
  name=f"grpo-swe-rollout:{instance_id}",
355
  as_type="agent",
356
- trace_context={"trace_id": self._trace_id},
357
  input={
358
  "instance_id": instance_id,
359
  "repo": repo,
@@ -381,12 +399,27 @@ class DeepAgentSWEEnv:
381
  end_on_exit=True,
382
  )
383
  self._root = self._root_cm.__enter__()
384
- _vlog("env.reset.langfuse_trace", trace_id=self._trace_id)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
385
  except Exception as e: # noqa: BLE001
386
  _vlog("env.reset.langfuse_failed", error=str(e))
387
  self._root_cm = None
388
  self._root = None
389
  self._trace_id = ""
 
390
 
391
  def _materialize() -> None:
392
  _vlog("env.reset.materialize", repo=repo, base_commit=base_commit[:12])
@@ -402,10 +435,12 @@ class DeepAgentSWEEnv:
402
  )
403
  _vlog("env.reset.sandbox_ready", sandbox=str(self._sandbox))
404
 
405
- if lf is not None and self._root is not None:
 
406
  with lf.start_as_current_observation(
407
  name="repo_materialize",
408
  as_type="span",
 
409
  input={"repo": repo, "base_commit": base_commit},
410
  ) as span:
411
  _materialize()
@@ -688,10 +723,12 @@ class DeepAgentSWEEnv:
688
 
689
  return float(report.get("reward_total") or 0.0)
690
 
691
- if lf is not None and self._root is not None:
 
692
  with lf.start_as_current_observation(
693
  name="reward_grader",
694
  as_type="evaluator",
 
695
  input={
696
  "instance_id": self._task.get("instance_id"),
697
  "n_tool_calls": len(self._tool_calls),
 
194
  self._root_cm: Any = None
195
  self._root: Any = None
196
  self._trace_id: str = ""
197
+ self._root_span_id: str = ""
198
 
199
  def _langfuse_enabled(self) -> bool:
200
  if os.getenv("DARL_LANGFUSE", "1").strip() in ("0", "false", "no"):
 
220
  print(f"[langfuse] init failed: {e}")
221
  return None
222
 
223
+ def _child_trace_context(self) -> dict[str, str] | None:
224
+ """Attach children to THIS rollout's root — never ambient OTEL context.
225
+
226
+ Passing only ``trace_id`` (no parent) makes the Langfuse SDK invent a
227
+ NonRecordingSpan parent id → orphan parents + mega-trace pollution when
228
+ parallel GRPO rollouts share get_client().
229
+ """
230
+ if not self._trace_id or not self._root_span_id:
231
+ return None
232
+ return {"trace_id": self._trace_id, "parent_span_id": self._root_span_id}
233
+
234
  def _end_trace(self) -> None:
235
  if self._root_cm is not None:
236
  try:
 
239
  pass
240
  self._root_cm = None
241
  self._root = None
242
+ self._root_span_id = ""
243
+ # Keep _trace_id until next reset for logging; clear after flush.
244
  if self._lf is not None:
245
  try:
246
  self._lf.flush()
 
263
  input=payload,
264
  )
265
  lf = self._lf
266
+ ctx = self._child_trace_context()
267
+ if lf is None or ctx is None:
268
  try:
269
  out = method(self, *args, **kwargs)
270
  except Exception as e: # noqa: BLE001
 
280
  with lf.start_as_current_observation(
281
  name=name,
282
  as_type="tool",
283
+ trace_context=ctx,
284
  input=payload,
285
  metadata={
286
  "rollout_id": self._rollout_id,
287
  "instance_id": self._task.get("instance_id"),
288
  "harness": "deepagents-swe-grpo",
289
  "tool_index": n,
290
+ "trace_id": self._trace_id,
291
  },
292
  ) as span:
293
  try:
 
365
  _vlog("env.reset.langfuse", enabled=lf is not None)
366
  if lf is not None:
367
  try:
368
+ # IMPORTANT: do NOT pass trace_context={"trace_id": ...} alone.
369
+ # Langfuse SDK then fabricates a NonRecordingSpan parent → orphan
370
+ # parent ids in the UI and children from other rollouts can leak in.
371
+ # A plain start_as_current_observation creates a clean root trace.
372
  self._root_cm = lf.start_as_current_observation(
373
  name=f"grpo-swe-rollout:{instance_id}",
374
  as_type="agent",
 
375
  input={
376
  "instance_id": instance_id,
377
  "repo": repo,
 
399
  end_on_exit=True,
400
  )
401
  self._root = self._root_cm.__enter__()
402
+ self._trace_id = str(
403
+ getattr(self._root, "trace_id", None)
404
+ or getattr(self._root, "traceId", None)
405
+ or ""
406
+ )
407
+ self._root_span_id = str(
408
+ getattr(self._root, "id", None)
409
+ or getattr(self._root, "observation_id", None)
410
+ or ""
411
+ )
412
+ _vlog(
413
+ "env.reset.langfuse_trace",
414
+ trace_id=self._trace_id,
415
+ root_span_id=self._root_span_id,
416
+ )
417
  except Exception as e: # noqa: BLE001
418
  _vlog("env.reset.langfuse_failed", error=str(e))
419
  self._root_cm = None
420
  self._root = None
421
  self._trace_id = ""
422
+ self._root_span_id = ""
423
 
424
  def _materialize() -> None:
425
  _vlog("env.reset.materialize", repo=repo, base_commit=base_commit[:12])
 
435
  )
436
  _vlog("env.reset.sandbox_ready", sandbox=str(self._sandbox))
437
 
438
+ ctx = self._child_trace_context()
439
+ if lf is not None and ctx is not None:
440
  with lf.start_as_current_observation(
441
  name="repo_materialize",
442
  as_type="span",
443
+ trace_context=ctx,
444
  input={"repo": repo, "base_commit": base_commit},
445
  ) as span:
446
  _materialize()
 
723
 
724
  return float(report.get("reward_total") or 0.0)
725
 
726
+ ctx = self._child_trace_context()
727
+ if lf is not None and ctx is not None:
728
  with lf.start_as_current_observation(
729
  name="reward_grader",
730
  as_type="evaluator",
731
+ trace_context=ctx,
732
  input={
733
  "instance_id": self._task.get("instance_id"),
734
  "n_tool_calls": len(self._tool_calls),