feat(obs): Langfuse exporter wired into chat completions - #21

Merged
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter
Apr 20, 2026
Merged

feat(obs): Langfuse exporter wired into chat completions#21
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter

Conversation

@moonming

Copy link
Copy Markdown
Member

Summary

New `aisix-obs::langfuse` module that pushes per-chat-completion
generation events to a Langfuse `/api/public/ingestion` endpoint.

There is no first-party Rust SDK for Langfuse, so this is a hand-rolled
HTTP client that:

  • Accepts events on a bounded mpsc channel (4096 cap) — emit is
    non-blocking, drops on overflow.
  • Drains the channel into batches of 50 (or every 1s, whichever fires
    first).
  • POSTs each batch with HTTP basic auth (public_key:secret_key) to
    `{host}/api/public/ingestion`.
  • Logs all upstream errors at WARN; never blocks the request hot path.

Wiring

  • Bootstrap (`aisix-server`): `langfuse::spawn(&cfg.observability)`
    returns `Ok(None)` when disabled, an opaque `LangfuseHandle` when
    enabled. The handle is held for the lifetime of the process.
  • `ProxyState` gains `langfuse: Option<Arc>` plus a
    `with_langfuse(...)` builder.
  • `chat::chat_completions` emits one event per request, success or
    failure — no behavior change when langfuse is `None`.

Tests (20 new)

  • Disabled config returns `None`
  • Enabled-without-host errors with `MissingHost`
  • Enabled-without-key-env errors with the specific env var name
  • Round-trip: wiremock upstream, emit one event, wait for the 1s
    flush, assert exactly one POST received
  • ISO timestamp round-trip including unix epoch + leap year
  • Base64 basic auth encoding matches the expected `pk:sk -> cGs6c2s=`
  • Channel-full does not block the emitter

Test plan

  • `cargo test --workspace` — 395 tests pass (+20 from baseline)
  • `cargo clippy --workspace --all-targets -- -D warnings` clean
  • `cargo fmt --check` clean
  • CI all-green
  • Streaming chat handler + other endpoints (messages/embeddings/
    rerank/audio/images/responses) emit langfuse events — deferred to
    a follow-up PR
    to keep this one reviewable

🤖 Generated with Claude Code

Per spec §3.6 / §9 / plan §4.9. New aisix-obs::langfuse module exports:
- LangfuseEvent / LangfuseSender / LangfuseHandle types
- spawn() that returns Ok(None) when disabled, or starts a background
batch flusher (50 events or 1s, whichever comes first) when enabled
The exporter authenticates with HTTP basic (public_key:secret_key),
POSTs to {host}/api/public/ingestion in the documented batch shape
({batch: [{type: 'generation-create', body: {...}}]}), and never
blocks the request hot path — full queues drop the event silently.
Wired into:
- aisix-server bootstrap (spawn after metrics, hold handle for life
of process)
- ProxyState (new optional langfuse: Option<Arc<LangfuseSender>>)
- chat::chat_completions (emits one event per request, success or
failure)
Tests (20 new):
- Disabled config returns None
- Enabled-without-host errors clearly
- Enabled-without-key-env errors with the missing env name
- Wiremock round-trip: emit one event, wait for the 1s flush
interval, assert the upstream received exactly one POST
- ISO timestamp round-trip including unix epoch + leap year
- Base64 basic auth encoding matches expected output
- Channel-full does not block the emitter
Streaming chat handler emission, plus other endpoints
(messages/embeddings/etc.), follow in a separate PR.
CopilotAI review requested due to automatic review settings April 20, 2026 00:32

CopilotAI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds an optional Langfuse ingestion exporter to the observability layer and wires it into the non-streaming chat completions handler so each request emits a generation event when enabled.

Changes:

  • Introduces aisix-obs::langfuse with a bounded, non-blocking event channel and background batch flusher to Langfuse /api/public/ingestion.
  • Wires a LangfuseSender into ProxyState and emits one LangfuseEvent per /v1/chat/completions request (success or failure).
  • Updates workspace plumbing (exports/deps) to support the new module.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file
FileDescription
crates/aisix-server/src/main.rsSpawns optional Langfuse exporter and injects sender into ProxyState.
crates/aisix-proxy/src/state.rsAdds optional langfuse sender field + builder.
crates/aisix-proxy/src/chat.rsEmits LangfuseEvent on completion for success/failure paths.
crates/aisix-obs/src/lib.rsExposes new langfuse module and re-exports its types.
crates/aisix-obs/src/langfuse.rsImplements batching exporter + tests.
crates/aisix-obs/Cargo.tomlAdds deps for HTTP/JSON/base64/uuid and wiremock for tests.
crates/aisix-core/src/lib.rsRe-exports LangfuseConfig.
Cargo.lockLocks new dependencies pulled in by the exporter/tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

start_time: iso_offset(ev.latency),
end_time: now_iso.clone(),
status_message: (ev.status_code != 200)
.then(|| format!("upstream status {}", ev.status_code)),
Comment on lines +462 to +467
// SAFETY: tests run single-threaded by default and we only mutate
// env vars that are scoped to this test's spawn() call.
unsafe {
std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test");
std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test");
}
Comment on lines +88 to +99
// Optional Langfuse exporter — disabled in config by default.
// When enabled, the proxy gets an Arc<LangfuseSender> through
// ProxyState and emits one event per chat completion at
// end-of-request. We keep the handle alive for the lifetime of
// the process so the background flush task continues running.
let langfuse_handle = match langfuse::spawn(&cfg.observability) {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, "langfuse exporter disabled");
None
}
};

