Swiftide tracing with Future AGI

[dependencies]
swiftide = { version = "=0.33.0", features = ["swiftide-agents", "openai"] }
# Record model, usage, input and output fields on Swiftide's spans. Adds no exporter.
swiftide-agents = { version = "=0.33.0", features = ["langfuse"] }
swiftide-integrations = { version = "=0.33.0", features = ["langfuse"] }
async-openai = { version = "=0.41.3", default-features = false }
anyhow = "1"
schemars = "1"
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
tracing-opentelemetry = "=0.34.0"
opentelemetry = { version = "=0.33.1", default-features = false, features = ["trace"] }
opentelemetry_sdk = { version = "=0.33.1", default-features = false, features = ["trace"] }
opentelemetry-otlp = { version = "=0.33.1", default-features = false, features = ["trace", "http-proto", "reqwest-blocking-client", "reqwest-rustls"] }

# Run on the same host as the agent. The agent sends raw Swiftide fields to this receiver.
receivers:
  otlp:
    protocols:
      http:
        endpoint: 127.0.0.1:4318

processors:
  # Only spans from the swiftide-agent tracer are forwarded. Spans with links are dropped.
  filter/scope:
    error_mode: propagate
    traces:
      span:
        - 'instrumentation_scope.name != "swiftide-agent"'
        - 'Len(span.links) > 0'

  # Map Swiftide's structured fields to Future AGI attributes and drop everything else.
  transform/swiftide:
    error_mode: propagate
    trace_statements:
      # Span types.
      - context: span
        statements:
          - keep_keys(span.attributes, ["langfuse.type", "step_count"])
          - set(span.attributes["fi.span.kind"], "CHAIN")
          - set(span.attributes["fi.span.kind"], "AGENT") where IsMatch(span.name, "^agent\\.")
          - set(span.attributes["fi.span.kind"], "TOOL") where IsMatch(span.name, "^tool\\.")
          - set(span.attributes["fi.span.kind"], "LLM")
            where span.attributes["langfuse.type"] == "GENERATION"
      # Model name and token counts of each model call.
      - context: spanevent
        conditions:
          - span.attributes["fi.span.kind"] == "LLM"
        statements:
          - set(span.attributes["llm.model_name"], spanevent.attributes["langfuse.model"])
            where spanevent.attributes["langfuse.model"] != nil
          - set(spanevent.cache["usage"], ParseJSON(spanevent.attributes["langfuse.usage"]))
            where IsString(spanevent.attributes["langfuse.usage"])
            and spanevent.attributes["langfuse.usage"] != ""
          - set(span.attributes["llm.token_count.prompt"],
              Int(spanevent.cache["usage"]["prompt_tokens"]))
            where spanevent.cache["usage"]["prompt_tokens"] != nil
          - set(span.attributes["llm.token_count.completion"],
              Int(spanevent.cache["usage"]["completion_tokens"]))
            where spanevent.cache["usage"]["completion_tokens"] != nil
          - set(span.attributes["llm.token_count.total"],
              Int(spanevent.cache["usage"]["total_tokens"]))
            where spanevent.cache["usage"]["total_tokens"] != nil
      # Inputs and outputs, only when FI_CAPTURE_CONTENT=true.
      - context: spanevent
        conditions:
          - '"${env:FI_CAPTURE_CONTENT:-false}" == "true"'
        statements:
          - set(span.attributes["input.value"], spanevent.attributes["langfuse.input"])
            where IsString(spanevent.attributes["langfuse.input"])
            and spanevent.attributes["langfuse.input"] != ""
          - set(span.attributes["output.value"], spanevent.attributes["langfuse.output"])
            where IsString(spanevent.attributes["langfuse.output"])
            and spanevent.attributes["langfuse.output"] != ""
      # Keep mapped attributes; clear error status messages and strip backtraces.
      - context: span
        statements:
          - keep_keys(span.attributes, ["fi.span.kind", "step_count", "llm.model_name",
              "llm.token_count.prompt", "llm.token_count.completion",
              "llm.token_count.total", "input.value", "output.value"])
          - replace_pattern(span.attributes["input.value"], "(?s)Stack backtrace:.*", "")
            where span.attributes["input.value"] != nil
          - replace_pattern(span.attributes["output.value"], "(?s)Stack backtrace:.*", "")
            where span.attributes["output.value"] != nil
          - set(span.status.message, "")
          - set(span.trace_state, "")
          - keep_keys(instrumentation_scope.attributes, [])
          - keep_keys(resource.attributes, ["service.name"])
          - set(resource.attributes["project_name"], "${env:FI_PROJECT_NAME}")
          - set(resource.attributes["project_type"], "observe")

  # Swiftide's log events carry message text, so none are forwarded.
  filter/events:
    error_mode: propagate
    traces:
      spanevent:
        - 'true'

  batch:
    timeout: 1s

exporters:
  otlp_http/futureagi:
    traces_endpoint: ${env:FI_TRACES_ENDPOINT:-https://api.futureagi.com/tracer/v1/traces}
    headers:
      X-Api-Key: ${env:FI_API_KEY}
      X-Secret-Key: ${env:FI_SECRET_KEY}
    compression: none

service:
  telemetry:
    logs:
      level: error
    metrics:
      level: none
  pipelines:
    traces:
      receivers: [otlp]
      processors: [filter/scope, transform/swiftide, filter/events, batch]
      exporters: [otlp_http/futureagi]

fn main() -> Result<()> {
    // Raw fields go only to the local Collector, never directly to Future AGI.
    let exporter = opentelemetry_otlp::SpanExporter::builder()
        .with_http()
        .with_protocol(Protocol::HttpBinary)
        .with_endpoint("http://127.0.0.1:4318/v1/traces")
        .build()?;
    let provider = SdkTracerProvider::builder()
        .with_batch_exporter(exporter)
        .with_resource(
            Resource::builder()
                .with_service_name("swiftide-agent")
                .build(),
        )
        .build();
    tracing_subscriber::registry()
        .with(EnvFilter::new(
            "info,swiftide_agents=debug,swiftide_integrations=debug",
        ))
        .with(
            tracing_opentelemetry::layer()
                .with_tracer(provider.tracer("swiftide-agent"))
                .with_location(false)
                .with_threads(false),
        )
        .try_init()?;

    let question = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "What is the weather in Paris?".into());
    let runtime = tokio::runtime::Runtime::new()?;
    let outcome = runtime.block_on(run_agent(question));
    drop(runtime);
    // Flush even when the agent fails; keep shutdown outside the async runtime.
    if let Err(err) = provider.shutdown() {
        eprintln!("tracer shutdown: {err}");
    }
    outcome
}