dsh-trajectory-persistence
Verifieddsh-trajectory-persistence Β· v0.1.0 Β· MIT
DeepSeek Harness (dsh) plugin: persist session trajectories to S3/OSS (JSONL) and OTel GenAI spans (OTLP).
Install
dsh plugin add dsh-trajectory-persistence Confirm the layer applied with dsh --profile default --dump-config β see the install guide.
Source
Tags
Creators
Readme
dsh-trajectory-persistence
Trajectory persistence plugin for the DeepSeek Harness (dsh).
An observe-only cordis plugin that subscribes to the session event firehose
(session/event / session/created / session/flush / session/disposed)
and persists every session's trajectory to two independently toggleable sinks:
- S3 / OSS sink β two delivery modes:
push(default, legacy) buffers the live event stream and uploads JSONL part files compatible with the@deepseek-ai/dsh-session-persistence-jsonlartifact layout (header line + one event per line), with a bounded in-memory ring buffer, batch uploads, exponential-backoff retry, and a local dead-letter directory;shiptails the official jsonl backend's on-disk artifact read-only and uploads byte-aligned zstd frame segments plus a per-session manifest β restorable on another machine with the bundledsync-downCLI. - OTel GenAI sink β spans following the OpenTelemetry GenAI semantic conventions, exported over OTLP HTTP/protobuf straight to Jaeger, to an OTel Collector (which can land them in ClickHouse, β¦), or SigV4-signed to AWS CloudWatch / Bedrock AgentCore Observability.
Compatibility notice. The DeepSeek Harness is in developer preview with no compatibility guarantees. This plugin was built and verified against monorepo commit
47f943859bef60e4160492346772ded9b24f765a(masteras of 2026-08-14,@deepseek-ai/[email protected]).
Architecture
flowchart LR
subgraph dsh [DeepSeek Harness]
SS[SessionStore<br/>ctx.sessions] -->|session/created| P
SS -->|session/event seq-ordered firehose| P
SS -->|session/flush| P
SS -->|session/disposed| P
P[dsh-trajectory-persistence<br/>observe-only plugin]
end
subgraph sinks [Sinks β independently enabled]
S3S[S3 sink β push mode<br/>ring buffer + batcher<br/>backoff retry]
OTS[OTel GenAI sink<br/>GenAISpanMapper<br/>BatchSpanProcessor]
end
P --> S3S
P --> OTS
S3S -->|PutObject JSONL parts<br/>prefix/projectId/sessionId/seqStart-seqEnd.jsonl| S3[(AWS S3 / Aliyun OSS / MinIO)]
S3S -->|final failure| DL[(local dead-letter dir)]
OTS -->|OTLP HTTP/protobuf| J[Jaeger :4318]
OTS -->|OTLP HTTP/protobuf| C[OTel Collector]
OTS -->|OTLP + SigV4| CW[(CloudWatch / AgentCore Observability)]
C -->|clickhouse exporter| CH[(ClickHouse)]
Event β GenAI span mapping
| dsh session events | span | key attributes |
|---|---|---|
turn/start β¦ turn/end |
gen_ai.turn |
gen_ai.operation.name=turn, gen_ai.conversation.id=<sessionId> |
step/start β¦ step/end + assistant/message + assistant/chunk |
chat (child of turn) |
gen_ai.operation.name=chat, gen_ai.request.model (from the latest request/context), gen_ai.usage.input_tokens / gen_ai.usage.output_tokens (from assistant/message.usage), chunk aggregation in dsh.assistant.* |
tool/call β¦ tool/result (paired by callId) |
execute_tool (child of the step's chat span) |
gen_ai.tool.name, gen_ai.tool.call.id, gen_ai.tool.call.arguments (truncated at 8192 characters); failures (error or isError) set span status ERROR |
Spans still open when a session is disposed are ended defensively; a
tool/call whose turn ends without tool/result is closed with status ERROR.
S3 delivery modes: push vs ship
The S3 sink runs in one of two modes (sinks.s3.mode, default push):
push (legacy, default) |
ship |
|
|---|---|---|
| Data source | Live event stream (session/event firehose) |
Official jsonl backend's on-disk artifact (read-only tail) |
| Artifact format | Self-contained JSONL parts: {prefix}/{projectId}/{sessionId}/{seqStart}-{seqEnd}.jsonl |
Byte-aligned zstd frame segments + _manifest.json: {prefix}/{projectDir}/{sessionId}/{offsetStart}-{offsetEnd}.jsonl.zstd |
| Latency | Near real-time (flush at batchSize / session/flush) |
Poll-driven (pollIntervalMs; a segment ships at segmentBytes, after segmentMaxDelayMs without growth, on dormancy, or on close) |
| Crash safety | Buffered-but-unflushed events are lost; failed parts dead-letter locally | Only complete zstd frames ship; the torn tail a crash leaves behind never leaves the machine |
| Resumability | Parts are independent; no cross-machine resume story | Manifest watermark resumes after state loss; sync-down restores the artifact on another machine |
| Best for | Trajectory analytics, long-term archival | Backup/restore, moving sessions between machines, dsh resume elsewhere |
push is legacy but fully supported and remains the default. In ship
mode the sink never subscribes to the event stream and never re-serializes:
complete zstd frames are cut out of the official artifact and uploaded as-is,
and the per-session _manifest.json watermark is authoritative β on first
contact with a session it wins over the local ship-state, so a lost state file
resumes instead of re-shipping. A manifest owned by another machine's
writerId logs a warning (do not run two shippers on one artifact); an
artifact whose size regresses below the watermark is marked conflicted and
skipped. See Ship & Sync
for the full picture.
Restoring sessions on another machine (sync-down)
Ship mode plus the bundled CLI covers the machine-switch workflow β with single-writer discipline: end the session on the old machine before switching.
Machine A β run dsh with
sinks.s3.mode: ship; end the session (the dormant trigger or the sink's close flush ships the remaining tail).Machine B β with dsh not running against the local session root:
dsh-trajectory-persistence sync-down --bucket dsh-trajectories --region us-east-1Machine B β start dsh and
dsh resume <session>; the restoredsession.jsonl.zstdis the official backend's own format.
Restoring is a concatenation of the manifest's segments (validated to tile
[0, watermark) contiguously), published with the official backend's
durability semantics (temp file + fsync, atomic publish, directory fsync).
Local artifacts are never silently overwritten: an identical one is skipped, a
byte-prefix one is completed in place (appended), and a diverged one is
refused unless --force is given β which first backs it up to
session.jsonl.zstd.bak-<epochMs>. Useful flags: --session <id> (one
session only), --root <dir> (default $DSH_HOME/sessions), --prefix,
--endpoint / --force-path-style for OSS/MinIO; credentials come from the
AWS default provider chain.
S3 sink flush triggers (push mode)
session/flushevent (the harness's durability checkpoint),- the session's buffer reaching
batchSizeevents, session/disposed,- cordis fiber disposal (graceful drain of all sinks).
Each flush uploads part files named {prefix}/{projectId}/{sessionId}/{seqStart}-{seqEnd}.jsonl,
where projectId is the readable projectKey(cwd) slug (_no-cwd when the
session has no cwd) and sessionId is ~XXXX-escaped β byte-compatible with
the jsonl persistence backend's encoding. Every part starts with the
type: "session" header line, so each part is independently parseable.
Uploads retry maxRetries times with exponential backoff (retry n waits
retryBaseDelayMs * 2^(n-1), plus 25 % jitter). A part that still fails is written verbatim to
{deadLetterDir}/{projectId}/{sessionId}/{seqStart}-{seqEnd}.jsonl.
Install into a profile
The package carries the bundle manifest expected by the CLI:
// package.json
"dsh": { "bundle": { "patch": "./cordis.patch.yml" } }
dsh plugin add dsh-trajectory-persistence
This applies the shipped cordis.patch.yml, which inserts one (inert) row:
- insert:
- id: trajectory-persistence
name: dsh-trajectory-persistence
All sinks default to disabled; enable and configure them in your profile's
own cordis.patch.yml (later layers replace a row's whole config, so restate
the full config block), or β with no restart at all β through
settings.yaml, see Live settings & status:
# $DSH_HOME/profiles/<your-profile>/cordis.patch.yml
- replace:
- id: trajectory-persistence
name: dsh-trajectory-persistence
config:
sinks:
s3:
enabled: true
bucket: my-trajectory-bucket
otel:
enabled: true
url: http://localhost:4318/v1/traces
Live settings & status
The plugin integrates two optional harness capabilities. Both degrade gracefully: on a profile without the settings or commands service the plugin loads exactly as before, just without these two features.
settings.yaml switches (hot-reload)
When a settings provider is mounted (the standard harness profiles have one),
the plugin registers its whole Config schema as the trajectory-persistence
settings namespace. The composed plugin config (cordis.patch.yml layers)
becomes the base; $DSH_HOME/settings.yaml layers on top of it β deep-merged,
so you only restate the keys you change:
# $DSH_HOME/settings.yaml
trajectory-persistence:
sinks:
s3:
enabled: true
bucket: my-trajectory-bucket
Every committed change applies live, without a restart: the plugin compares
the newly resolved config per sink and rebuilds exactly the sinks whose
effective config changed (sinks.s3.* and sinks.otel.* are independent). The
replaced sink first drains β buffered events are uploaded or dead-lettered,
open spans are ended and flushed β so no trajectory data is lost on a switch.
New session events go to the new sink immediately. A rebuild whose config the
sink cannot use keeps the previous sink running and logs a warning; a write
that violates the cross-field rules (sinks.s3.enabled without bucket,
batchSize > maxBufferedEvents, sinks.otel.enabled without url or aws,
sinks.otel.url together with sinks.otel.aws) is refused
upfront by the namespace's validate hook, before anything persists. If the
settings service itself goes away, the plugin falls back to the composed
config.
/trajectory-status
When the commands service is mounted, the plugin registers a slash command that reports the switch state, upload statistics, and the most recent error of each sink, plus whether the settings namespace has taken over the config:
trajectory-persistence status
settings: managed by the settings service (namespace "trajectory-persistence") β edit $DSH_HOME/settings.yaml; changes apply without a restart
s3 sink: enabled
uploaded parts: 12, dead-lettered: 0
sessions: 1, buffered events: 34, dropped by overflow: 0
last upload: 2026-08-17T16:00:00.000Z
otel sink: disabled
Independently of this command, the plugin's cordis fiber (with its current phase) is always visible in the web UI under Settings β Plugins β ε ¨ι¨.
Configuration reference
config:
sinks:
s3:
enabled: false # master switch
mode: push # 'push' (event stream, legacy default) | 'ship' (tail on-disk artifact)
bucket: '' # required when enabled
prefix: dsh-trajectories # key prefix
region: us-east-1 # signing region
endpoint: ~ # S3-compatible endpoint (OSS, MinIO)
forcePathStyle: ~ # path-style addressing (MinIO, OSS)
credentials: ~ # { accessKeyId, secretAccessKey }; absent = AWS default chain
batchSize: 100 # push: flush trigger, buffered events per session
maxBufferedEvents: 10000 # push: ring cap; oldest dropped with a warning beyond it
maxRetries: 3 # retries after the first attempt
retryBaseDelayMs: 200 # backoff base (retry n waits base * 2^(n-1) + jitter)
deadLetterDir: .dsh/trajectory-deadletter # push: parts whose upload finally failed
root: ~/.dsh/sessions # ship: official jsonl backend's session root ($DSH_HOME/sessions)
pollIntervalMs: 5000 # ship: poll interval for artifact growth
segmentBytes: 262144 # ship: target segment size (never splits a zstd frame)
segmentMaxDelayMs: 60000 # ship: ship a short segment after this without growth
dormantAfterMs: 300000 # ship: dormant after this without change
writerId: ~ # ship: stable writer identity override
otel:
enabled: false
url: '' # full OTLP traces endpoint, e.g. http://jaeger:4318/v1/traces (mutually exclusive with aws)
aws: ~ # { region, url?, service? }; SigV4-signed OTLP to CloudWatch / AgentCore
headers: ~ # extra HTTP headers (auth, β¦)
serviceName: dsh-trajectory-persistence
maxExportBatchSize: 512 # BatchSpanProcessor
scheduledDelayMillis: 5000 # BatchSpanProcessor
shutdownTimeoutMillis: 3000 # dispose drain allowance
AWS S3
config:
sinks:
s3:
enabled: true
bucket: prod-dsh-trajectories
prefix: trajectories
region: ap-southeast-1
# credentials omitted: the AWS default provider chain applies
# (env AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY, shared config, IAM role)
Aliyun OSS (S3-compatible endpoint)
config:
sinks:
s3:
enabled: true
bucket: dsh-trajectories
prefix: trajectories
region: oss-cn-hangzhou
endpoint: https://oss-cn-hangzhou.aliyuncs.com
forcePathStyle: true
credentials:
accessKeyId: ${OSS_ACCESS_KEY_ID}
secretAccessKey: ${OSS_ACCESS_KEY_SECRET}
MinIO is the same shape: endpoint: http://minio:9000, forcePathStyle: true.
Jaeger (OTLP)
Run Jaeger with the OTLP receiver (enabled by default in jaegertracing/all-in-one),
then:
config:
sinks:
otel:
enabled: true
url: http://localhost:4318/v1/traces
AWS CloudWatch / Bedrock AgentCore Observability (SigV4)
Instead of a plain OTLP endpoint, the otel sink can sign each batch with
AWS SigV4 and POST it straight to the
CloudWatch OTLP endpoint
(https://xray.<region>.amazonaws.com/v1/traces, service name xray) β the
same ingest path Bedrock AgentCore Observability uses. Spans land in the
CloudWatch aws/spans log group, and the standard gen_ai.* span attributes
this sink already emits are recognized by the CloudWatch GenAI observability
dashboard (enable Transaction Search in CloudWatch to see them).
config:
sinks:
otel:
enabled: true
aws:
region: us-west-2
# url: ~ # endpoint override β VPC endpoint or another partition
# (e.g. https://xray.cn-north-1.amazonaws.com.cn/v1/traces)
# service: ~ # SigV4 service name override; defaults to xray
aws is mutually exclusive with url, and aws.region is required.
Credentials come from the AWS default provider chain β nothing to put in
the config. For static credentials, use the environment:
export AWS_ACCESS_KEY_ID=<your-access-key-id>
export AWS_SECRET_ACCESS_KEY=<your-secret-access-key>
# export AWS_SESSION_TOKEN=<your-session-token> # only for temporary credentials
Extra entries in headers are merged into the signed request as usual.
OTel Collector β ClickHouse
Point the plugin at the collector:
config:
sinks:
otel:
enabled: true
url: http://localhost:4318/v1/traces
Collector configuration (otel-collector.yml), using the
ClickHouse exporter:
receivers:
otlp:
protocols:
http:
endpoint: 0.0.0.0:4318
grpc:
endpoint: 0.0.0.0:4317
processors:
batch:
send_batch_size: 1000
timeout: 5s
exporters:
clickhouse:
endpoint: tcp://clickhouse:9000?database=otel
ttl: 72h
traces_table_name: otel_traces
service:
pipelines:
traces:
receivers: [otlp]
processors: [batch]
exporters: [clickhouse]
The GenAI span attributes (gen_ai.operation.name, gen_ai.conversation.id,
gen_ai.usage.input_tokens, gen_ai.tool.name, β¦) land in the ClickHouse
trace table's attributes map and are directly queryable, e.g. token usage per
conversation:
SELECT
SpanAttributes['gen_ai.conversation.id'] AS session,
sum(toUInt64OrZero(SpanAttributes['gen_ai.usage.output_tokens'])) AS output_tokens
FROM otel.otel_traces
WHERE SpanAttributes['gen_ai.operation.name'] = 'chat'
GROUP BY session
ORDER BY output_tokens DESC;
Local development with --patch
You can load the plugin straight from a source checkout, without installing it into a profile:
git clone <this-repo> && cd dsh-trajectory-persistence
pnpm install && pnpm build
# overlay patch that points at the local build ($PWD expands to the checkout)
cat > /tmp/traj.patch.yml <<YAML
- insert:
- id: trajectory-persistence
name: file://$PWD/lib/index.js
config:
sinks:
otel:
enabled: true
url: http://localhost:4318/v1/traces
YAML
dsh web --patch /tmp/traj.patch.yml # or: dsh tui / pnpm dsh web from a source checkout
--patch overlays apply after bundle and profile patches, so the row above
wins over any installed trajectory-persistence row.
Development
pnpm install
pnpm build # tsc -> lib/
pnpm test # vitest: span mapping, S3 buffering/flush triggers, retry
Layout:
βββ cordis.patch.yml # bundle layer: inserts the plugin row (sinks disabled)
βββ src/
β βββ index.ts # cordis plugin entry: name/inject/Config/apply, event wiring, dispose drain,
β β # settings namespace + /trajectory-status registration (optional capabilities)
β βββ config.ts # Schemastery schema (sinks.s3 / sinks.otel) + cross-field validateConfig
β βββ sinks.ts # TrajectorySinks: hot-swappable sinks, per-sink rebuild, status snapshot
β βββ jsonl.ts # jsonl-persistence-compatible header line, projectKey, encodeSegment
β βββ sink-utils.ts # EventBuffer (ring) + BufferedPartSink (batch/retry/dead-letter, stats())
β βββ s3-sink.ts # S3TrajectorySink (push mode): S3 transport (uploader + key layout) over BufferedPartSink
β βββ shipper.ts # S3ShipperSink (ship mode): read-only artifact tail, zstd frame segments, manifest watermark
β βββ manifest.ts # ship mode: _manifest.json (watermark/segments/writerId) + per-machine writer id
β βββ ship-state.ts # ship mode: local per-session ship-state (offset, dormancy, conflicted)
β βββ zstd-scan.ts # ship mode: complete zstd frame scanner (torn tail never ships)
β βββ sync-down.ts # restore local artifacts from shipped segments (prefix-append, diverge-refuse, force backup)
β βββ cli.ts # `dsh-trajectory-persistence sync-down` command line
β βββ otel-sink.ts # GenAISpanMapper (pure mapping) + OtelTrajectorySink (OTLP pipeline)
β βββ sigv4-otlp-exporter.ts # SigV4-signed OTLP exporter: CloudWatch / AgentCore Observability
β βββ retry.ts # exponential backoff helper
βββ test/ # vitest: otel-map, sigv4-otlp-exporter, config, s3-sink, shipper, manifest,
# ship-state, zstd-scan, sync-down, sinks (rebuild), retry, integration (real cordis ctx)
Notes & caveats
- Imports. Only published packages are used:
@deepseek-ai/cordis,@deepseek-ai/dsh-session(type-only, forSession/SessionEvent/SessionHeaderand its cordis event augmentation),@deepseek-ai/schemastery,@deepseek-ai/dsh-settings(runtime, forinstallSettingsSection/settingsNamespace), and@deepseek-ai/dsh-commands(type-only: the command registry is reached throughctx.commands, so no runtime import is needed). The header-line / path-encoding helpers of@deepseek-ai/dsh-session-persistence-jsonlare reimplemented insrc/jsonl.ts(kept byte-compatible): the published jsonl package loads a native zstd binding (koffi) at import time, which a remote-only sink must not require. If the monorepo changes that format, updatesrc/jsonl.ts. - No model identity on events. The
chatspan'sgen_ai.request.modelcomes from the most recentrequest/contextevent seen in the session; sessions that never log one produce spans without the model attribute. - Observation, not authoring. This plugin never writes through
Session.append; it is purely a consumer of the firehose. Thesession/flushlistener returns the sink's upload promise, so the harness's awaited durability checkpoint covers this sink: a settled flush means the buffered trajectory has been uploaded (or dead-lettered after exhausting retries). Turn latency is unaffected β only the checkpoint itself waits. - Requires Node.js β₯ 22 (matching the harness).
Documentation
The full documentation site lives at https://aws-cnug-o11y.github.io/dsh-trajectory-persistence/:
- Getting Started β install, first trace with a local Jaeger
- Configuration Reference β every field, defaults, validation rules, hot-reload
- S3 / OSS Sink β push mode: JSONL part layout, ring buffer, retry & dead-letter
- Ship & Sync β ship mode: zstd frame segments + manifest,
sync-downrestore, machine switching - OTel GenAI Sink β event β span mapping, Jaeger / Collector / Langfuse
- AWS CloudWatch & AgentCore β SigV4 delivery, China endpoints, Transaction Search
- Development β build/test, sink architecture, adding a sink
The site is built with VitePress from docs/
(pnpm docs:dev / pnpm docs:build) and deployed by
.github/workflows/deploy-docs.yml on every push to main that touches
docs/**. Repository admins: under Settings β Pages, set Source to
GitHub Actions for the deployment to go live.