loop {
tokio::select! {
biased;
Comment on lines +31 to +32
/// never blocks the proxy thread; if Langfuse is offline we drop the
/// oldest events at the edges.
@moonming
moonming merged commit 5185b54 into mainApr 20, 2026
10 checks passed
@moonming
moonming deleted the feat/langfuse-exporter branch April 20, 2026 00:44
jarvis9443 added a commit that referenced this pull request Jun 2, 2026
Output guardrails only inspected message.content, so client-visible
output that lives elsewhere bypassed content/DLP checks:
- tool_calls / Anthropic tool_use (normalized into message.extra) are
now folded into a single guardrail-inspected text view via
ChatResponse::guardrail_output_text(), used by the keyword, text-
moderation, Bedrock, and Prompt Shield output checks (#3/#18/#21).
Reasoning/thinking content is intentionally left out of scope.
- Non-streaming cache hits now run the resolved output guardrail chain
before returning the stored body, instead of replaying it unchecked
(#28). Streaming output guardrails already run end-of-stream.
Part of #448 (findings #3, #18, #21, #28)
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@moonming
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Skip to content

feat(obs): Langfuse exporter wired into chat completions - #21

Merged
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter
Apr 20, 2026
Merged

feat(obs): Langfuse exporter wired into chat completions#21
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter

Conversation

@moonming

Copy link
Copy Markdown
Member

Summary

New `aisix-obs::langfuse` module that pushes per-chat-completion
generation events to a Langfuse `/api/public/ingestion` endpoint.

There is no first-party Rust SDK for Langfuse, so this is a hand-rolled
HTTP client that:

  • Accepts events on a bounded mpsc channel (4096 cap) — emit is
    non-blocking, drops on overflow.
  • Drains the channel into batches of 50 (or every 1s, whichever fires
    first).
  • POSTs each batch with HTTP basic auth (public_key:secret_key) to
    `{host}/api/public/ingestion`.
  • Logs all upstream errors at WARN; never blocks the request hot path.

Wiring

  • Bootstrap (`aisix-server`): `langfuse::spawn(&cfg.observability)`
    returns `Ok(None)` when disabled, an opaque `LangfuseHandle` when
    enabled. The handle is held for the lifetime of the process.
  • `ProxyState` gains `langfuse: Option<Arc>` plus a
    `with_langfuse(...)` builder.
  • `chat::chat_completions` emits one event per request, success or
    failure — no behavior change when langfuse is `None`.

Tests (20 new)

  • Disabled config returns `None`
  • Enabled-without-host errors with `MissingHost`
  • Enabled-without-key-env errors with the specific env var name
  • Round-trip: wiremock upstream, emit one event, wait for the 1s
    flush, assert exactly one POST received
  • ISO timestamp round-trip including unix epoch + leap year
  • Base64 basic auth encoding matches the expected `pk:sk -> cGs6c2s=`
  • Channel-full does not block the emitter

Test plan

  • `cargo test --workspace` — 395 tests pass (+20 from baseline)
  • `cargo clippy --workspace --all-targets -- -D warnings` clean
  • `cargo fmt --check` clean
  • CI all-green
  • Streaming chat handler + other endpoints (messages/embeddings/
    rerank/audio/images/responses) emit langfuse events — deferred to
    a follow-up PR
    to keep this one reviewable

🤖 Generated with Claude Code

Per spec §3.6 / §9 / plan §4.9. New aisix-obs::langfuse module exports:
- LangfuseEvent / LangfuseSender / LangfuseHandle types
- spawn() that returns Ok(None) when disabled, or starts a background
batch flusher (50 events or 1s, whichever comes first) when enabled
The exporter authenticates with HTTP basic (public_key:secret_key),
POSTs to {host}/api/public/ingestion in the documented batch shape
({batch: [{type: 'generation-create', body: {...}}]}), and never
blocks the request hot path — full queues drop the event silently.
Wired into:
- aisix-server bootstrap (spawn after metrics, hold handle for life
of process)
- ProxyState (new optional langfuse: Option<Arc<LangfuseSender>>)
- chat::chat_completions (emits one event per request, success or
failure)
Tests (20 new):
- Disabled config returns None
- Enabled-without-host errors clearly
- Enabled-without-key-env errors with the missing env name
- Wiremock round-trip: emit one event, wait for the 1s flush
interval, assert the upstream received exactly one POST
- ISO timestamp round-trip including unix epoch + leap year
- Base64 basic auth encoding matches expected output
- Channel-full does not block the emitter
Streaming chat handler emission, plus other endpoints
(messages/embeddings/etc.), follow in a separate PR.
CopilotAI review requested due to automatic review settings April 20, 2026 00:32

CopilotAI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds an optional Langfuse ingestion exporter to the observability layer and wires it into the non-streaming chat completions handler so each request emits a generation event when enabled.

Changes:

  • Introduces aisix-obs::langfuse with a bounded, non-blocking event channel and background batch flusher to Langfuse /api/public/ingestion.
  • Wires a LangfuseSender into ProxyState and emits one LangfuseEvent per /v1/chat/completions request (success or failure).
  • Updates workspace plumbing (exports/deps) to support the new module.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file
FileDescription
crates/aisix-server/src/main.rsSpawns optional Langfuse exporter and injects sender into ProxyState.
crates/aisix-proxy/src/state.rsAdds optional langfuse sender field + builder.
crates/aisix-proxy/src/chat.rsEmits LangfuseEvent on completion for success/failure paths.
crates/aisix-obs/src/lib.rsExposes new langfuse module and re-exports its types.
crates/aisix-obs/src/langfuse.rsImplements batching exporter + tests.
crates/aisix-obs/Cargo.tomlAdds deps for HTTP/JSON/base64/uuid and wiremock for tests.
crates/aisix-core/src/lib.rsRe-exports LangfuseConfig.
Cargo.lockLocks new dependencies pulled in by the exporter/tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

start_time: iso_offset(ev.latency),
end_time: now_iso.clone(),
status_message: (ev.status_code != 200)
.then(|| format!("upstream status {}", ev.status_code)),
Comment on lines +462 to +467
// SAFETY: tests run single-threaded by default and we only mutate
// env vars that are scoped to this test's spawn() call.
unsafe {
std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test");
std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test");
}
Comment on lines +88 to +99
// Optional Langfuse exporter — disabled in config by default.
// When enabled, the proxy gets an Arc<LangfuseSender> through
// ProxyState and emits one event per chat completion at
// end-of-request. We keep the handle alive for the lifetime of
// the process so the background flush task continues running.
let langfuse_handle = match langfuse::spawn(&cfg.observability) {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, "langfuse exporter disabled");
None
}
};

loop {
tokio::select! {
biased;
Comment on lines +31 to +32
/// never blocks the proxy thread; if Langfuse is offline we drop the
/// oldest events at the edges.
@moonming
moonming merged commit 5185b54 into mainApr 20, 2026
10 checks passed
@moonming
moonming deleted the feat/langfuse-exporter branch April 20, 2026 00:44
jarvis9443 added a commit that referenced this pull request Jun 2, 2026
Output guardrails only inspected message.content, so client-visible
output that lives elsewhere bypassed content/DLP checks:
- tool_calls / Anthropic tool_use (normalized into message.extra) are
now folded into a single guardrail-inspected text view via
ChatResponse::guardrail_output_text(), used by the keyword, text-
moderation, Bedrock, and Prompt Shield output checks (#3/#18/#21).
Reasoning/thinking content is intentionally left out of scope.
- Non-streaming cache hits now run the resolved output guardrail chain
before returning the stored body, instead of replaying it unchecked
(#28). Streaming output guardrails already run end-of-stream.
Part of #448 (findings #3, #18, #21, #28)
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@moonming
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix"); } } catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

feat(obs): Langfuse exporter wired into chat completions - #21

Merged
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter
Apr 20, 2026
Merged

feat(obs): Langfuse exporter wired into chat completions#21
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter

Conversation

@moonming

Copy link
Copy Markdown
Member

Summary

New `aisix-obs::langfuse` module that pushes per-chat-completion
generation events to a Langfuse `/api/public/ingestion` endpoint.

There is no first-party Rust SDK for Langfuse, so this is a hand-rolled
HTTP client that:

  • Accepts events on a bounded mpsc channel (4096 cap) — emit is
    non-blocking, drops on overflow.
  • Drains the channel into batches of 50 (or every 1s, whichever fires
    first).
  • POSTs each batch with HTTP basic auth (public_key:secret_key) to
    `{host}/api/public/ingestion`.
  • Logs all upstream errors at WARN; never blocks the request hot path.

Wiring

  • Bootstrap (`aisix-server`): `langfuse::spawn(&cfg.observability)`
    returns `Ok(None)` when disabled, an opaque `LangfuseHandle` when
    enabled. The handle is held for the lifetime of the process.
  • `ProxyState` gains `langfuse: Option<Arc>` plus a
    `with_langfuse(...)` builder.
  • `chat::chat_completions` emits one event per request, success or
    failure — no behavior change when langfuse is `None`.

Tests (20 new)

  • Disabled config returns `None`
  • Enabled-without-host errors with `MissingHost`
  • Enabled-without-key-env errors with the specific env var name
  • Round-trip: wiremock upstream, emit one event, wait for the 1s
    flush, assert exactly one POST received
  • ISO timestamp round-trip including unix epoch + leap year
  • Base64 basic auth encoding matches the expected `pk:sk -> cGs6c2s=`
  • Channel-full does not block the emitter

Test plan

  • `cargo test --workspace` — 395 tests pass (+20 from baseline)
  • `cargo clippy --workspace --all-targets -- -D warnings` clean
  • `cargo fmt --check` clean
  • CI all-green
  • Streaming chat handler + other endpoints (messages/embeddings/
    rerank/audio/images/responses) emit langfuse events — deferred to
    a follow-up PR
    to keep this one reviewable

🤖 Generated with Claude Code

Per spec §3.6 / §9 / plan §4.9. New aisix-obs::langfuse module exports:
- LangfuseEvent / LangfuseSender / LangfuseHandle types
- spawn() that returns Ok(None) when disabled, or starts a background
batch flusher (50 events or 1s, whichever comes first) when enabled
The exporter authenticates with HTTP basic (public_key:secret_key),
POSTs to {host}/api/public/ingestion in the documented batch shape
({batch: [{type: 'generation-create', body: {...}}]}), and never
blocks the request hot path — full queues drop the event silently.
Wired into:
- aisix-server bootstrap (spawn after metrics, hold handle for life
of process)
- ProxyState (new optional langfuse: Option<Arc<LangfuseSender>>)
- chat::chat_completions (emits one event per request, success or
failure)
Tests (20 new):
- Disabled config returns None
- Enabled-without-host errors clearly
- Enabled-without-key-env errors with the missing env name
- Wiremock round-trip: emit one event, wait for the 1s flush
interval, assert the upstream received exactly one POST
- ISO timestamp round-trip including unix epoch + leap year
- Base64 basic auth encoding matches expected output
- Channel-full does not block the emitter
Streaming chat handler emission, plus other endpoints
(messages/embeddings/etc.), follow in a separate PR.
CopilotAI review requested due to automatic review settings April 20, 2026 00:32

CopilotAI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds an optional Langfuse ingestion exporter to the observability layer and wires it into the non-streaming chat completions handler so each request emits a generation event when enabled.

Changes:

  • Introduces aisix-obs::langfuse with a bounded, non-blocking event channel and background batch flusher to Langfuse /api/public/ingestion.
  • Wires a LangfuseSender into ProxyState and emits one LangfuseEvent per /v1/chat/completions request (success or failure).
  • Updates workspace plumbing (exports/deps) to support the new module.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file
FileDescription
crates/aisix-server/src/main.rsSpawns optional Langfuse exporter and injects sender into ProxyState.
crates/aisix-proxy/src/state.rsAdds optional langfuse sender field + builder.
crates/aisix-proxy/src/chat.rsEmits LangfuseEvent on completion for success/failure paths.
crates/aisix-obs/src/lib.rsExposes new langfuse module and re-exports its types.
crates/aisix-obs/src/langfuse.rsImplements batching exporter + tests.
crates/aisix-obs/Cargo.tomlAdds deps for HTTP/JSON/base64/uuid and wiremock for tests.
crates/aisix-core/src/lib.rsRe-exports LangfuseConfig.
Cargo.lockLocks new dependencies pulled in by the exporter/tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

start_time: iso_offset(ev.latency),
end_time: now_iso.clone(),
status_message: (ev.status_code != 200)
.then(|| format!("upstream status {}", ev.status_code)),
Comment on lines +462 to +467
// SAFETY: tests run single-threaded by default and we only mutate
// env vars that are scoped to this test's spawn() call.
unsafe {
std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test");
std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test");
}
Comment on lines +88 to +99
// Optional Langfuse exporter — disabled in config by default.
// When enabled, the proxy gets an Arc<LangfuseSender> through
// ProxyState and emits one event per chat completion at
// end-of-request. We keep the handle alive for the lifetime of
// the process so the background flush task continues running.
let langfuse_handle = match langfuse::spawn(&cfg.observability) {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, "langfuse exporter disabled");
None
}
};

loop {
tokio::select! {
biased;
Comment on lines +31 to +32
/// never blocks the proxy thread; if Langfuse is offline we drop the
/// oldest events at the edges.
@moonming
moonming merged commit 5185b54 into mainApr 20, 2026
10 checks passed
@moonming
moonming deleted the feat/langfuse-exporter branch April 20, 2026 00:44
jarvis9443 added a commit that referenced this pull request Jun 2, 2026
Output guardrails only inspected message.content, so client-visible
output that lives elsewhere bypassed content/DLP checks:
- tool_calls / Anthropic tool_use (normalized into message.extra) are
now folded into a single guardrail-inspected text view via
ChatResponse::guardrail_output_text(), used by the keyword, text-
moderation, Bedrock, and Prompt Shield output checks (#3/#18/#21).
Reasoning/thinking content is intentionally left out of scope.
- Non-streaming cache hits now run the resolved output guardrail chain
before returning the stored body, instead of replaying it unchecked
(#28). Streaming output guardrails already run end-of-stream.
Part of #448 (findings #3, #18, #21, #28)
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@moonming
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms"); } } catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

feat(obs): Langfuse exporter wired into chat completions - #21

Merged
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter
Apr 20, 2026
Merged

feat(obs): Langfuse exporter wired into chat completions#21
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter

Conversation

@moonming

Copy link
Copy Markdown
Member

Summary

New `aisix-obs::langfuse` module that pushes per-chat-completion
generation events to a Langfuse `/api/public/ingestion` endpoint.

There is no first-party Rust SDK for Langfuse, so this is a hand-rolled
HTTP client that:

  • Accepts events on a bounded mpsc channel (4096 cap) — emit is
    non-blocking, drops on overflow.
  • Drains the channel into batches of 50 (or every 1s, whichever fires
    first).
  • POSTs each batch with HTTP basic auth (public_key:secret_key) to
    `{host}/api/public/ingestion`.
  • Logs all upstream errors at WARN; never blocks the request hot path.

Wiring

  • Bootstrap (`aisix-server`): `langfuse::spawn(&cfg.observability)`
    returns `Ok(None)` when disabled, an opaque `LangfuseHandle` when
    enabled. The handle is held for the lifetime of the process.
  • `ProxyState` gains `langfuse: Option<Arc>` plus a
    `with_langfuse(...)` builder.
  • `chat::chat_completions` emits one event per request, success or
    failure — no behavior change when langfuse is `None`.

Tests (20 new)

  • Disabled config returns `None`
  • Enabled-without-host errors with `MissingHost`
  • Enabled-without-key-env errors with the specific env var name
  • Round-trip: wiremock upstream, emit one event, wait for the 1s
    flush, assert exactly one POST received
  • ISO timestamp round-trip including unix epoch + leap year
  • Base64 basic auth encoding matches the expected `pk:sk -> cGs6c2s=`
  • Channel-full does not block the emitter

Test plan

  • `cargo test --workspace` — 395 tests pass (+20 from baseline)
  • `cargo clippy --workspace --all-targets -- -D warnings` clean
  • `cargo fmt --check` clean
  • CI all-green
  • Streaming chat handler + other endpoints (messages/embeddings/
    rerank/audio/images/responses) emit langfuse events — deferred to
    a follow-up PR
    to keep this one reviewable

🤖 Generated with Claude Code

Per spec §3.6 / §9 / plan §4.9. New aisix-obs::langfuse module exports:
- LangfuseEvent / LangfuseSender / LangfuseHandle types
- spawn() that returns Ok(None) when disabled, or starts a background
batch flusher (50 events or 1s, whichever comes first) when enabled
The exporter authenticates with HTTP basic (public_key:secret_key),
POSTs to {host}/api/public/ingestion in the documented batch shape
({batch: [{type: 'generation-create', body: {...}}]}), and never
blocks the request hot path — full queues drop the event silently.
Wired into:
- aisix-server bootstrap (spawn after metrics, hold handle for life
of process)
- ProxyState (new optional langfuse: Option<Arc<LangfuseSender>>)
- chat::chat_completions (emits one event per request, success or
failure)
Tests (20 new):
- Disabled config returns None
- Enabled-without-host errors clearly
- Enabled-without-key-env errors with the missing env name
- Wiremock round-trip: emit one event, wait for the 1s flush
interval, assert the upstream received exactly one POST
- ISO timestamp round-trip including unix epoch + leap year
- Base64 basic auth encoding matches expected output
- Channel-full does not block the emitter
Streaming chat handler emission, plus other endpoints
(messages/embeddings/etc.), follow in a separate PR.
CopilotAI review requested due to automatic review settings April 20, 2026 00:32

CopilotAI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds an optional Langfuse ingestion exporter to the observability layer and wires it into the non-streaming chat completions handler so each request emits a generation event when enabled.

Changes:

  • Introduces aisix-obs::langfuse with a bounded, non-blocking event channel and background batch flusher to Langfuse /api/public/ingestion.
  • Wires a LangfuseSender into ProxyState and emits one LangfuseEvent per /v1/chat/completions request (success or failure).
  • Updates workspace plumbing (exports/deps) to support the new module.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file
FileDescription
crates/aisix-server/src/main.rsSpawns optional Langfuse exporter and injects sender into ProxyState.
crates/aisix-proxy/src/state.rsAdds optional langfuse sender field + builder.
crates/aisix-proxy/src/chat.rsEmits LangfuseEvent on completion for success/failure paths.
crates/aisix-obs/src/lib.rsExposes new langfuse module and re-exports its types.
crates/aisix-obs/src/langfuse.rsImplements batching exporter + tests.
crates/aisix-obs/Cargo.tomlAdds deps for HTTP/JSON/base64/uuid and wiremock for tests.
crates/aisix-core/src/lib.rsRe-exports LangfuseConfig.
Cargo.lockLocks new dependencies pulled in by the exporter/tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

start_time: iso_offset(ev.latency),
end_time: now_iso.clone(),
status_message: (ev.status_code != 200)
.then(|| format!("upstream status {}", ev.status_code)),
Comment on lines +462 to +467
// SAFETY: tests run single-threaded by default and we only mutate
// env vars that are scoped to this test's spawn() call.
unsafe {
std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test");
std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test");
}
Comment on lines +88 to +99
// Optional Langfuse exporter — disabled in config by default.
// When enabled, the proxy gets an Arc<LangfuseSender> through
// ProxyState and emits one event per chat completion at
// end-of-request. We keep the handle alive for the lifetime of
// the process so the background flush task continues running.
let langfuse_handle = match langfuse::spawn(&cfg.observability) {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, "langfuse exporter disabled");
None
}
};

loop {
tokio::select! {
biased;
Comment on lines +31 to +32
/// never blocks the proxy thread; if Langfuse is offline we drop the
/// oldest events at the edges.
@moonming
moonming merged commit 5185b54 into mainApr 20, 2026
10 checks passed
@moonming
moonming deleted the feat/langfuse-exporter branch April 20, 2026 00:44
jarvis9443 added a commit that referenced this pull request Jun 2, 2026
Output guardrails only inspected message.content, so client-visible
output that lives elsewhere bypassed content/DLP checks:
- tool_calls / Anthropic tool_use (normalized into message.extra) are
now folded into a single guardrail-inspected text view via
ChatResponse::guardrail_output_text(), used by the keyword, text-
moderation, Bedrock, and Prompt Shield output checks (#3/#18/#21).
Reasoning/thinking content is intentionally left out of scope.
- Non-streaming cache hits now run the resolved output guardrail chain
before returning the stored body, instead of replaying it unchecked
(#28). Streaming output guardrails already run end-of-stream.
Part of #448 (findings #3, #18, #21, #28)
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@moonming
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links"); } } catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); } })(); (function(){ try { var __m = "youtube.com"; var __re = new RegExp('^' + "youtube\\.com" + '
Skip to content

feat(obs): Langfuse exporter wired into chat completions - #21

Merged
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter
Apr 20, 2026
Merged

feat(obs): Langfuse exporter wired into chat completions#21
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter

Conversation

@moonming

Copy link
Copy Markdown
Member

Summary

New `aisix-obs::langfuse` module that pushes per-chat-completion
generation events to a Langfuse `/api/public/ingestion` endpoint.

There is no first-party Rust SDK for Langfuse, so this is a hand-rolled
HTTP client that:

  • Accepts events on a bounded mpsc channel (4096 cap) — emit is
    non-blocking, drops on overflow.
  • Drains the channel into batches of 50 (or every 1s, whichever fires
    first).
  • POSTs each batch with HTTP basic auth (public_key:secret_key) to
    `{host}/api/public/ingestion`.
  • Logs all upstream errors at WARN; never blocks the request hot path.

Wiring

  • Bootstrap (`aisix-server`): `langfuse::spawn(&cfg.observability)`
    returns `Ok(None)` when disabled, an opaque `LangfuseHandle` when
    enabled. The handle is held for the lifetime of the process.
  • `ProxyState` gains `langfuse: Option<Arc>` plus a
    `with_langfuse(...)` builder.
  • `chat::chat_completions` emits one event per request, success or
    failure — no behavior change when langfuse is `None`.

Tests (20 new)

  • Disabled config returns `None`
  • Enabled-without-host errors with `MissingHost`
  • Enabled-without-key-env errors with the specific env var name
  • Round-trip: wiremock upstream, emit one event, wait for the 1s
    flush, assert exactly one POST received
  • ISO timestamp round-trip including unix epoch + leap year
  • Base64 basic auth encoding matches the expected `pk:sk -> cGs6c2s=`
  • Channel-full does not block the emitter

Test plan

  • `cargo test --workspace` — 395 tests pass (+20 from baseline)
  • `cargo clippy --workspace --all-targets -- -D warnings` clean
  • `cargo fmt --check` clean
  • CI all-green
  • Streaming chat handler + other endpoints (messages/embeddings/
    rerank/audio/images/responses) emit langfuse events — deferred to
    a follow-up PR
    to keep this one reviewable

🤖 Generated with Claude Code

Per spec §3.6 / §9 / plan §4.9. New aisix-obs::langfuse module exports:
- LangfuseEvent / LangfuseSender / LangfuseHandle types
- spawn() that returns Ok(None) when disabled, or starts a background
batch flusher (50 events or 1s, whichever comes first) when enabled
The exporter authenticates with HTTP basic (public_key:secret_key),
POSTs to {host}/api/public/ingestion in the documented batch shape
({batch: [{type: 'generation-create', body: {...}}]}), and never
blocks the request hot path — full queues drop the event silently.
Wired into:
- aisix-server bootstrap (spawn after metrics, hold handle for life
of process)
- ProxyState (new optional langfuse: Option<Arc<LangfuseSender>>)
- chat::chat_completions (emits one event per request, success or
failure)
Tests (20 new):
- Disabled config returns None
- Enabled-without-host errors clearly
- Enabled-without-key-env errors with the missing env name
- Wiremock round-trip: emit one event, wait for the 1s flush
interval, assert the upstream received exactly one POST
- ISO timestamp round-trip including unix epoch + leap year
- Base64 basic auth encoding matches expected output
- Channel-full does not block the emitter
Streaming chat handler emission, plus other endpoints
(messages/embeddings/etc.), follow in a separate PR.
CopilotAI review requested due to automatic review settings April 20, 2026 00:32

CopilotAI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds an optional Langfuse ingestion exporter to the observability layer and wires it into the non-streaming chat completions handler so each request emits a generation event when enabled.

Changes:

  • Introduces aisix-obs::langfuse with a bounded, non-blocking event channel and background batch flusher to Langfuse /api/public/ingestion.
  • Wires a LangfuseSender into ProxyState and emits one LangfuseEvent per /v1/chat/completions request (success or failure).
  • Updates workspace plumbing (exports/deps) to support the new module.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file
FileDescription
crates/aisix-server/src/main.rsSpawns optional Langfuse exporter and injects sender into ProxyState.
crates/aisix-proxy/src/state.rsAdds optional langfuse sender field + builder.
crates/aisix-proxy/src/chat.rsEmits LangfuseEvent on completion for success/failure paths.
crates/aisix-obs/src/lib.rsExposes new langfuse module and re-exports its types.
crates/aisix-obs/src/langfuse.rsImplements batching exporter + tests.
crates/aisix-obs/Cargo.tomlAdds deps for HTTP/JSON/base64/uuid and wiremock for tests.
crates/aisix-core/src/lib.rsRe-exports LangfuseConfig.
Cargo.lockLocks new dependencies pulled in by the exporter/tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

start_time: iso_offset(ev.latency),
end_time: now_iso.clone(),
status_message: (ev.status_code != 200)
.then(|| format!("upstream status {}", ev.status_code)),
Comment on lines +462 to +467
// SAFETY: tests run single-threaded by default and we only mutate
// env vars that are scoped to this test's spawn() call.
unsafe {
std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test");
std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test");
}
Comment on lines +88 to +99
// Optional Langfuse exporter — disabled in config by default.
// When enabled, the proxy gets an Arc<LangfuseSender> through
// ProxyState and emits one event per chat completion at
// end-of-request. We keep the handle alive for the lifetime of
// the process so the background flush task continues running.
let langfuse_handle = match langfuse::spawn(&cfg.observability) {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, "langfuse exporter disabled");
None
}
};

loop {
tokio::select! {
biased;
Comment on lines +31 to +32
/// never blocks the proxy thread; if Langfuse is offline we drop the
/// oldest events at the edges.
@moonming
moonming merged commit 5185b54 into mainApr 20, 2026
10 checks passed
@moonming
moonming deleted the feat/langfuse-exporter branch April 20, 2026 00:44
jarvis9443 added a commit that referenced this pull request Jun 2, 2026
Output guardrails only inspected message.content, so client-visible
output that lives elsewhere bypassed content/DLP checks:
- tool_calls / Anthropic tool_use (normalized into message.extra) are
now folded into a single guardrail-inspected text view via
ChatResponse::guardrail_output_text(), used by the keyword, text-
moderation, Bedrock, and Prompt Shield output checks (#3/#18/#21).
Reasoning/thinking content is intentionally left out of scope.
- Non-streaming cache hits now run the resolved output guardrail chain
before returning the stored body, instead of replaying it unchecked
(#28). Streaming output guardrails already run end-of-stream.
Part of #448 (findings #3, #18, #21, #28)
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@moonming
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default"); } } catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

feat(obs): Langfuse exporter wired into chat completions - #21

Merged
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter
Apr 20, 2026
Merged

feat(obs): Langfuse exporter wired into chat completions#21
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter

Conversation

@moonming

Copy link
Copy Markdown
Member

Summary

New `aisix-obs::langfuse` module that pushes per-chat-completion
generation events to a Langfuse `/api/public/ingestion` endpoint.

There is no first-party Rust SDK for Langfuse, so this is a hand-rolled
HTTP client that:

  • Accepts events on a bounded mpsc channel (4096 cap) — emit is
    non-blocking, drops on overflow.
  • Drains the channel into batches of 50 (or every 1s, whichever fires
    first).
  • POSTs each batch with HTTP basic auth (public_key:secret_key) to
    `{host}/api/public/ingestion`.
  • Logs all upstream errors at WARN; never blocks the request hot path.

Wiring

  • Bootstrap (`aisix-server`): `langfuse::spawn(&cfg.observability)`
    returns `Ok(None)` when disabled, an opaque `LangfuseHandle` when
    enabled. The handle is held for the lifetime of the process.
  • `ProxyState` gains `langfuse: Option<Arc>` plus a
    `with_langfuse(...)` builder.
  • `chat::chat_completions` emits one event per request, success or
    failure — no behavior change when langfuse is `None`.

Tests (20 new)

  • Disabled config returns `None`
  • Enabled-without-host errors with `MissingHost`
  • Enabled-without-key-env errors with the specific env var name
  • Round-trip: wiremock upstream, emit one event, wait for the 1s
    flush, assert exactly one POST received
  • ISO timestamp round-trip including unix epoch + leap year
  • Base64 basic auth encoding matches the expected `pk:sk -> cGs6c2s=`
  • Channel-full does not block the emitter

Test plan

  • `cargo test --workspace` — 395 tests pass (+20 from baseline)
  • `cargo clippy --workspace --all-targets -- -D warnings` clean
  • `cargo fmt --check` clean
  • CI all-green
  • Streaming chat handler + other endpoints (messages/embeddings/
    rerank/audio/images/responses) emit langfuse events — deferred to
    a follow-up PR
    to keep this one reviewable

🤖 Generated with Claude Code

Per spec §3.6 / §9 / plan §4.9. New aisix-obs::langfuse module exports:
- LangfuseEvent / LangfuseSender / LangfuseHandle types
- spawn() that returns Ok(None) when disabled, or starts a background
batch flusher (50 events or 1s, whichever comes first) when enabled
The exporter authenticates with HTTP basic (public_key:secret_key),
POSTs to {host}/api/public/ingestion in the documented batch shape
({batch: [{type: 'generation-create', body: {...}}]}), and never
blocks the request hot path — full queues drop the event silently.
Wired into:
- aisix-server bootstrap (spawn after metrics, hold handle for life
of process)
- ProxyState (new optional langfuse: Option<Arc<LangfuseSender>>)
- chat::chat_completions (emits one event per request, success or
failure)
Tests (20 new):
- Disabled config returns None
- Enabled-without-host errors clearly
- Enabled-without-key-env errors with the missing env name
- Wiremock round-trip: emit one event, wait for the 1s flush
interval, assert the upstream received exactly one POST
- ISO timestamp round-trip including unix epoch + leap year
- Base64 basic auth encoding matches expected output
- Channel-full does not block the emitter
Streaming chat handler emission, plus other endpoints
(messages/embeddings/etc.), follow in a separate PR.
CopilotAI review requested due to automatic review settings April 20, 2026 00:32

CopilotAI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds an optional Langfuse ingestion exporter to the observability layer and wires it into the non-streaming chat completions handler so each request emits a generation event when enabled.

Changes:

  • Introduces aisix-obs::langfuse with a bounded, non-blocking event channel and background batch flusher to Langfuse /api/public/ingestion.
  • Wires a LangfuseSender into ProxyState and emits one LangfuseEvent per /v1/chat/completions request (success or failure).
  • Updates workspace plumbing (exports/deps) to support the new module.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file
FileDescription
crates/aisix-server/src/main.rsSpawns optional Langfuse exporter and injects sender into ProxyState.
crates/aisix-proxy/src/state.rsAdds optional langfuse sender field + builder.
crates/aisix-proxy/src/chat.rsEmits LangfuseEvent on completion for success/failure paths.
crates/aisix-obs/src/lib.rsExposes new langfuse module and re-exports its types.
crates/aisix-obs/src/langfuse.rsImplements batching exporter + tests.
crates/aisix-obs/Cargo.tomlAdds deps for HTTP/JSON/base64/uuid and wiremock for tests.
crates/aisix-core/src/lib.rsRe-exports LangfuseConfig.
Cargo.lockLocks new dependencies pulled in by the exporter/tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

start_time: iso_offset(ev.latency),
end_time: now_iso.clone(),
status_message: (ev.status_code != 200)
.then(|| format!("upstream status {}", ev.status_code)),
Comment on lines +462 to +467
// SAFETY: tests run single-threaded by default and we only mutate
// env vars that are scoped to this test's spawn() call.
unsafe {
std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test");
std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test");
}
Comment on lines +88 to +99
// Optional Langfuse exporter — disabled in config by default.
// When enabled, the proxy gets an Arc<LangfuseSender> through
// ProxyState and emits one event per chat completion at
// end-of-request. We keep the handle alive for the lifetime of
// the process so the background flush task continues running.
let langfuse_handle = match langfuse::spawn(&cfg.observability) {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, "langfuse exporter disabled");
None
}
};

loop {
tokio::select! {
biased;
Comment on lines +31 to +32
/// never blocks the proxy thread; if Langfuse is offline we drop the
/// oldest events at the edges.
@moonming
moonming merged commit 5185b54 into mainApr 20, 2026
10 checks passed
@moonming
moonming deleted the feat/langfuse-exporter branch April 20, 2026 00:44
jarvis9443 added a commit that referenced this pull request Jun 2, 2026
Output guardrails only inspected message.content, so client-visible
output that lives elsewhere bypassed content/DLP checks:
- tool_calls / Anthropic tool_use (normalized into message.extra) are
now folded into a single guardrail-inspected text view via
ChatResponse::guardrail_output_text(), used by the keyword, text-
moderation, Bedrock, and Prompt Shield output checks (#3/#18/#21).
Reasoning/thinking content is intentionally left out of scope.
- Non-streaming cache hits now run the resolved output guardrail chain
before returning the stored body, instead of replaying it unchecked
(#28). Streaming output guardrails already run end-of-stream.
Part of #448 (findings #3, #18, #21, #28)
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@moonming
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers"); } } catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

feat(obs): Langfuse exporter wired into chat completions - #21

Merged
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter
Apr 20, 2026
Merged

feat(obs): Langfuse exporter wired into chat completions#21
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter

Conversation

@moonming

Copy link
Copy Markdown
Member

Summary

New `aisix-obs::langfuse` module that pushes per-chat-completion
generation events to a Langfuse `/api/public/ingestion` endpoint.

There is no first-party Rust SDK for Langfuse, so this is a hand-rolled
HTTP client that:

  • Accepts events on a bounded mpsc channel (4096 cap) — emit is
    non-blocking, drops on overflow.
  • Drains the channel into batches of 50 (or every 1s, whichever fires
    first).
  • POSTs each batch with HTTP basic auth (public_key:secret_key) to
    `{host}/api/public/ingestion`.
  • Logs all upstream errors at WARN; never blocks the request hot path.

Wiring

  • Bootstrap (`aisix-server`): `langfuse::spawn(&cfg.observability)`
    returns `Ok(None)` when disabled, an opaque `LangfuseHandle` when
    enabled. The handle is held for the lifetime of the process.
  • `ProxyState` gains `langfuse: Option<Arc>` plus a
    `with_langfuse(...)` builder.
  • `chat::chat_completions` emits one event per request, success or
    failure — no behavior change when langfuse is `None`.

Tests (20 new)

  • Disabled config returns `None`
  • Enabled-without-host errors with `MissingHost`
  • Enabled-without-key-env errors with the specific env var name
  • Round-trip: wiremock upstream, emit one event, wait for the 1s
    flush, assert exactly one POST received
  • ISO timestamp round-trip including unix epoch + leap year
  • Base64 basic auth encoding matches the expected `pk:sk -> cGs6c2s=`
  • Channel-full does not block the emitter

Test plan

  • `cargo test --workspace` — 395 tests pass (+20 from baseline)
  • `cargo clippy --workspace --all-targets -- -D warnings` clean
  • `cargo fmt --check` clean
  • CI all-green
  • Streaming chat handler + other endpoints (messages/embeddings/
    rerank/audio/images/responses) emit langfuse events — deferred to
    a follow-up PR
    to keep this one reviewable

🤖 Generated with Claude Code

Per spec §3.6 / §9 / plan §4.9. New aisix-obs::langfuse module exports:
- LangfuseEvent / LangfuseSender / LangfuseHandle types
- spawn() that returns Ok(None) when disabled, or starts a background
batch flusher (50 events or 1s, whichever comes first) when enabled
The exporter authenticates with HTTP basic (public_key:secret_key),
POSTs to {host}/api/public/ingestion in the documented batch shape
({batch: [{type: 'generation-create', body: {...}}]}), and never
blocks the request hot path — full queues drop the event silently.
Wired into:
- aisix-server bootstrap (spawn after metrics, hold handle for life
of process)
- ProxyState (new optional langfuse: Option<Arc<LangfuseSender>>)
- chat::chat_completions (emits one event per request, success or
failure)
Tests (20 new):
- Disabled config returns None
- Enabled-without-host errors clearly
- Enabled-without-key-env errors with the missing env name
- Wiremock round-trip: emit one event, wait for the 1s flush
interval, assert the upstream received exactly one POST
- ISO timestamp round-trip including unix epoch + leap year
- Base64 basic auth encoding matches expected output
- Channel-full does not block the emitter
Streaming chat handler emission, plus other endpoints
(messages/embeddings/etc.), follow in a separate PR.
CopilotAI review requested due to automatic review settings April 20, 2026 00:32

CopilotAI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds an optional Langfuse ingestion exporter to the observability layer and wires it into the non-streaming chat completions handler so each request emits a generation event when enabled.

Changes:

  • Introduces aisix-obs::langfuse with a bounded, non-blocking event channel and background batch flusher to Langfuse /api/public/ingestion.
  • Wires a LangfuseSender into ProxyState and emits one LangfuseEvent per /v1/chat/completions request (success or failure).
  • Updates workspace plumbing (exports/deps) to support the new module.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file
FileDescription
crates/aisix-server/src/main.rsSpawns optional Langfuse exporter and injects sender into ProxyState.
crates/aisix-proxy/src/state.rsAdds optional langfuse sender field + builder.
crates/aisix-proxy/src/chat.rsEmits LangfuseEvent on completion for success/failure paths.
crates/aisix-obs/src/lib.rsExposes new langfuse module and re-exports its types.
crates/aisix-obs/src/langfuse.rsImplements batching exporter + tests.
crates/aisix-obs/Cargo.tomlAdds deps for HTTP/JSON/base64/uuid and wiremock for tests.
crates/aisix-core/src/lib.rsRe-exports LangfuseConfig.
Cargo.lockLocks new dependencies pulled in by the exporter/tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

start_time: iso_offset(ev.latency),
end_time: now_iso.clone(),
status_message: (ev.status_code != 200)
.then(|| format!("upstream status {}", ev.status_code)),
Comment on lines +462 to +467
// SAFETY: tests run single-threaded by default and we only mutate
// env vars that are scoped to this test's spawn() call.
unsafe {
std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test");
std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test");
}
Comment on lines +88 to +99
// Optional Langfuse exporter — disabled in config by default.
// When enabled, the proxy gets an Arc<LangfuseSender> through
// ProxyState and emits one event per chat completion at
// end-of-request. We keep the handle alive for the lifetime of
// the process so the background flush task continues running.
let langfuse_handle = match langfuse::spawn(&cfg.observability) {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, "langfuse exporter disabled");
None
}
};

loop {
tokio::select! {
biased;
Comment on lines +31 to +32
/// never blocks the proxy thread; if Langfuse is offline we drop the
/// oldest events at the edges.
@moonming
moonming merged commit 5185b54 into mainApr 20, 2026
10 checks passed
@moonming
moonming deleted the feat/langfuse-exporter branch April 20, 2026 00:44
jarvis9443 added a commit that referenced this pull request Jun 2, 2026
Output guardrails only inspected message.content, so client-visible
output that lives elsewhere bypassed content/DLP checks:
- tool_calls / Anthropic tool_use (normalized into message.extra) are
now folded into a single guardrail-inspected text view via
ChatResponse::guardrail_output_text(), used by the keyword, text-
moderation, Bedrock, and Prompt Shield output checks (#3/#18/#21).
Reasoning/thinking content is intentionally left out of scope.
- Non-streaming cache hits now run the resolved output guardrail chain
before returning the stored body, instead of replaying it unchecked
(#28). Streaming output guardrails already run end-of-stream.
Part of #448 (findings #3, #18, #21, #28)
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@moonming
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode"); } } catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); } })(); })();
Skip to content

feat(obs): Langfuse exporter wired into chat completions - #21

Merged
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter
Apr 20, 2026
Merged

feat(obs): Langfuse exporter wired into chat completions#21
moonming merged 1 commit into
mainfrom
feat/langfuse-exporter

Conversation

@moonming

Copy link
Copy Markdown
Member

Summary

New `aisix-obs::langfuse` module that pushes per-chat-completion
generation events to a Langfuse `/api/public/ingestion` endpoint.

There is no first-party Rust SDK for Langfuse, so this is a hand-rolled
HTTP client that:

  • Accepts events on a bounded mpsc channel (4096 cap) — emit is
    non-blocking, drops on overflow.
  • Drains the channel into batches of 50 (or every 1s, whichever fires
    first).
  • POSTs each batch with HTTP basic auth (public_key:secret_key) to
    `{host}/api/public/ingestion`.
  • Logs all upstream errors at WARN; never blocks the request hot path.

Wiring

  • Bootstrap (`aisix-server`): `langfuse::spawn(&cfg.observability)`
    returns `Ok(None)` when disabled, an opaque `LangfuseHandle` when
    enabled. The handle is held for the lifetime of the process.
  • `ProxyState` gains `langfuse: Option<Arc>` plus a
    `with_langfuse(...)` builder.
  • `chat::chat_completions` emits one event per request, success or
    failure — no behavior change when langfuse is `None`.

Tests (20 new)

  • Disabled config returns `None`
  • Enabled-without-host errors with `MissingHost`
  • Enabled-without-key-env errors with the specific env var name
  • Round-trip: wiremock upstream, emit one event, wait for the 1s
    flush, assert exactly one POST received
  • ISO timestamp round-trip including unix epoch + leap year
  • Base64 basic auth encoding matches the expected `pk:sk -> cGs6c2s=`
  • Channel-full does not block the emitter

Test plan

  • `cargo test --workspace` — 395 tests pass (+20 from baseline)
  • `cargo clippy --workspace --all-targets -- -D warnings` clean
  • `cargo fmt --check` clean
  • CI all-green
  • Streaming chat handler + other endpoints (messages/embeddings/
    rerank/audio/images/responses) emit langfuse events — deferred to
    a follow-up PR
    to keep this one reviewable

🤖 Generated with Claude Code

Per spec §3.6 / §9 / plan §4.9. New aisix-obs::langfuse module exports:
- LangfuseEvent / LangfuseSender / LangfuseHandle types
- spawn() that returns Ok(None) when disabled, or starts a background
batch flusher (50 events or 1s, whichever comes first) when enabled
The exporter authenticates with HTTP basic (public_key:secret_key),
POSTs to {host}/api/public/ingestion in the documented batch shape
({batch: [{type: 'generation-create', body: {...}}]}), and never
blocks the request hot path — full queues drop the event silently.
Wired into:
- aisix-server bootstrap (spawn after metrics, hold handle for life
of process)
- ProxyState (new optional langfuse: Option<Arc<LangfuseSender>>)
- chat::chat_completions (emits one event per request, success or
failure)
Tests (20 new):
- Disabled config returns None
- Enabled-without-host errors clearly
- Enabled-without-key-env errors with the missing env name
- Wiremock round-trip: emit one event, wait for the 1s flush
interval, assert the upstream received exactly one POST
- ISO timestamp round-trip including unix epoch + leap year
- Base64 basic auth encoding matches expected output
- Channel-full does not block the emitter
Streaming chat handler emission, plus other endpoints
(messages/embeddings/etc.), follow in a separate PR.
CopilotAI review requested due to automatic review settings April 20, 2026 00:32

CopilotAI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds an optional Langfuse ingestion exporter to the observability layer and wires it into the non-streaming chat completions handler so each request emits a generation event when enabled.

Changes:

  • Introduces aisix-obs::langfuse with a bounded, non-blocking event channel and background batch flusher to Langfuse /api/public/ingestion.
  • Wires a LangfuseSender into ProxyState and emits one LangfuseEvent per /v1/chat/completions request (success or failure).
  • Updates workspace plumbing (exports/deps) to support the new module.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file
FileDescription
crates/aisix-server/src/main.rsSpawns optional Langfuse exporter and injects sender into ProxyState.
crates/aisix-proxy/src/state.rsAdds optional langfuse sender field + builder.
crates/aisix-proxy/src/chat.rsEmits LangfuseEvent on completion for success/failure paths.
crates/aisix-obs/src/lib.rsExposes new langfuse module and re-exports its types.
crates/aisix-obs/src/langfuse.rsImplements batching exporter + tests.
crates/aisix-obs/Cargo.tomlAdds deps for HTTP/JSON/base64/uuid and wiremock for tests.
crates/aisix-core/src/lib.rsRe-exports LangfuseConfig.
Cargo.lockLocks new dependencies pulled in by the exporter/tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

start_time: iso_offset(ev.latency),
end_time: now_iso.clone(),
status_message: (ev.status_code != 200)
.then(|| format!("upstream status {}", ev.status_code)),
Comment on lines +462 to +467
// SAFETY: tests run single-threaded by default and we only mutate
// env vars that are scoped to this test's spawn() call.
unsafe {
std::env::set_var("LANGFUSE_PUBLIC_KEY", "pk-test");
std::env::set_var("LANGFUSE_SECRET_KEY", "sk-test");
}
Comment on lines +88 to +99
// Optional Langfuse exporter — disabled in config by default.
// When enabled, the proxy gets an Arc<LangfuseSender> through
// ProxyState and emits one event per chat completion at
// end-of-request. We keep the handle alive for the lifetime of
// the process so the background flush task continues running.
let langfuse_handle = match langfuse::spawn(&cfg.observability) {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, "langfuse exporter disabled");
None
}
};

loop {
tokio::select! {
biased;
Comment on lines +31 to +32
/// never blocks the proxy thread; if Langfuse is offline we drop the
/// oldest events at the edges.
@moonming
moonming merged commit 5185b54 into mainApr 20, 2026
10 checks passed
@moonming
moonming deleted the feat/langfuse-exporter branch April 20, 2026 00:44
jarvis9443 added a commit that referenced this pull request Jun 2, 2026
Output guardrails only inspected message.content, so client-visible
output that lives elsewhere bypassed content/DLP checks:
- tool_calls / Anthropic tool_use (normalized into message.extra) are
now folded into a single guardrail-inspected text view via
ChatResponse::guardrail_output_text(), used by the keyword, text-
moderation, Bedrock, and Prompt Shield output checks (#3/#18/#21).
Reasoning/thinking content is intentionally left out of scope.
- Non-streaming cache hits now run the resolved output guardrail chain
before returning the stored body, instead of replaying it unchecked
(#28). Streaming output guardrails already run end-of-stream.
Part of #448 (findings #3, #18, #21, #28)
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@moonming