perf(providers): parallelize the per-trace Langfuse observation fan-out

[satyaborg]

Problem

Both Langfuse import paths fetch observations with one sequential HTTP request per trace. There is no batching and no concurrency anywhere in src/kensa/providers/langfuse.py.

Legacy path, src/kensa/providers/langfuse.py:235:

observations: list[dict[str, Any]] = []
for trace in traces:
    observations.extend(
        _fetch_legacy_observation_rows(
            client=client,
            endpoint=endpoint,
            trace_id=_langfuse_trace_id(trace),
        )
    )

Observations v2 path, src/kensa/providers/langfuse.py:329:

rows: list[dict[str, Any]] = []
for trace_id in trace_ids:
    rows.extend(
        _parse_observations_v2_io(
            _fetch_observation_rows(
                client=client,
                endpoint=endpoint,
                trace_id=trace_id,
                fields=_OBSERVATIONS_V2_FIELDS,
            )
        )
    )

Each call is itself a paginating loop (_fetch_legacy_observation_rows at :284, _fetch_observation_rows at :382), so a trace with many observations costs several round trips on its own. Every request carries timeout_in_seconds=30 and max_retries=3 (_CLIENT_TIMEOUT_SECONDS, _SDK_MAX_RETRIES).

Cost of the trace listing itself is fine: _fetch_trace_rows pages at 100 per request. The N+1 is entirely in the per-trace observation fan-out.

Impact

Wall-clock time on kensa import scales linearly with trace count, dominated by network latency rather than payload size.

• Default connected import limit is 50 (_LANGFUSE_CONNECTED_DEFAULT_LIMIT, src/kensa/cli.py:176), so a default kensa import --from langfuse is 50+ sequential round trips.
• kensa import --from langfuse --since 7d --limit 200 (the example in docs/cli.mdx:88) is 200+ sequential round trips.

At a realistic 200-400ms per round trip this is minutes of mostly-idle waiting, and a single slow trace stalls the whole import.

Proposed Solution

Fan the per-trace observation fetches out over a bounded concurrent.futures.ThreadPoolExecutor, preserving deterministic output ordering by mapping results back to the input trace order rather than appending as futures complete.

Points to settle:

• Worker cap. Small and fixed (4-8) keeps this well clear of Langfuse rate limits; make it a module constant rather than unbounded.
• Error semantics. Today the first LangfuseProviderError aborts the import. Preserve that, and make sure the raised error is the one from the earliest trace so the message is reproducible across runs.
• Langfuse client thread safety. The SDK client wraps httpx, which is thread-safe for concurrent requests, but confirm before relying on sharing one client across workers.
• Ordering. _fetch_observations_v2_export currently returns rows grouped by trace in trace_ids order; keep that so import output stays byte-stable and the existing manifest/redaction tests do not become order-dependent.

Alternatives Considered

• A server-side batch endpoint accepting multiple trace ids would beat client concurrency, but observations_v1.get_many and observations.get_many both take a single trace_id.
• Fetching observations without a trace filter and grouping client-side avoids the fan-out, but pulls far more data than the --limit asked for.

Additional Context

No concurrency primitives exist in the provider today (grep -n "ThreadPool\|concurrent\|asyncio\|Executor" src/kensa/providers/langfuse.py is empty), so this is additive rather than a rework.