LangfuseBridge leaks root span and session context when stream_result fails

Start with stream_result in burr/core/application.py: call_pre runs before the try block, so an exception from the fast-forward self.run(...) skips call_post. Then read pre_run_execute_call and post_run_execute_call in burr/integrations/langfuse.py, which own the root span and the propagate_attributes context. Run the repro script from the issue; done means the failed call leaves no current span and a later run() emits its run root. Apply the same check to astream_result.

When stream_result raises before it starts streaming (an action fails during the fast-forward, or a bad halt_after name), LangfuseBridge never closes its root span and never exits the Langfuse propagate_attributes context. Later spans in the same context are filed under the failed call's trace, and spans from other OpenTelemetry instrumentation inherit its session.id and user.id.

Current behavior

With LangfuseBridge attached, I call app.stream_result(halt_after=["b"]) on a graph a -> b where a raises. The RuntimeError reaches the caller, but afterwards:

• the current OpenTelemetry span is still the stream_result root, still recording, and it is never exported;
• the next third-party span in that context is parented to that root and carries session.id=app-ff / user.id=user-app-ff from the failed app;
• a run() of a completely different application opens its run root as a child of the failed call's root, so it lands in the failed app's trace.

astream_result behaves the same way. stream_result(halt_after=["nope"]) raises the expected ValueError but leaks the same way, and the same app's next run() then emits no run span at all, because the app id is still registered in _run_call_var and the hooks are skipped.

Since this lives in contextvars, a long-lived thread that serves many requests would keep attributing later traces to the session and user of whichever call failed first. I only checked this within one thread.

Stack Traces

No traceback beyond the original exception. The problem is the state left behind after it.

Steps to replicate behavior

1. pip install "apache-burr==0.43.0" "langfuse==5.0.1" opentelemetry-sdk
2. Save the script below as repro.py and run python repro.py. It needs no Langfuse keys and makes no network calls: it installs Langfuse's own LangfuseSpanProcessor with an InMemorySpanExporter (dummy keys, unused base URL), and passes a Mock() as the bridge's langfuse_client since the bridge only stores it.

"""LangfuseBridge leaves its root span and Langfuse session/user context attached
when stream_result fails before it starts streaming.

No network: Langfuse's own span processor is used with an in-memory exporter
and dummy keys, and the bridge is given a placeholder client.
"""

import asyncio
import contextvars
import logging
from unittest.mock import Mock

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

from langfuse._client.span_processor import LangfuseSpanProcessor

from burr.core import ApplicationBuilder, State, action
from burr.integrations.langfuse import LangfuseBridge

# hide the action error banner Burr logs, the exceptions are printed below
logging.getLogger("burr").setLevel(logging.CRITICAL)

exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(
    LangfuseSpanProcessor(
        public_key="pk-test",
        secret_key="sk-test",
        base_url="http://localhost.invalid",
        span_exporter=exporter,
        should_export_span=lambda span: True,
    )
)
trace.set_tracer_provider(provider)
other_tracer = provider.get_tracer("third.party")


@action(reads=[], writes=["x"])
def a(state: State, fail: bool = False) -> State:
    if fail:
        raise RuntimeError("boom")
    return state.update(x=1)


@action(reads=["x"], writes=["x"])
def b(state: State) -> State:
    return state.update(x=state["x"] + 1)


def make_app(app_id: str):
    return (
        ApplicationBuilder()
        .with_actions(a=a, b=b)
        .with_transitions(("a", "b"))
        .with_entrypoint("a")
        .with_identifiers(app_id=app_id, partition_key=f"user-{app_id}")
        .with_hooks(LangfuseBridge(langfuse_client=Mock()))
        .build()
    )


def dump(title):
    provider.force_flush()
    spans = exporter.get_finished_spans()
    print(f"--- {title}: exported spans")
    for s in spans:
        parent = s.parent.span_id if s.parent else None
        print(
            f"  {s.name:<14} span={s.context.span_id:x} parent="
            f"{'None' if parent is None else format(parent, 'x')} "
            f"session.id={s.attributes.get('session.id')} user.id={s.attributes.get('user.id')}"
        )
    exporter.clear()
    return spans


def check_leak(title, dead):
    dead_root, dead_trace = dead
    with other_tracer.start_as_current_span("third_party_call"):
        pass
    make_app("other-app").run(halt_after=["b"])
    by_name = {s.name: s for s in dump(title)}
    for name, expected_session in (("third_party_call", None), ("run", "other-app")):
        s = by_name[name]
        parent = "None"
        if s.parent is not None:
            parent = "failed call's root" if s.parent.span_id == dead_root else f"{s.parent.span_id:x}"
        same_trace = s.context.trace_id == dead_trace
        print(f"OBSERVED {name}: parent={parent} in failed call's trace={same_trace} "
              f"session.id={s.attributes.get('session.id')} user.id={s.attributes.get('user.id')}")
        print(f"EXPECTED {name}: parent=None in failed call's trace=False "
              f"session.id={expected_session}")
    print()


def report_current(title):
    current = trace.get_current_span()
    print(f"after failed {title}: current span = {getattr(current, 'name', None)!r}, "
          f"still recording={current.is_recording()}")
    ctx = current.get_span_context()
    return ctx.span_id, ctx.trace_id


def case_sync():
    app = make_app("app-ff")
    try:
        app.stream_result(halt_after=["b"], inputs={"fail": True})
    except RuntimeError as e:
        print(f"stream_result raised {e!r}")
    check_leak("stream_result", report_current("stream_result"))


async def case_async():
    app = make_app("app-ff-async")
    try:
        await app.astream_result(halt_after=["b"], inputs={"fail": True})
    except RuntimeError as e:
        print(f"astream_result raised {e!r}")
    check_leak("astream_result", report_current("astream_result"))


