dsh-plugin-langfuse
DeepSeek Harness 的 Langfuse 可观测性：OpenTelemetry 追踪、压缩、反馈评分和分叉谱系
插件会安装到这里；不确定时保持 web。
npx -y @deepseek-ai/dsh plugin --profile web add dsh-plugin-langfuse@0.6.0Plugin Langfuse 以 dsh-plugin-langfuse 发布，当前版本为 0.6.0。Plugin Hub 会校验它的 manifest，并保存精确安装来源，便于复现安装结果。
English | 中文
Langfuse observability for DeepSeek Harness (dsh): exports each turn as an OpenTelemetry trace — model step → generation, tool call → tool span — groups turns by session, records session remarks and message ratings as Langfuse Scores, and preserves fork/subagent lineage.
This is a community plugin (dsh-plugin topic), not part of the official repository. It implements the harness's public telemetry seam (@deepseek-ai/dsh-session-telemetry) as an alternative backend to the official OTLP-logs exporter.
The commands below assume the installed dsh CLI. Running the official harness from a source checkout instead? Run each of them as pnpm dsh … from the checkout root (after its pnpm run build) — same commands, same web profile.
As a profile bundle (the package ships a cordis.patch.yml patch layer):
dsh plugin --profile web add dsh-plugin-langfuse
export LANGFUSE_PUBLIC_KEY=pk-lf-…
export LANGFUSE_SECRET_KEY=sk-lf-…
# optional, defaults to https://cloud.langfuse.com (EU region); note the plugin
# reads LANGFUSE_HOST, not the Langfuse SDK's LANGFUSE_BASE_URL
export LANGFUSE_HOST=https://us.cloud.langfuse.com
dsh web # alias for: dsh --profile web
The bundled patch disables the base profile's session-telemetry-otel row (the telemetry seam accepts exactly one backend per context; a duplicate load throws) and mounts this backend in FULL mode when a Langfuse key is present, DISABLED otherwise. It also enables feedback Scores when both project keys are present. LANGFUSE_TELEMETRY_MODE=FEEDBACK_ONLY narrows sharing to feedback-gated release.
Both the bundle layer and the env vars are read at boot: an already-running instance must be restarted after installing, from a shell that has the variables set. dsh --profile web --dump-config shows the composed result without booting — a # == dsh-plugin-langfuse layer that patches the base telemetry row and adds session-telemetry-langfuse with its env-driven mode. After the next turn, traces appear in the Langfuse console of the region LANGFUSE_HOST points at — keys are region-scoped, so a US project shows nothing on the EU console. dsh plugin --profile web remove dsh-plugin-langfuse removes both the dependency and the layer.
Or as an explicit cordis.yml row:
- id: session-telemetry-langfuse
name: dsh-plugin-langfuse
config:
mode: FULL # FULL | FEEDBACK_ONLY | DISABLED (default)
exporter: # passed verbatim to the SDK's OTLP/HTTP trace exporter
url: https://cloud.langfuse.com/api/public/otel/v1/traces
auth:
publicKey: !!js process.env.LANGFUSE_PUBLIC_KEY
secretKey: !!js process.env.LANGFUSE_SECRET_KEY
feedbackScores: # optional; disabled by default for explicit rows
enabled: true
url: https://cloud.langfuse.com/api/public/scores
maxQueueSize: 256
requestTimeoutMillis: 3000
content: # optional privacy/content controls
turnInputMode: user # none | user | user-and-context
cwdMode: omit # omit | basename | full
toolMetaAllowlist: [] # exact top-level tool-result meta keys
metadata: # optional static Langfuse grouping
environment: production
tags: [dsh]
health:
warningIntervalMillis: 60000
maxErrorChars: 1000
processor: {} # optional; passed verbatim to BatchSpanProcessor
shutdownTimeoutMillis: 3000
| Field | Meaning |
|---|---|
mode | FULL exports every session live; FEEDBACK_ONLY replays and exports the canonical session log only when the user records feedback; DISABLED (default) constructs nothing and nothing leaves the process. FULL is this plugin's continuous tracing mode; Harness 0.1.5's official OTLP-logs backend offers FEEDBACK_ONLY and DISABLED. |
exporter | The complete OTLPExporterNodeConfigBase object, passed to the OTLP/HTTP trace exporter. url is required outside DISABLED and must be the full traces path (…/api/public/otel/v1/traces). The plugin defaults an x-langfuse-ingestion-version: 4 header — without it new spans do not land on Langfuse's v4 data model in real time. An explicit entry (any casing) wins, whether supplied in a plain exporter.headers object or returned by a HeadersFactory. |
auth | Langfuse project key pair, turned into the endpoint's Basic-auth header. Mutually exclusive with an explicit exporter.headers authorization; uploading modes require exactly one of the two. |
correlation | Host-identity correlation: userId/sessionId stamped as langfuse.user.id/langfuse.session.id on every exported span so an embedding host's traces and this plugin's group under one Langfuse user/session. See Correlating with an embedding host. |
feedbackScores | Optional session-level TEXT Scores for feedback/record and CATEGORICAL message ratings for feedback/message-put, with updates and deletion. enabled defaults to false; url must be the full …/api/public/scores path. maxQueueSize defaults to 256 and requestTimeoutMillis to 3000. The bounded in-memory queue is failure-isolated from tracing and drains best-effort on shutdown. The bundled profile enables it when both project keys exist. |
content | Export-content policy. turnInputMode defaults to user (aggregate human messages only); user-and-context also includes plugin-injected context and none omits root input. defaults to ; exports only the final directory and exports the full path. defaults empty and admits only named top-level keys. |
Misconfiguration fails loud at plugin load: a missing/malformed/non-http(s) exporter URL, missing credentials, ambiguous double auth, a non-positive maxExportBatchSize (the SDK would hang on shutdown), invalid correlation/content/metadata/health settings, an enabled Score sink without a valid URL/queue/timeout, or an unknown mode all throw before any transport is constructed.
LangfuseSessionTelemetryBackend.status() returns a synchronous detached snapshot with overall and per-channel state, trace batch/span success and failure counts, consecutive failures, recent timestamps, sanitized last error, and Score queued/delivered/dropped/skipped/failed counts. States are disabled, starting, healthy, degraded, and stopped; Score remains an independent channel. The OTel SDK does not expose BatchSpanProcessor queue depth, so traces.queuedBySdk is explicitly unknown. First failure, rate-limited continuing failure, and recovery are also logged without Authorization or Langfuse keys.
In a standard interactive Harness profile, inspect the same snapshot without leaving the conversation UI:
/langfuse status
/langfuse status --json
The first form is a compact human-readable report; --json returns a stable envelope containing the plugin version, session-sharing policy, and the complete status() snapshot. The command is local-only: it does not contact Langfuse, retry delivery, inspect credentials, or force an SDK flush. It is available whenever the Harness commands service is composed (including the standard web profile); telemetry-only/headless compositions without that optional service continue to load the backend without registering the command. starting means no trace export batch has completed yet, not that the command is probing the endpoint. Even DISABLED mode registers the command when possible, so /langfuse status can confirm that nothing is being shared.
A host application that embeds the dsh runtime and already emits its own traces into the same Langfuse project can steer this plugin's identity so both views group under one Langfuse user/session — the host typically injects its ids as env vars when spawning the runtime:
config:
correlation:
userId: !!js process.env.HOST_USER_ID
sessionId: !!js process.env.HOST_SESSION_ID
langfuse.session.id/langfuse.user.id ride every exported span — turn, generation, tool, and compaction — because Langfuse's v4 query model filters and aggregates per observation, not only per trace (propagation contract).sessionId defaults to the dsh session id, and the original dsh session id stays on each logical root as dsh.session.id — the pointer back into $DSH_HOME/sessions for local diagnosis.turn/start record carrying langfuse.user.id/langfuse.session.id attributes overrides the static config for that turn — a deployment injects them through a session-telemetry/record waterfall listener. The snapshot is locked at turn/start; identity attributes on later records are ignored. Precedence: record attributes > correlation config > dsh session id.FEEDBACK_ONLY replay — otherwise the replayed tree exports under a different identity than live capture would have.correlation values bypass the redaction waterfall: the waterfall transforms records, and these values never transit one.| dsh session event | Langfuse concept |
|---|---|
session (session.id) | session (langfuse.session.id on every exported observation/span) |
turn/start / turn/end | trace root observation (root span; error end reasons set span status ERROR) |
step/start / step/end + request/header + request/context + assistant/message | generation — model, provider, safe request parameters/context window, output, canonical gen_ai.usage.* tokens (input/output/cache-read/cache-creation/reasoning); the latest assistant message also becomes the root observation's overall output. Interrupted partial output is retained and marks both observations with dsh.assistant.interrupted=true without classifying the interruption as an error |
llm/retry / llm/retry-started | structured scheduled/started events on the existing generation, including retry id/attempt/policy/delay and clipped failure details; no synthetic generation is created because the current Harness event contract provides no per-attempt usage lifecycle |
assistant/message.stream | langfuse.observation.completion_start_time from the first token of the committed message, using original stream timestamps |
assistant/attempt | bounded attempt metadata on the step generation; known usage is included once, with terminal outcome retained |
system/message | event type/sequence only; system prompt text is not exported |
tool/call + tool/result | tool span (arguments as input, the full result content array as output, structured error name/code/outcome, isError → status ERROR; private meta is omitted unless allowlisted) |
approval/asked + approval/decided | timed internal approval span under the corresponding tool when callId resolves, otherwise under the current generation/turn; incomplete approvals are force-closed as ERROR |
user/message | aggregate root observation input according to content.turnInputMode; deprecated trace input is retained for legacy evaluator compatibility |
Token accounting follows the OpenTelemetry GenAI inclusive-total contract. DSH reports mutually exclusive input buckets (inputTokens is uncached input), so the exported gen_ai.usage.input_tokens is reconstructed as inputTokens + cacheReadTokens + cacheWriteTokens; cache read/write and reasoning remain canonical detail attributes. Langfuse can then normalize them into mutually exclusive usage buckets exactly once.
The harness's rule is model-visible ⟺ logged: everything that reaches a model request is reconstructable from the canonical session log, and new behavior lands as a plugin on documented extension points, never as agent-loop changes. The telemetry seam (@deepseek-ai/dsh-session-telemetry) is exactly the extension point built for handing session records to a reporting SDK. Implementing its SessionTelemetryBackend buys, for free and with guaranteed consistency:
session-telemetry/record redaction waterfall (deployment-owned scrub rules apply to the exported copy; the canonical log is never rewritten);FEEDBACK_ONLY consent semantics (nothing leaves until the user records feedback, and only the committed canonical event is consent);Instrumenting the LLM adapter or agent loop directly would duplicate all of that, drift from the log, and break the moment the loop changes.
The official session-telemetry-otel backend cannot feed Langfuse: it exports OTLP logs, and Langfuse's OTLP endpoint (/api/public/otel) accepts traces only, over OTLP/HTTP (JSON or protobuf; no gRPC), with Basic auth. That mismatch — not a missing URL — is why this plugin exists.
The trace pipeline is the plain OTel traces SDK (BasicTracerProvider → BatchSpanProcessor → OTLPTraceExporter), the same SDK family and configuration surface as the official backend, with attributes following the OTel GenAI semantic conventions plus Langfuse's documented langfuse.* property mapping. Feedback Scores use a small native-HTTP transport instead of a second tracing SDK so the plugin can reuse its async/custom auth contract and keep trace versus Score failures isolated. The Langfuse SDK can replace that internal transport later without changing the telemetry seam or public configuration.
The seam's records mirror session-log events one-to-one; Langfuse needs trace → observation hierarchy. SessionSpanFolder is a state machine keyed by (session.id, turn, step, compactionId) that folds records into open OTel spans. Its contract-critical choices:
FEEDBACK_ONLY canonical-log replay produce identical trees (span start/end times are explicit — the OTel API supports historical stamps).assistant/message.stream or assistant/attempt.stream; the plugin reads timing, finish and usage without exporting the raw stream. completion_start_time describes the committed message, while failed attempts keep their own first-token timestamp as event metadata. Empty/redacted streams do not acquire invented timing.tool/call and tool/result land inside the step's boundaries — so the generation span temporally contains its tool executions. A call whose step is no longer open (crash-window replay) falls back to the turn span.user/message supplies its input, and each completed assistant message replaces its output so the final message remains at turn end. Deprecated langfuse.trace.input/output aliases are emitted only for legacy trace-level evaluator compatibility.compaction/start through compaction/end. Its duration deliberately includes orchestration around the provider call and is not labeled as pure model latency. compaction/summary enriches the span with the compacted summary, provider/model/usage, and aggregate shadow statistics; provider rawOutput and the full shadowedSeqs list are never exported. The paired replacement user/message (source.plugin=compact) remains model-visible context but never overwrites the turn's human input. A missing owner becomes a stable standalone trace, while missing/malformed lifecycle records degrade to point events or an ERROR span rather than fabricated timing.(dsh session id, turn) and (dsh session id, compaction id) supply stable 32-hex Trace IDs across live export and FEEDBACK_ONLY replay. A valid W3C traceparent still wins for distributed tracing; the deterministic ID remains queryable metadata.dsh_user_feedback TEXT Scores; message ratings become dsh_message_feedback CATEGORICAL Scores with dshMessageId metadata. Both use a bounded single-worker queue. Message edits and deletions remain ordered behind earlier retries. Ratings are attached to the resolved Langfuse session, not a fabricated observation. Notes are clipped to 500 characters and categories are retained.dsh.lineage.linked=false; no context is fabricated.The capture cursor marks handed off, not delivered; whatever sits in the SDK batch queue at crash time is lost, and a cursor-less re-adoption (hot reload) may re-hand a prefix, producing duplicate spans. Receivers correlate on langfuse.session.id + dsh.turn + dsh.event.seq. A durable outbox is deliberately out of scope, matching the seam's own stance.
In uploading modes, span attributes carry user and assistant message content, tool arguments and results, compaction summaries and aggregate shadow statistics, and model/usage metadata, as returned by the session-telemetry/record waterfall. Compaction provider rawOutput and full shadowedSeqs lists are deliberately omitted. This plugin ships no redaction rules; a deployment exporting beyond a trusted boundary mounts its own waterfall listener. Provider credential fields are not added by this plugin; arbitrary user/tool content still requires deployment redaction. Serialized payloads are clipped at maxAttributeChars per attribute (default 32768); the canonical log keeps the full bytes.
None, as this plugin only observes the session stream through the telemetry seam and hands folded spans to the OTel SDK; it never contributes to a model request.
None; this plugin neither assembles nor sends a provider request.
npm run typecheck && npm run typecheck:tests # production code and all test fixtures
npm test # unit: status command + folding projection + config fail-loud paths
npm run build && npm run test:e2e # REAL composition: boots a real dsh app via the
# Loader (mock model, real bash round trip) and asserts
# the OTLP payload a mock Langfuse collector received
npm run test:e2e:cloud # opt-in real Langfuse round trip; loads credentials from .env
npm run test:package # npm pack + empty-consumer install/import + bundle composition
The e2e follows the official repository's REAL-composition pattern (@deepseek-ai/dsh-app-boot + @deepseek-ai/dsh-loader-smoke): the fixture cordis.yml loads the built lib/index.js — the same file a deployment loads — and assertions run against the wire, including compact V3 first-token timing, retry payloads, approval/tool-error metadata, a complete standalone compaction, and a seed-boundary orphan compaction, not against internals.
The cold-feedback E2E uses the real JSONL persistence and message-feedback service with the built backend: it verifies history replay, ordered rating create/edit/delete over HTTP, duplicate suppression and write-handle release. Unit regressions additionally cover empty/category-only consent, restored/forked sessions, forged events, waterfall redaction, failed/retried/cancelled attempts and V3 Score identities.
The status-command suite exercises human and JSON rendering, strict argument handling, Harness command registration, recordInput: false, disabled-mode diagnostics, and a headless composition without the optional command service. A separate local e2e drives the real OTLPTraceExporter and BatchSpanProcessor through an HTTP 503 followed by a 200 response and verifies the public backend status transition from degraded to healthy. The backend unit suite independently locks the observer-to-status() wiring.
npm run test:e2e:cloud loads the gitignored .env file and runs only the opt-in Langfuse Cloud round trip. Copy .env.example to .env and provide LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and the matching regional LANGFUSE_HOST. The test checks the v4 Observations API for root input/output, usage, per-observation correlation, parent/child metadata, standalone compaction identity/summary/usage, approval outcome, structured tool errors, and the Scores API for feedback readback. It also verifies that a retry lifecycle does not create a duplicate generation. A separate Cloud case reads back a CATEGORICAL message rating after creation and revision, then verifies its deletion through the same stable Score ID. The retry event payload itself is locked by the local raw-OTLP wire test because the Observations API returns the observation, not its embedded OTel span events. The generic npm run test:e2e command still runs the Cloud case when those variables are already exported, and otherwise self-skips it. Set LANGFUSE_REQUIRE_TOTAL_COST=1 to additionally require a finite positive totalCost for the fixture’s deepseek-v4-flash step generations; LANGFUSE_E2E_COST_MODEL may select an isolated test alias with the same pricing. This is an opt-in validation of the test project's Langfuse model-pricing setup, never a price hard-coded by the plugin.
DeepSeek Harness is in developer preview with no compatibility promises; this plugin pins exact @deepseek-ai/dsh-* versions.
| dsh-plugin-langfuse | @deepseek-ai/dsh-* |
|---|---|
| 0.1.x | 0.1.0-rc.6 |
| 0.2.x | 0.1.0-rc.6 |
| 0.3.x | 0.1.0-rc.7 |
| 0.4.x | 0.1.0-rc.8 |
| 0.5.0 | 0.1.0-rc.8 |
| 0.5.1 | 0.1.1-rc.1 |
| 0.5.2 | 0.1.2-rc.1 |
| 0.6.0 | 0.1.5-rc.1 |
The separate Upstream compatibility canary workflow resolves every @deepseek-ai/* dependency to its newest published version. Pull requests and main pushes are advisory; the weekly schedule and manual dispatch are strict and run source/test typechecks, unit tests, build, REAL-composition e2e, and package smoke. Failed runs retain the resolved manifest and lockfile for reproduction.
Use matching versions of the Harness and plugin; 0.6.0 targets @deepseek-ai/dsh-* 0.1.5-rc.1. Update the plugin bundle with dsh plugin --profile web add dsh-plugin-langfuse@0.6.0, restart the Harness with the Langfuse environment variables present, then check dsh --profile web --dump-config and /langfuse status --json. Source-checkout users run the same commands as pnpm dsh … after building the matching Harness revision. Keep plugin 0.5.2 with Harness 0.1.2-rc.1.
Harness owns log migration to Session V3 and preserves prior generations; upgraded logs cannot be read by the older runtime. The plugin consumes the current canonical events and does not rewrite stored logs. Turn/compaction Trace ID algorithms remain unchanged. Session TEXT Score IDs now include the source format version to prevent collisions when migration renumbers event sequences; an old exported remark may therefore appear as a separate Score after migration/replay. The legacy two-argument createDshFeedbackScoreId API retains its old vectors.
FEEDBACK_ONLY includes restored history through the new, canonical feedback event, including category-only or empty session remarks and message-rating mutations. Cold message feedback is handled through feedback/committed; its borrowed snapshot is cloned and only queued for export, without opening another write handle or waiting for network. Fork-inherited events are not exported as child-owned observations or Scores, and inherited/foreign feedback cannot authorize child capture. FULL follows live events and handles cold feedback, but does not automatically backfill every resumed log. DISABLED builds no exporters.
A bounded 1,024-session/format handoff cursor suppresses repeated cold snapshots in one backend lifetime. It is not persisted and marks handoff, not delivery: eviction, restart, HMR, or migration may replay data. Message-rating upserts/deletes use a stable owning-session/message ID. No durable outbox or cross-process exactly-once guarantee is provided.
Generation duration remains the complete step, including retry waits and tools. Known usage from settled failed attempts and the committed message is summed once; a message's top-level usage wins over its embedded usage copy. Intermediate failure does not mark a later successful generation as ERROR. Terminal model errors/no-message settlements are ERROR; user cancellation is recorded as aborted with partial output retained, without treating a normal user stop as a service failure. System prompt text and raw streams stay out of exported attributes; turnInputMode: none only omits root input, not assistant/tool output.
cwdModeomitbasenamefulltoolMetaAllowlisttool/result.metametadata | Optional static Langfuse environment and tags, propagated to every observation for v4 querying. Environment follows Langfuse's lowercase a-z0-9-_ format, must not start with langfuse, and is at most 40 characters; at most 50 tags of at most 200 characters each are accepted. |
health | Delivery diagnostics. warningIntervalMillis defaults to 60000 and rate-limits continuing failure warnings (0 suppresses repeats after the first warning); maxErrorChars defaults to 1000 after credential/URL sanitization. These settings do not add retries or change SDK buffering. |
processor | Passed verbatim to BatchSpanProcessor (scheduledDelayMillis, maxQueueSize, maxExportBatchSize, …); batching, retry, and loss policy are the SDK's documented behavior. |
maxAttributeChars | Serialized-payload ceiling per span attribute (default 32768); longer payloads are clipped with an …[clipped] marker while the canonical session log keeps the full bytes. |
shutdownTimeoutMillis | Plugin-owned outer deadline on the SDK's shutdown drain (default 3000). |
session/title / subagent/descriptor / agent-preset/selected | session semantic state used to set langfuse.trace.name and safe browsing metadata on current/future observations without changing stable span names or IDs |
session/end-seed | closes compactions inherited without a matching end at the seed boundary and marks them incomplete/ERROR |
feedback/record | session-level dsh_user_feedback TEXT Score for post-waterfall text; category metadata is retained; absent text still authorizes replay without creating a TEXT Score |
feedback/message-put / feedback/message-delete | session-level dsh_message_feedback CATEGORICAL rating (positive / negative) keyed by the owning session and message; edits upsert and removals delete the same Score ID |
| forked child session | independent child turn trace plus queryable parent/seed metadata; an OTel Link points to the completed parent turn when its in-process context is retained |
agent-error ops record | agent-error span event + status ERROR on the open turn |
compaction/start + compaction/summary + compaction/end | one generation spanning the whole compaction transaction; child of its owning turn when available, otherwise a stable standalone trace; includes provider/model/usage and shadowed range/count/token statistics |
compaction/prune | point-in-time span event with the pruned range/count/token statistics |
| every other event type (todo, plan, hooks, plugin events) | point-in-time span event on the open turn |
dsh.force_endedturn/startshutdown