def case_unknown_halt():
    app = make_app("app-nope")
    try:
        app.stream_result(halt_after=["nope"])
    except ValueError:
        print("stream_result(halt_after=['nope']) raised ValueError")
    report_current("stream_result(halt_after=['nope'])")
    app.run(halt_after=["b"])
    names = sorted(s.name for s in dump("same app, run() afterwards"))
    print(f"OBSERVED span names: {names}")
    print("EXPECTED span names: ['a', 'b', 'run']")
    print()


if __name__ == "__main__":
    print("== case 1: stream_result, first action raises during the fast-forward")
    contextvars.Context().run(case_sync)
    print("== case 2: astream_result, same app shape")
    asyncio.run(case_async())
    print("== case 3: stream_result with an unknown halt_after name")
    contextvars.Context().run(case_unknown_halt)

Output:

== case 1: stream_result, first action raises during the fast-forward
stream_result raised RuntimeError('boom')
after failed stream_result: current span = 'stream_result', still recording=True
--- stream_result: exported spans
  a              span=ab57643fe94d08c6 parent=9ed9b673daccb555 session.id=app-ff user.id=user-app-ff
  third_party_call span=3f66028f1c8b2005 parent=9ed9b673daccb555 session.id=app-ff user.id=user-app-ff
  a              span=62c2cf041e2fd982 parent=6f3e1423799b5029 session.id=other-app user.id=user-other-app
  b              span=48d756bc277b66a2 parent=6f3e1423799b5029 session.id=other-app user.id=user-other-app
  run            span=6f3e1423799b5029 parent=9ed9b673daccb555 session.id=other-app user.id=user-other-app
OBSERVED third_party_call: parent=failed call's root in failed call's trace=True session.id=app-ff user.id=user-app-ff
EXPECTED third_party_call: parent=None in failed call's trace=False session.id=None
OBSERVED run: parent=failed call's root in failed call's trace=True session.id=other-app user.id=user-other-app
EXPECTED run: parent=None in failed call's trace=False session.id=other-app

== case 2: astream_result, same app shape
astream_result raised RuntimeError('boom')
after failed astream_result: current span = 'stream_result', still recording=True
--- astream_result: exported spans
  a              span=9c5be94245a5b177 parent=94b15ef7b4014b89 session.id=app-ff-async user.id=user-app-ff-async
  third_party_call span=54d6a9e1cff8384b parent=94b15ef7b4014b89 session.id=app-ff-async user.id=user-app-ff-async
  a              span=f06b61848614b3a parent=e875e6b8875b1808 session.id=other-app user.id=user-other-app
  b              span=2923382b22f72ae8 parent=e875e6b8875b1808 session.id=other-app user.id=user-other-app
  run            span=e875e6b8875b1808 parent=94b15ef7b4014b89 session.id=other-app user.id=user-other-app
OBSERVED third_party_call: parent=failed call's root in failed call's trace=True session.id=app-ff-async user.id=user-app-ff-async
EXPECTED third_party_call: parent=None in failed call's trace=False session.id=None
OBSERVED run: parent=failed call's root in failed call's trace=True session.id=other-app user.id=user-other-app
EXPECTED run: parent=None in failed call's trace=False session.id=other-app

== case 3: stream_result with an unknown halt_after name
stream_result(halt_after=['nope']) raised ValueError
after failed stream_result(halt_after=['nope']): current span = 'stream_result', still recording=True
--- same app, run() afterwards: exported spans
  a              span=e897c108f9d646d2 parent=33151d71c9809055 session.id=app-nope user.id=user-app-nope
  b              span=57ccd0f651d39a79 parent=33151d71c9809055 session.id=app-nope user.id=user-app-nope
OBSERVED span names: ['a', 'b']
EXPECTED span names: ['a', 'b', 'run']

Library & System Information

• apache-burr 0.43.0 (code below cited at main f8488d0, the relevant parts are unchanged from 0.43.0)
• langfuse 5.0.1, opentelemetry-sdk 1.45.1
• Python 3.12.3, Linux

Expected behavior

If stream_result / astream_result raise before returning, the bridge should end the root span with the error and detach both the span context and the Langfuse attribute context, the same way a failing run() already does. A later unrelated span should have no parent and no session.id, another application's run should start its own trace, and the same app's next run() should emit its run root.

Additional context

Cause, at main f8488d0:

• burr/integrations/langfuse.py line 309: pre_run_execute_call enters propagate_attributes through _enter_trace_attribute_context (lines 264-272). Only post_run_execute_call (lines 311-328) exits it, and the root span from the OpenTelemetry bridge is likewise only detached and ended in the post hook.
• burr/core/application.py line 1505: stream_result calls call_pre first, then runs _process_control_flow_params, _validate_halt_conditions, the "no next action found" check and the fast-forward self.run(...) at line 1518. None of that is inside the try at line 1542, which is the only path to call_post on an error. The nested run sees the app already in _run_call_var and skips its own hooks, so its exception leaves stream_result with no post hook at all. astream_result has the same layout (acall_pre at line 1758, await self.arun(...) at 1771, try at 1793).

The call ordering in stream_result is older than the Langfuse integration, but before it the worst outcome was an unclosed span. Holding an attached propagate_attributes context across two hooks is what turns it into session and user leakage across applications.

Proposed fix: in stream_result and astream_result, wrap everything between call_pre and the existing try in try/except Exception as e: call_post(self, e); raise (or move call_pre inside the existing try), so the post hook always runs. Independently, the bridge could detach a stale propagation context it finds on the stack in pre_run_execute_call. A regression test would drive a failing fast-forward through stream_result and assert that no span or attribute context is left attached.

I'll open a PR with the fix and a regression test for this.