diff --git a/Cargo.toml b/Cargo.toml index 5d9f64d..b65795a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,7 +60,8 @@ wreq = { version = "6.0.0-rc.31", default-features = false, features = ["brotli" wreq-util = { version = "3.0.0-rc.14", features = ["emulation-compression"], optional = true } uuid = { version = "1.23.4", features = ["v4", "v5"] } wasm-bindgen = { version = "0.2.126", optional = true } -# Decoding the rollouts Codex compresses; the library Codex writes them with. +# Decoding the rollouts Codex compresses and the session logs DeepSeek Harness +# compresses; the library both write them with. zstd = { version = "0.13.3", default-features = false, optional = true } [target.'cfg(target_os = "macos")'.dependencies] @@ -104,7 +105,7 @@ required-features = ["search"] [features] # Explicit cloud Cowork reads reuse the Claude Desktop transport. cowork_remote = ["claude_chat"] -default = ["opencode", "hermes", "claude_chat", "cowork_remote", "chatgpt", "search", "codex_zstd"] +default = ["opencode", "hermes", "claude_chat", "cowork_remote", "chatgpt", "search", "codex_zstd", "dsh_zstd"] # The OpenCode harness store reads OpenCode's SQLite database. The OpenCode # codec itself is always available; only the on-disk store needs rusqlite. opencode = ["dep:rusqlite"] @@ -140,6 +141,10 @@ chatgpt = [ # leaves the compressed ones out of discovery; with it, both list and load. # zstd is a C library, so the wasm build goes without. codex_zstd = ["dep:zstd"] +# DeepSeek Harness compresses session logs to `session.jsonl.zstd` by default. +# Without this feature the dsh store still reads `compression: 'none'` roots +# (`session.jsonl`) and leaves compressed sessions out of discovery. +dsh_zstd = ["dep:zstd"] # Fuzzy/substring search over transcripts (`txcript::search`). nucleo-matcher # is pure Rust (helix's matcher) and compiles to wasm32; combine with `wasm` # to expose the bindings. diff --git a/README.md b/README.md index 9727819..51a25f6 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,7 @@ Each name links to its format documentation. Use the ID with `--from` and `--wit | [Cowork](docs/formats/cowork.md) | `cowork` | Yes | Yes | | [Grok CLI](docs/formats/grok.md) | `grok` | Yes | Yes | | [Grok Bot](docs/formats/grok-bot.md) | `grok_bot` | Yes | Via local gateway | +| [DeepSeek Harness](docs/formats/dsh.md) | `dsh` | Yes | Yes | | [fx](docs/formats/fx.md) | `fx` | Yes | Yes | | [Antigravity](docs/formats/antigravity.md) | `antigravity` | Yes | Yes | | [Hermes Agent](docs/formats/hermes.md) | `hermes` | Yes | No | diff --git a/cli/src/lib.rs b/cli/src/lib.rs index 30feb8b..5e85e2f 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -68,7 +68,7 @@ pub mod mcp; mod pager; mod view; -pub const HARNESSES: &str = "harnesses: claude_code, claude_chat, chatgpt, codex, opencode, pi, campfire, cursor, cursor_desktop, grok, fx, hermes, \ +pub const HARNESSES: &str = "harnesses: claude_code, claude_chat, chatgpt, codex, opencode, pi, campfire, cursor, cursor_desktop, grok, dsh, fx, hermes, \ amp, antigravity, simple, cowork, cowork_remote"; /// The `txcript` binary's command line. @@ -1172,6 +1172,13 @@ mod identity_tests { assert!(error.contains("pull-only")); } + #[test] + fn dsh_can_be_continued_in_place() { + // Only the two pull-only remote sources are refused here. dsh is a + // local store txcript writes, so it takes the normal write path. + assert!(ensure_resumable_source(HarnessId::Dsh, HarnessId::Dsh).is_ok()); + } + #[test] fn resume_alias_is_accepted_for_continue_command() { use clap::Parser; @@ -1318,6 +1325,7 @@ mod style { HarnessId::CursorDesktop => "\x1b[96m", // bright cyan HarnessId::Grok => "\x1b[37m", // white HarnessId::GrokBot => "\x1b[97m", // bright white + HarnessId::Dsh => "\x1b[38;5;43m", // teal HarnessId::Fx => "\x1b[38;5;39m", // azure HarnessId::Hermes => "\x1b[93m", // bright yellow HarnessId::Amp => "\x1b[95m", // bright magenta diff --git a/docs/formats/README.md b/docs/formats/README.md index 633fdde..bbf83c5 100644 --- a/docs/formats/README.md +++ b/docs/formats/README.md @@ -51,6 +51,7 @@ than none. | [amp.md](amp.md) | Amp (Sourcegraph) | `src/harness/amp.rs` | | [grok.md](grok.md) | Grok CLI (xAI) | `src/harness/grok.rs` | | [grok-bot.md](grok-bot.md) | Grok Bot (desktop assistant) | `src/harness/grok_bot.rs` | +| [dsh.md](dsh.md) | DeepSeek Harness (`dsh`) | `src/harness/dsh.rs` | | [fx.md](fx.md) | fx (Vercel) | `src/harness/fx.rs` | | [hermes.md](hermes.md) | Hermes Agent | `src/harness/hermes.rs` | | [antigravity.md](antigravity.md) | Antigravity (Google) | `src/harness/antigravity.rs` | diff --git a/docs/formats/dsh.md b/docs/formats/dsh.md new file mode 100644 index 0000000..dc8cb36 --- /dev/null +++ b/docs/formats/dsh.md @@ -0,0 +1,223 @@ +# DeepSeek Harness (`dsh`) + +DeepSeek Harness stores sessions under: + +```text +$DSH_HOME/sessions/ # default ~/.dsh/sessions + ----/ # or _no-cwd/ + / + session.v4.jsonl.zstd # current: concatenated checksummed Zstandard frames + session.v4.jsonl # only when compression: 'none' + session.jsonl.zstd # older generations (v0 has no `.vN`) may remain + session.lock +``` + +Home resolution is configured path, then `$DSH_HOME`, then `~/.dsh`. An empty +`$DSH_HOME` is treated as unset. + +Each format generation has its own file name, `session[.vN].jsonl[.zstd]`, +where a missing `.vN` is version 0. When dsh opens an older session for +writing, it migrates it and publishes the new generation **beside** the old +one; the old file is kept byte-identical. The numerically highest generation is +the live log, and that is the one txcript reads. Lock files and in-flight +`.tmp` files are not logs. + +Reading `.jsonl.zstd` needs txcript's `dsh_zstd` feature, on by default. +Without it (and in the wasm build), compressed sessions are left out of +discovery and loading one fails; uncompressed `.jsonl` logs still work. + +## Native representation + +`txcript::harness::dsh::DshSession` retains the first JSONL line as `header` +and every subsequent line as raw JSON values. Packed storage rows +(`text-chunks`, `reasoning-chunks`, `tool-call-chunks`, v0–v1 only) and +unknown event types stay in the native body so a text load/render round trip +does not drop bookkeeping the Common projection does not understand. + +txcript reads generations 0 through 4 directly rather than migrating them. +The differences that matter to the Common projection are small: + +| Version | Change that affects the reader | +| --- | --- | +| v1 | file renamed `session.v1.jsonl`; layout otherwise unchanged | +| v2 | top-level `assistant/chunk` and packed chunk rows gone (the stream is embedded in `assistant/message`); header `seedLength` replaced by `isSeeded` | +| v3 | system prompt becomes a `system/message` surface node; replace endpoints renamed `start`/`end` → `startSeq`/`endSeq` | +| v4 | `tool/result` becomes a tool-role message (no `tool-result` wrapper); message sources become direct `kind`s; `developer/message` surface node | + +### Header + +The first line is tagged `type: "session"` and carries `version`, `id`, +`createdAt` (epoch milliseconds), `delegationDepth`, `isSeeded` (v2+; +`seedLength` before), and optional `cwd`, `parentSession`, `origin`, and +`agentPreset`. + +### Events + +Each event is `{ type, seq, time, data, ... }`. Surface events +(`system/message`, `developer/message`, `user/message`, `assistant/message`, +`tool/result`) carry `surfaceOp` — `"append"`, or a replace op naming an +inclusive span — and, except on `assistant/message`, may carry +`sourceEventSeqs`. Log-only events (turn/step markers, `tool/call`, +`request/header`, inbox splices, delivery and title bookkeeping, …) never enter +the Common conversation. + +## Common projection + +| dsh event | Common representation | +| --- | --- | +| `user/message` with text parts | user text message, unless it is environment context (below) | +| `assistant/message` `reasoning` / `text` / `tool-call` | thinking / text / tool-use | +| `assistant/message` `usage` / `interrupted` | `Usage` / `StopReason::Aborted` | +| `tool/result` | user tool-result (`isError` kept), from the v4 tool-role message or the earlier wrapper block | +| `system/message`, `developer/message` | nothing, but they hold surface positions | + +The ordered surface is rebuilt before projection. Surface nodes are tracked by +event `seq`, because log-only events sit between them and a node's seq is not +its surface position. A replace op names the inclusive span it shadows — both +endpoints must be on the current surface — and substitutes the replacing node +for that whole run; a range txcript cannot resolve shadows nothing. System and +developer messages are surface nodes too, so a span is counted across them even +though they produce no Common message. + +dsh injects its own environment into the conversation as extra `user/message`s: +runtime and sandbox snapshots, the skill catalog, instruction files, clocks. +Those sources (`runtime-context`, `skill-catalog`, `agent-instructions`, +`time-context`, `tmux-context`, `repeat-tool-reminder`, `model-selection`, +`plan-mode`; before v4, the same names as `{ kind: "plugin", plugin }`, with the +system-prompt plugin's messages standing for `runtime-context`) are not +projected — they describe dsh, not the conversation, and would mislead another +harness the session continues in. Every other source is kept, unknown ones +included: a `compact-checkpoint` message is the only remaining copy of the +history it replaced. + +`usage` maps `inputTokens`/`outputTokens` and the optional +`cacheReadTokens`/`cacheWriteTokens` onto Common's `Usage`. `reasoningTokens` +has no Common counterpart and survives in the native body only. Shadowed +surface nodes likewise stay in the native body, so a text round trip keeps the +full log even though Common shows the model-visible surface. + +## Lineage + +`parentSession` is the id of the session this one was seeded from. A subagent +run also records `origin: "subagent"` and becomes a Spawn `Meta.lineage`; a +user fork records no origin and becomes a Fork. dsh resumes a session under its +own id, so there is no continuation link. Lineage is filled at discovery as +well as on load. + +## Store capabilities + +The store reads and writes. dsh ships no session-import command, but it does +not need one: it discovers sessions by walking its root, so writing one is a +matter of reproducing the layout it scans for. + +What makes that exact rather than approximate is dsh's validation. Three of its +checks fail the **entire** listing rather than skipping the one bad session, so +each is a hard requirement on the writer: + +| Check | Requirement | +| --- | --- | +| `assertZstdHeaderFrame` | the first Zstandard frame decodes to exactly one line — the header | +| `assertStoredIdentity` | the header's own `id` and `cwd` name the path the log was found at | +| `checkRootEncoding` | one root never mixes `.jsonl` with `.jsonl.zstd` | +| generation selection | the file name's `.vN` matches the header's `version` | + +A duplicate session id across two project directories is rejected the same way. + +### Layout derivation + +- **Project directory** — `----`, where `key` collapses each run of `/`, + `\`, or `:` to a single `-`, keeps `[A-Za-z0-9._-]`, escapes every other + UTF-16 code unit as `~XXXX`, strips leading dashes, falls back to `root` if + nothing remains, and truncates to 251 characters. A session with no cwd goes + under `_no-cwd`. +- **Session directory** — the id under the same `~XXXX` escape, with `.` and + `..` special-cased whole. This is what contains a traversing id: escaping the + separators turns `../../evil` into the literal directory + `..~002F..~002Fevil`. + +Escaping operates on UTF-16 code units, not Unicode scalars, which is what +makes it injective over lone surrogates. + +### What `save` writes + +`///session.v4.jsonl.zstd` — a checksummed Zstandard +frame holding the header line, then one holding the event lines. A native dsh +body keeps its own version, and the file is named for it (`session.jsonl.zstd` +for v0); dsh migrates an older generation itself when it next opens the +session. Any other generation in the directory is removed, since the saved log +replaces the whole session and a newer one beside it would be the one read. The +header is stamped with the id and cwd that built the path, because a copy given +a new identity would otherwise keep pointing at the original's. + +The physical encoding follows whatever the root already uses; only an empty +root falls back to dsh's own default of `zstd`. Re-saving a session whose cwd +changed removes the copy under the old project directory, since leaving it +would be the duplicate-id corruption above — but only after the replacement is +on disk, so a failed save leaves the previous copy in place. The log itself is +written to a temporary file and renamed into place. + +### From Common + +A Common conversation is written as format v4, laid out as the turn and step +lifecycle dsh's own reader checks. Each human message opens a turn and its +first step. A model response joins the open step, or opens the next one once a +tool result has answered it. Each tool call is recorded as started +(`tool/call`) immediately before its result, which cites that call in +`sourceEventSeqs`. + +A few events exist only because dsh requires them: + +- **System head** — the first step starts with an empty `system/message`. dsh + appends its current system prompt when it resumes a session, and accepts + that only behind a system message heading the surface; without one, the log + dsh itself extends fails its next load. +- **Unanswered calls** — a tool call with no result in Common is closed the + way dsh closes an interrupted one: an error result with its exact + `TOOL_NOT_STARTED` repair id and wording, and a turn ending `interrupted`. + A result whose call is not open in the current step has nothing dsh would + accept it against, and is left out. +- **Model source** — every assistant message names a provider and model. + Common records no provider, so it is `txcript`; the model is the message's, + else the session's, else `unknown`. +- **Stream** — an assistant message embeds its provider stream, which dsh + replays against the message when it verifies a log. Common has no stream, + and an empty one is the form dsh skips. + +A Common title is recorded as a `user` title: dsh's `fallback` kind claims the +title was derived from a cited prompt. + +`delete` removes the session directory. dsh's persistence seam has no delete +API, but it also keeps no index — an absent directory is simply not scanned. +Before removing anything, `delete` requires a session log and refuses any path +that doesn't resolve, symlinks included, to `//`. + +The native resume command documented for a TUI profile is +`dsh --profile tui --resume `. + +## Provenance + +Open source. Layout, event vocabulary, and the per-version changes follow the +DeepSeek Harness packages `@deepseek-ai/dsh-session`, +`@deepseek-ai/dsh-session-persistence-jsonl`, and the adjacent migration +packages `@deepseek-ai/dsh-session-format-v0-to-v1` through `-v3-to-v4` +(format version 4 at 0.2.0-rc.2; the project still marks the format a +developer preview and warns of compatibility-breaking changes). + +The reader was checked against sessions recorded by dsh 0.2.0-rc.2 — a +tool-using session resumed for a second turn, and a subagent run with its +parent — and against a session recorded by dsh around 2026-08-14 in format v0, +read both as v0 and after the official catalog migrated it to v4: the two +generations project to identical Common messages. + +The writer was checked against the official format catalog under its +strictest settings (`recovery: 'strict'`, `validation: 'current'`) with +sessions converted from Claude Code, Codex, and OpenCode, the largest about +14,000 events. A converted session was then resumed twice with +`dsh --profile headless --session-id`: dsh answered from the imported history, +and the log it extended still passed the same check after each turn. An earlier +layout without the system head passed the first resume and then failed the +second — the reason the head exists. Compressing the log as one frame instead +of two — which txcript itself still reads — makes dsh's own check fail with +`first frame is not exactly one header line`. + +Last verified: 2026-10-15, against `@deepseek-ai/dsh` 0.2.0-rc.2. diff --git a/docs/usage.md b/docs/usage.md index e9ad6c8..8c6f32c 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -123,7 +123,7 @@ txcript = "0.14" # txcript = { version = "0.14", default-features = false } ``` -Default features: `opencode` (the SQLite stores: OpenCode, both Cursors, Antigravity), `hermes`, `claude_chat`, `cowork_remote`, `chatgpt`, `search`, and `codex_zstd` (the rollouts Codex compresses to `.jsonl.zst`). +Default features: `opencode` (the SQLite stores: OpenCode, both Cursors, Antigravity), `hermes`, `claude_chat`, `cowork_remote`, `chatgpt`, `search`, `codex_zstd` (the rollouts Codex compresses to `.jsonl.zst`), and `dsh_zstd` (the session logs DeepSeek Harness compresses to `.jsonl.zstd`). Three layers, smallest to largest: @@ -225,7 +225,7 @@ writeFileSync("session.jsonl", convert(input, "codex", "claude_code")); const common = JSON.parse(toCommon(input, "codex")); // { meta, messages } const pi = fromCommon(JSON.stringify(common), "pi"); -harnesses(); // ["claude_code","claude_chat","cowork_remote","chatgpt","codex","opencode","pi","campfire","cursor","cursor_desktop","grok","grok_bot","fx","hermes","amp","antigravity","simple","cowork"] +harnesses(); // ["claude_code","claude_chat","cowork_remote","chatgpt","codex","opencode","pi","campfire","cursor","cursor_desktop","grok","grok_bot","dsh","fx","hermes","amp","antigravity","simple","cowork"] ``` Text-in / text-out: `input` is the source harness's native session text and the result is the target's. Invalid harness names or unparseable input throw a JS `Error`. @@ -255,6 +255,7 @@ const matches = JSON.parse(index.query(JSON.stringify({ pattern: "relay bug" })) | `cursor_desktop` | JSON dump of the session's `state.vscdb` rows | | `grok` | JSON bundle of the session directory's files | | `grok_bot` | agent-transcript JSONL (`role`/`message` envelopes) | +| `dsh` | JSON object of the session header and its decompressed JSONL events | | `fx` | JSON bundle of the session directory's files | | `hermes` | `hermes sessions export` JSON object | | `amp` | `amp threads export` JSON | diff --git a/examples/search_bench.rs b/examples/search_bench.rs index 19275bd..f33e00f 100644 --- a/examples/search_bench.rs +++ b/examples/search_bench.rs @@ -10,7 +10,7 @@ use std::time::Instant; -use txcript::harness::{campfire, claude_code, codex, cursor, grok, pi}; +use txcript::harness::{campfire, claude_code, codex, cursor, dsh, grok, pi}; use txcript::search::{DocKey, Index, Origin, Query}; use txcript::{Codec, Common, HarnessId, Store, Transcript}; @@ -62,6 +62,13 @@ fn main() { &mut loaded, &mut failed, ); + load_files::( + HarnessId::Dsh, + dsh::DshStore::default_root(), + &mut index, + &mut loaded, + &mut failed, + ); let build = started.elapsed(); println!( diff --git a/src/harness/dsh.rs b/src/harness/dsh.rs new file mode 100644 index 0000000..5329f97 --- /dev/null +++ b/src/harness/dsh.rs @@ -0,0 +1,1614 @@ +//! `dsh` sessions: `$DSH_HOME/sessions` (default `~/.dsh/sessions`). +//! +//! Official persistence writes one append-only JSONL log per session, Zstandard +//! framed by default, named for its format generation +//! (`session[.vN].jsonl[.zstd]`; currently `session.v4.jsonl.zstd`). The first +//! line is a `type: session` header; the rest are `SessionEvent` records (and, +//! in v0–v1, packed `*-chunks` rows). The reader understands generations 0 +//! through 4 directly. +//! +//! dsh finds sessions by walking its root, so no import command is needed — +//! but it also validates what it finds, and three of its checks fail the +//! *whole* listing rather than skipping one session: a first Zstandard frame +//! that is not exactly the header line, a header whose id and cwd do not name +//! the path it was found at, and a root that mixes `.jsonl` with +//! `.jsonl.zstd`. Writing therefore reproduces dsh's own layout exactly. +//! +//! Native body retains the header and every log line as raw JSON, including +//! packed chunk rows and unknown event types. The Common projection uses the +//! ordered surface (`system/message`, `developer/message`, `user/message`, +//! `assistant/message`, `tool/result`), minus the environment context dsh +//! injects as user messages. + +use std::collections::HashMap; +use std::fs; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use uuid::Uuid; + +use crate::common::{ + Block, Lineage, Message, Meta, Relation, Role, StopReason, Tool, ToolOutput, Usage, +}; +use crate::error::{Error, Result}; +use crate::harness::jsonl; +use crate::transcript::{Codec, Common, Discovered, Harness, Saved, Store, TextCodec, Transcript}; + +/// On-disk format version stamped into every header txcript writes from +/// Common. Official readers migrate older versions and refuse newer ones; +/// txcript still loads the native body so unknown future logs are not silently +/// dropped. +const SESSION_FORMAT_VERSION: u64 = 4; + +/// The `dsh` harness marker. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Dsh; + +impl Harness for Dsh { + const NAME: &'static str = "dsh"; + type Body = DshSession; +} + +/// Header line plus the remaining JSONL records, kept raw. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct DshSession { + pub header: Value, + #[serde(default)] + pub events: Vec, +} + +impl TextCodec for Dsh { + fn from_text(text: &str) -> Result> { + let body: DshSession = serde_json::from_str(text)?; + let meta = meta_from_body(&body); + Ok(Transcript::new(meta, body)) + } + + fn to_text(transcript: &Transcript) -> Result { + Ok(serde_json::to_string_pretty(&transcript.body)?) + } +} + +impl Codec for Dsh { + fn to_common(transcript: &Transcript) -> Result> { + Ok(Transcript::new( + transcript.meta.clone(), + events_to_messages(&transcript.body.events, transcript.meta.timestamp), + )) + } + + fn from_common(transcript: &Transcript) -> Result> { + Ok(Transcript::new( + transcript.meta.clone(), + body_from_common(transcript), + )) + } +} + +/// Read/write access to `dsh` session directories. +#[derive(Debug, Clone)] +pub struct DshStore { + pub sessions_dir: PathBuf, +} + +impl DshStore { + pub fn new(path: impl Into) -> Self { + Self { + sessions_dir: path.into(), + } + } + + /// `$DSH_HOME/sessions`, then `~/.dsh/sessions`. Empty `$DSH_HOME` is unset. + #[must_use] + pub fn default_root() -> Option { + std::env::var_os("DSH_HOME") + .filter(|value| !value.is_empty() && !value.to_string_lossy().trim().is_empty()) + .map(|home| Self::new(PathBuf::from(home).join("sessions"))) + .or_else(|| super::home_dir().map(|home| Self::new(home.join(".dsh").join("sessions")))) + } +} + +impl Store for DshStore { + type H = Dsh; + type Ref = PathBuf; + + fn discover(&self) -> Result>> { + let mut found = Vec::new(); + let Ok(projects) = fs::read_dir(&self.sessions_dir) else { + return Ok(found); + }; + for project in projects.flatten().map(|entry| entry.path()) { + if !project.is_dir() { + continue; + } + let Ok(sessions) = fs::read_dir(&project) else { + continue; + }; + for session_dir in sessions.flatten().map(|entry| entry.path()) { + if !session_dir.is_dir() { + continue; + } + let Some(log) = log_path(&session_dir) else { + continue; + }; + let Ok(body) = load_body(&log) else { + continue; + }; + let mut meta = meta_from_body(&body); + if meta.id.is_empty() { + meta.id = jsonl::file_id(&session_dir); + } + found.push(Discovered { + meta, + reference: session_dir, + }); + } + } + Ok(found) + } + + fn load(&self, reference: &PathBuf) -> Result> { + let log = log_path(reference).ok_or_else(|| Error::Malformed { + harness: Dsh::NAME, + detail: format!("no session log in {}", reference.display()), + })?; + let body = load_body(&log)?; + let mut meta = meta_from_body(&body); + if meta.id.is_empty() { + meta.id = jsonl::file_id(reference); + } + Ok(Transcript::new(meta, body)) + } + + fn save(&self, transcript: &Transcript) -> Result> { + let id = session_id(transcript); + let cwd = session_cwd(transcript); + let dir = self + .sessions_dir + .join(project_key(cwd.as_deref())) + .join(encode_segment(&id)?); + + let mut body = transcript.body.clone(); + stamp_identity(&mut body.header, &id, cwd.as_deref()); + + let compressed = self.root_compresses(); + // dsh rejects a log whose file name and header disagree on the + // version, so a native body is written under its own generation. + let version = body + .header + .get("version") + .and_then(Value::as_u64) + .unwrap_or(0); + let name = log_file_name(version, compressed); + let bytes = if compressed { + encode_log(&body)? + } else { + plain_log(&body)?.into_bytes() + }; + fs::create_dir_all(&dir)?; + // Write beside the log and rename over it, so a failed write neither + // truncates a previous log at this path nor leaves a partial one. The + // temporary name carries no `.jsonl`, so dsh never mistakes it for a + // log of either encoding. + let tmp = dir.join(".txcript-save.tmp"); + if let Err(error) = fs::write(&tmp, bytes).and_then(|()| fs::rename(&tmp, dir.join(&name))) + { + let _ = fs::remove_file(&tmp); + return Err(error.into()); + } + // The saved log replaces the whole session. A newer generation left + // beside it would be the one dsh (and txcript) read, and the opposite + // encoding is the same poison as one elsewhere in the root. + retire_other_generations(&dir, &name); + // Only now that the replacement is on disk: a copy under another + // project directory is the last one left if this save had failed. + self.retire_other_copies(&id, &dir); + Ok(Saved { id, reference: dir }) + } + + fn delete(&self, reference: &PathBuf) -> Result<()> { + if log_path(reference).is_none() { + return Err(Error::Malformed { + harness: Dsh::NAME, + detail: format!("not a dsh session directory: {}", reference.display()), + }); + } + // Resolve symlinks before checking containment, and require the exact + // `/` depth so neither the root nor a project + // directory can be removed wholesale. + let canon = reference.canonicalize()?; + let root = self.sessions_dir.canonicalize()?; + let contained = canon + .strip_prefix(&root) + .is_ok_and(|rest| rest.components().count() == 2); + if !contained { + return Err(Error::Malformed { + harness: Dsh::NAME, + detail: format!( + "refusing to delete outside the sessions root: {}", + reference.display() + ), + }); + } + Ok(fs::remove_dir_all(canon)?) + } + + fn fingerprints(&self, refs: &[PathBuf]) -> Result> { + let mut output = HashMap::with_capacity(refs.len()); + for reference in refs { + let cursor = + log_path(reference).map_or_else(String::new, |path| file_fingerprint(&path)); + output.insert(reference.to_string_lossy().into_owned(), cursor); + } + Ok(output) + } +} + +impl DshStore { + /// Remove copies of `id` filed under a different project directory. + /// + /// A session's project directory is keyed by its cwd, so a re-save after a + /// cwd change lands somewhere new. dsh treats one id under two project + /// directories as corruption and fails its whole listing, so the stale copy + /// cannot stay. Best effort: a copy that cannot be removed is not worth + /// failing an otherwise good write over. + fn retire_other_copies(&self, id: &str, keep: &Path) { + let Ok(encoded) = encode_segment(id) else { + return; + }; + let Ok(projects) = fs::read_dir(&self.sessions_dir) else { + return; + }; + for project in projects.flatten().map(|entry| entry.path()) { + let candidate = project.join(&encoded); + if candidate != keep && log_path(&candidate).is_some() { + let _ = fs::remove_dir_all(&candidate); + } + } + } + + /// Whether a new log should be Zstandard framed. + /// + /// dsh refuses to list *any* session in a root that holds both `.jsonl` + /// and `.jsonl.zstd` artifacts — under either configuration, not just the + /// mismatched one. So the root's existing artifacts decide, and only an + /// empty root falls back to dsh's own default of `zstd`. + fn root_compresses(&self) -> bool { + let Ok(projects) = fs::read_dir(&self.sessions_dir) else { + return true; + }; + for project in projects.flatten().map(|entry| entry.path()) { + let Ok(sessions) = fs::read_dir(&project) else { + continue; + }; + for session_dir in sessions.flatten().map(|entry| entry.path()) { + if let Some(log) = log_path(&session_dir) { + return log.extension().is_some_and(|ext| ext == "zstd"); + } + } + } + true + } +} + +/// `meta` is the canonical identity, not the retained native header: a copy is +/// given a new id and timestamp before it is written, and the header it +/// inherited still names the original. +fn session_id(transcript: &Transcript) -> String { + if !transcript.meta.id.is_empty() { + return transcript.meta.id.clone(); + } + transcript + .body + .header + .get("id") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + .map_or_else( + || format!("session-{}", Uuid::new_v5(&NS, b"empty-id")), + String::from, + ) +} + +fn session_cwd(transcript: &Transcript) -> Option { + transcript + .meta + .cwd + .clone() + .or_else(|| { + transcript + .body + .header + .get("cwd") + .and_then(Value::as_str) + .map(String::from) + }) + .filter(|cwd| !cwd.is_empty()) +} + +/// dsh cross-checks a log it finds against the path its own header would +/// name, and a mismatch fails the whole listing rather than skipping the one +/// session. So the header is stamped with the id and cwd that built the path. +fn stamp_identity(header: &mut Value, id: &str, cwd: Option<&str>) { + let Some(object) = header.as_object_mut() else { + return; + }; + object.insert("id".to_string(), json!(id)); + match cwd { + Some(cwd) => { + object.insert("cwd".to_string(), json!(cwd)); + } + // `cwd: null` is not the same as an absent `cwd`: dsh keys `_no-cwd` + // off `undefined`, and a null would not round-trip through its header + // schema. + None => { + object.remove("cwd"); + } + } +} + +/// dsh's `encodeSegment`: injective over UTF-16 code units, which is what +/// neutralizes `..`, absolute paths, and separators in an id that is otherwise +/// an unvalidated string. +fn encode_segment(raw: &str) -> Result { + use std::fmt::Write as _; + + if raw.is_empty() { + return Err(Error::Unconvertible { + harness: Dsh::NAME, + detail: "a dsh session id cannot be empty".to_string(), + }); + } + match raw { + "." => return Ok("~002E".to_string()), + ".." => return Ok("~002E~002E".to_string()), + _ => {} + } + let mut out = String::with_capacity(raw.len()); + for unit in raw.encode_utf16() { + match char::from_u32(u32::from(unit)) { + Some(ch) if is_path_safe(ch) => out.push(ch), + _ => { + let _ = write!(out, "~{unit:04X}"); + } + } + } + Ok(out) +} + +/// dsh's `projectKey`: separators collapse to a single `-`, unsafe units take +/// the same `~XXXX` escape, and the result is wrapped in `--`. Deliberately +/// lossy — it groups sessions for humans, and the session id below it is what +/// identifies them. +fn project_key(cwd: Option<&str>) -> String { + use std::fmt::Write as _; + + let Some(cwd) = cwd else { + return "_no-cwd".to_string(); + }; + let mut readable = String::with_capacity(cwd.len()); + let mut separator_run = false; + for unit in cwd.encode_utf16() { + match char::from_u32(u32::from(unit)) { + Some('/' | '\\' | ':') => { + if !separator_run { + readable.push('-'); + } + separator_run = true; + } + Some(ch) if is_path_safe(ch) => { + readable.push(ch); + separator_run = false; + } + _ => { + let _ = write!(readable, "~{unit:04X}"); + separator_run = false; + } + } + } + // Every retained unit is ASCII — safe characters are, and escapes are — + // so truncating by byte matches dsh's truncation by UTF-16 unit. + let trimmed = readable.trim_start_matches('-'); + let name = if trimmed.is_empty() { "root" } else { trimmed }; + format!("--{}--", &name[..name.len().min(251)]) +} + +fn is_path_safe(ch: char) -> bool { + ch != '~' && (ch.is_ascii_alphanumeric() || matches!(ch, '.' | '_' | '-')) +} + +/// The uncompressed artifact: the header line, then one line per event. +fn plain_log(body: &DshSession) -> Result { + Ok(format!("{}{}", header_line(body)?, event_lines(body)?)) +} + +/// The Zstandard artifact: one frame holding exactly the header line, then one +/// holding the event lines. dsh checks that the first frame decodes to a +/// single line, so the split is part of the format, not a writer's choice. +fn encode_log(body: &DshSession) -> Result> { + let mut out = compress_frame(header_line(body)?.as_bytes())?; + out.extend(compress_frame(event_lines(body)?.as_bytes())?); + Ok(out) +} + +fn header_line(body: &DshSession) -> Result { + Ok(format!("{}\n", serde_json::to_string(&body.header)?)) +} + +fn event_lines(body: &DshSession) -> Result { + let lines = body + .events + .iter() + .map(serde_json::to_string) + .collect::, _>>()? + .join("\n"); + Ok(format!("{lines}\n")) +} + +#[cfg(feature = "dsh_zstd")] +fn compress_frame(bytes: &[u8]) -> Result> { + use std::io::Write as _; + + // dsh compresses every frame with the content checksum enabled, and + // validates it on read. + let mut encoder = zstd::stream::write::Encoder::new(Vec::new(), 0)?; + encoder.include_checksum(true)?; + encoder.write_all(bytes)?; + Ok(encoder.finish()?) +} + +#[cfg(not(feature = "dsh_zstd"))] +fn compress_frame(_bytes: &[u8]) -> Result> { + Err(Error::Malformed { + harness: Dsh::NAME, + detail: "writing a compressed dsh session log needs txcript's `dsh_zstd` feature" + .to_string(), + }) +} + +/// The live log in a session directory. dsh names each format generation +/// `session[.vN].jsonl[.zstd]` (no `.vN` is version 0) and, when it migrates +/// a session, publishes the new generation beside the old one rather than +/// replacing it, so the numerically highest generation is the current one. +fn log_path(session_dir: &Path) -> Option { + fs::read_dir(session_dir) + .ok()? + .flatten() + .filter_map(|entry| { + let name = entry.file_name().into_string().ok()?; + let generation = log_generation(&name)?; + let path = entry.path(); + let compressed = path.extension().is_some_and(|ext| ext == "zstd"); + path.is_file().then_some(((generation, compressed), path)) + }) + .max_by_key(|(key, _)| *key) + .map(|(_, path)| path) +} + +/// The format generation a canonical log file name carries, or `None` for +/// anything else in the directory (the lock file, in-flight `.tmp` files). +/// dsh's canonical generation name: version 0 carries no `.vN`. +fn log_file_name(version: u64, compressed: bool) -> String { + let stem = if version == 0 { + "session.jsonl".to_string() + } else { + format!("session.v{version}.jsonl") + }; + if compressed { + format!("{stem}.zstd") + } else { + stem + } +} + +/// Remove every log in `dir` other than `keep`. Best effort, like +/// [`DshStore::retire_other_copies`]. +fn retire_other_generations(dir: &Path, keep: &str) { + let Ok(entries) = fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let Ok(name) = entry.file_name().into_string() else { + continue; + }; + if name != keep && log_generation(&name).is_some() { + let _ = fs::remove_file(entry.path()); + } + } +} + +fn log_generation(name: &str) -> Option { + let stem = name.strip_prefix("session")?; + let stem = stem.strip_suffix(".zstd").unwrap_or(stem); + let stem = stem.strip_suffix(".jsonl")?; + if stem.is_empty() { + return Some(0); + } + let digits = stem.strip_prefix(".v")?; + if digits.is_empty() || !digits.bytes().all(|byte| byte.is_ascii_digit()) { + return None; + } + digits.parse().ok() +} + +fn load_body(path: &Path) -> Result { + let text = read_log_text(path)?; + parse_log(&text) +} + +fn parse_log(text: &str) -> Result { + let mut lines = text.lines().filter(|line| !line.trim().is_empty()); + let Some(first) = lines.next() else { + return Err(Error::Malformed { + harness: Dsh::NAME, + detail: "empty session log".to_string(), + }); + }; + let header: Value = serde_json::from_str(first)?; + if header.get("type").and_then(Value::as_str) != Some("session") { + return Err(Error::Malformed { + harness: Dsh::NAME, + detail: "first line is not a session header".to_string(), + }); + } + let events = lines + .filter_map(|line| serde_json::from_str::(line).ok()) + .collect(); + Ok(DshSession { header, events }) +} + +fn read_log_text(path: &Path) -> Result { + let bytes = fs::read(path)?; + if path.extension().is_some_and(|ext| ext == "zstd") { + return decode_zstd(&bytes); + } + String::from_utf8(bytes).map_err(|error| Error::Malformed { + harness: Dsh::NAME, + detail: format!("session log is not utf-8: {error}"), + }) +} + +#[cfg(feature = "dsh_zstd")] +fn decode_zstd(bytes: &[u8]) -> Result { + use std::io::Read as _; + + // The official JSONL backend concatenates independent Zstandard frames + // (header frame, then one per append batch). The streaming decoder parses + // frame structure and continues across frame boundaries on its own, so the + // whole file decodes in one pass. Splitting on the frame magic instead + // would be wrong: those four bytes also occur inside compressed block + // payloads and checksums, and a false split turns a valid log into a + // decoder error. + let mut out = Vec::new(); + zstd::stream::read::Decoder::new(std::io::Cursor::new(bytes))?.read_to_end(&mut out)?; + String::from_utf8(out).map_err(|error| Error::Malformed { + harness: Dsh::NAME, + detail: format!("zstd session log is not utf-8: {error}"), + }) +} + +#[cfg(not(feature = "dsh_zstd"))] +fn decode_zstd(_bytes: &[u8]) -> Result { + Err(Error::Malformed { + harness: Dsh::NAME, + detail: "a compressed dsh session log needs txcript's `dsh_zstd` feature".to_string(), + }) +} + +fn file_fingerprint(path: &Path) -> String { + let Ok(metadata) = fs::metadata(path) else { + return String::new(); + }; + let modified = metadata + .modified() + .ok() + .map(|time| { + DateTime::::from(time) + .timestamp_nanos_opt() + .unwrap_or_default() + }) + .unwrap_or_default(); + format!("{}:{modified}", metadata.len()) +} + +fn meta_from_body(body: &DshSession) -> Meta { + let header = &body.header; + let id = header + .get("id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let timestamp = header + .get("createdAt") + .and_then(Value::as_i64) + .and_then(DateTime::from_timestamp_millis) + .or_else(|| first_event_time(&body.events)) + .unwrap_or(DateTime::::UNIX_EPOCH); + let cwd = header + .get("cwd") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(String::from); + let title = body.events.iter().rev().find_map(|event| { + (event.get("type").and_then(Value::as_str) == Some("session/title")) + .then(|| { + event + .get("data") + .and_then(|data| data.get("title")) + .and_then(Value::as_str) + .filter(|title| !title.is_empty()) + .map(String::from) + }) + .flatten() + }); + let model = body.events.iter().rev().find_map(|event| { + match event.get("type").and_then(Value::as_str) { + Some("request/context") => event + .get("data") + .and_then(|data| data.get("model")) + .and_then(Value::as_str) + .map(String::from), + Some("request/header") => event + .get("data") + .and_then(|data| data.get("header")) + .and_then(|header| header.get("config")) + .and_then(|config| config.get("model")) + .and_then(Value::as_str) + .map(String::from), + _ => None, + } + }); + Meta { + id, + timestamp, + cwd, + git_branch: None, + title, + cli_version: None, + model, + lineage: lineage_from_header(header), + } +} + +/// `parentSession` is the id of the session this one was seeded from. A +/// subagent run also carries `origin: "subagent"`; a user fork carries no +/// origin. dsh resumes a session under its own id, so there is no +/// continuation link. +fn lineage_from_header(header: &Value) -> Option { + let parent = header + .get("parentSession") + .and_then(Value::as_str) + .filter(|parent| !parent.is_empty())?; + let relation = if header.get("origin").and_then(Value::as_str) == Some("subagent") { + Relation::Spawn + } else { + Relation::Fork + }; + Some(Lineage { + parent: parent.to_string(), + relation, + }) +} + +fn first_event_time(events: &[Value]) -> Option> { + events + .iter() + .find_map(|event| event.get("time").and_then(Value::as_i64)) + .and_then(DateTime::from_timestamp_millis) +} + +/// `data.usage` is dsh's `TokenUsage`: required `inputTokens`/`outputTokens` +/// plus optional cache and reasoning tallies. Common has no reasoning counter, +/// so `reasoningTokens` survives only in the native body. +fn usage_from_data(data: Option<&Value>) -> Option { + let usage = data?.get("usage")?; + Some(Usage { + input_tokens: usage.get("inputTokens").and_then(Value::as_u64)?, + output_tokens: usage.get("outputTokens").and_then(Value::as_u64)?, + cache_read_input_tokens: usage.get("cacheReadTokens").and_then(Value::as_u64), + cache_creation_input_tokens: usage.get("cacheWriteTokens").and_then(Value::as_u64), + }) +} + +fn event_time(event: &Value, fallback: DateTime) -> DateTime { + event + .get("time") + .and_then(Value::as_i64) + .and_then(DateTime::from_timestamp_millis) + .unwrap_or(fallback) +} + +/// Resolve a `replace` op's inclusive seq range to surface positions. `None` +/// when either endpoint is missing from the current surface or the range runs +/// backwards — the same conditions dsh itself rejects. +fn replacement_range( + surface: &[(Option, &Value)], + op: &serde_json::Map, +) -> Option<(usize, usize)> { + let position = |seq: Option<&Value>| { + let seq = seq.and_then(Value::as_u64)?; + surface.iter().position(|(node, _)| *node == Some(seq)) + }; + // v3 renamed the endpoints from `start`/`end` to `startSeq`/`endSeq`. + let start = position(op.get("startSeq").or_else(|| op.get("start")))?; + let end = position(op.get("endSeq").or_else(|| op.get("end")))?; + (start <= end).then_some((start, end)) +} + +/// Rebuild the ordered surface, then project those nodes into Common messages. +/// Packed `*-chunks` rows and log-only events stay in the native body. +fn events_to_messages(events: &[Value], fallback: DateTime) -> Vec { + // The surface is an ordered list of event seqs, not a list of log indexes: + // log-only events (turn/step markers, chunks, packed rows) sit between + // surface nodes, so a node's seq and its surface position diverge + // immediately. A `replace` op names the inclusive *seq* range it shadows, + // and both endpoints must currently be on the surface. + let mut surface: Vec<(Option, &Value)> = Vec::new(); + for event in events { + let Some(kind) = event.get("type").and_then(Value::as_str) else { + continue; + }; + // System and developer messages (v3+) never become Common messages, + // but they are surface nodes a replacement span is counted across. + if !matches!( + kind, + "system/message" + | "developer/message" + | "user/message" + | "assistant/message" + | "tool/result" + ) { + continue; + } + let seq = event.get("seq").and_then(Value::as_u64); + let node = (seq, event); + match event.get("surfaceOp") { + Some(Value::Object(op)) if op.get("op").and_then(Value::as_str) == Some("replace") => { + match replacement_range(&surface, op) { + Some((start, end)) => { + surface.splice(start..=end, std::iter::once(node)); + } + // A range we cannot resolve (truncated log, unknown seq) + // shadows nothing; keep the node rather than drop history. + None => surface.push(node), + } + } + _ => surface.push(node), + } + } + + let mut messages = Vec::new(); + for (_, event) in surface { + let timestamp = event_time(event, fallback); + match event.get("type").and_then(Value::as_str) { + Some("user/message") => { + if is_environment_injection(event.get("data")) { + continue; + } + let blocks = user_text_blocks(event.get("data")); + if !blocks.is_empty() { + messages.push(user_message(blocks, timestamp)); + } + } + Some("assistant/message") => { + let data = event.get("data"); + let content = data + .and_then(|value| value.get("message")) + .and_then(|message| message.get("content")); + let blocks = assistant_blocks(content); + if blocks.is_empty() { + continue; + } + let model = data + .and_then(|value| value.get("message")) + .and_then(|message| message.get("source")) + .and_then(|source| source.get("model")) + .and_then(Value::as_str) + .map(String::from); + let has_tool = blocks + .iter() + .any(|block| matches!(block, Block::ToolUse { .. })); + // `data.interrupted` is only ever `true`, and marks the partial + // prefix a cancelled turn had already streamed. + let interrupted = data + .and_then(|value| value.get("interrupted")) + .and_then(Value::as_bool) + .unwrap_or(false); + let stop_reason = if interrupted { + StopReason::Aborted + } else if has_tool { + StopReason::ToolUse + } else { + StopReason::EndTurn + }; + messages.push(Message { + role: Role::Assistant, + content: blocks, + timestamp, + model, + stop_reason: Some(stop_reason), + usage: usage_from_data(data), + }); + } + Some("tool/result") => { + if let Some(block) = tool_result_block(event.get("data")) { + messages.push(user_message(vec![block], timestamp)); + } + } + _ => {} + } + } + messages +} + +/// Message sources dsh regenerates from its own environment on every request — +/// runtime and sandbox snapshots, the skill catalog, instruction files, +/// clocks. They ride on `user/message` but are not part of the conversation, +/// and would misdescribe any other harness a session is continued in. Every +/// other source stays, unknown ones included: a compaction checkpoint, for +/// one, carries the only remaining copy of the history it replaced. +const ENVIRONMENT_SOURCES: &[&str] = &[ + "runtime-context", + "skill-catalog", + "agent-instructions", + "time-context", + "tmux-context", + "repeat-tool-reminder", + "model-selection", + "plan-mode", +]; + +fn is_environment_injection(message: Option<&Value>) -> bool { + let Some(source) = message.and_then(|message| message.get("source")) else { + return false; + }; + let kind = match source.get("kind").and_then(Value::as_str) { + // Before v4 a producer was `{ kind: "plugin", plugin }`; v4 renamed + // the system-prompt plugin's non-system messages `runtime-context` + // and kept every other first-party name. + Some("plugin") => match source.get("plugin").and_then(Value::as_str) { + Some("@deepseek-ai/dsh-system-prompt") => "runtime-context", + Some(plugin) => plugin, + None => return false, + }, + Some(kind) => kind, + None => return false, + }; + ENVIRONMENT_SOURCES.contains(&kind) +} + +fn user_message(content: Vec, timestamp: DateTime) -> Message { + Message { + role: Role::User, + content, + timestamp, + model: None, + stop_reason: None, + usage: None, + } +} + +fn user_text_blocks(data: Option<&Value>) -> Vec { + let Some(content) = data + .and_then(|value| value.get("content")) + .and_then(Value::as_array) + else { + return Vec::new(); + }; + content + .iter() + .filter_map(|part| { + if part.get("type").and_then(Value::as_str) != Some("text") { + return None; + } + Some(Block::Text { + text: part + .get("text") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + }) + }) + .filter(|block| match block { + Block::Text { text } => !text.is_empty(), + _ => true, + }) + .collect() +} + +fn assistant_blocks(content: Option<&Value>) -> Vec { + let Some(parts) = content.and_then(Value::as_array) else { + return Vec::new(); + }; + parts + .iter() + .filter_map(|part| match part.get("type").and_then(Value::as_str) { + Some("text") => { + let text = part.get("text").and_then(Value::as_str).unwrap_or_default(); + (!text.is_empty()).then(|| Block::Text { + text: text.to_string(), + }) + } + Some("reasoning") => { + let text = part.get("text").and_then(Value::as_str).unwrap_or_default(); + (!text.is_empty()).then(|| Block::Thinking { + text: text.to_string(), + signature: None, + encrypted: None, + }) + } + Some("tool-call") => { + let id = part + .get("id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let name = part + .get("name") + .and_then(Value::as_str) + .unwrap_or("unknown"); + let input = part + .get("arguments") + .and_then(Value::as_str) + .and_then(|raw| serde_json::from_str(raw).ok()) + .or_else(|| part.get("arguments").cloned()) + .unwrap_or(Value::Null); + Some(Block::ToolUse { + id, + tool: Tool::from_canonical(name, input), + }) + } + _ => None, + }) + .collect() +} + +fn tool_result_block(data: Option<&Value>) -> Option { + let message = data.and_then(|value| value.get("message"))?; + // v4 lifted the result onto a tool-role message; before that it was a + // user-role message holding one `tool-result` wrapper block. + let result = if message.get("role").and_then(Value::as_str) == Some("tool") { + message + } else { + message + .get("content")? + .as_array()? + .iter() + .find(|part| part.get("type").and_then(Value::as_str) == Some("tool-result"))? + }; + let tool_use_id = result + .get("toolCallId") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let is_error = result + .get("isError") + .and_then(Value::as_bool) + .unwrap_or(false); + let text = result + .get("content") + .and_then(Value::as_array) + .map(|inner| { + inner + .iter() + .filter(|part| part.get("type").and_then(Value::as_str) == Some("text")) + .map(|part| part.get("text").and_then(Value::as_str).unwrap_or_default()) + .collect::>() + .join("\n") + }) + .unwrap_or_default(); + Some(Block::ToolResult { + tool_use_id, + content: ToolOutput::Text(text), + is_error, + }) +} + +const NS: Uuid = Uuid::from_bytes([ + 0x4d, 0x73, 0x68, 0x2d, 0x74, 0x78, 0x63, 0x72, 0x69, 0x70, 0x74, 0x2d, 0x64, 0x73, 0x68, 0x31, +]); + +/// What dsh itself writes for a tool call the model asked for but that never +/// started. The official validator accepts an unanswered call only as this +/// exact repair, so a Common call without a result is closed the same way. +const NOT_STARTED_TEXT: &str = "The tool call was interrupted before the Harness recorded it as started. Retry it if it is still needed."; + +/// dsh requires a provider and model on every assistant source; Common +/// records neither provider nor, always, a model. +const IMPORTED_PROVIDER: &str = "txcript"; +const UNKNOWN_MODEL: &str = "unknown"; + +fn body_from_common(transcript: &Transcript) -> DshSession { + let created = transcript.meta.timestamp.timestamp_millis(); + let id = if transcript.meta.id.is_empty() { + format!("session-{}", Uuid::new_v5(&NS, b"empty-id")) + } else { + transcript.meta.id.clone() + }; + let mut header = json!({ + "type": "session", + "version": SESSION_FORMAT_VERSION, + "id": id, + "createdAt": created, + "isSeeded": false, + "delegationDepth": 0, + }); + if let Some(cwd) = &transcript.meta.cwd { + header["cwd"] = json!(cwd); + } + let model = transcript.meta.model.as_deref().unwrap_or(UNKNOWN_MODEL); + let mut log = LogWriter::new(&id, created); + if let Some(title) = &transcript.meta.title { + // A `fallback` title must cite the human message it was derived + // from; this one was set elsewhere, which dsh records as `user`. + log.push(json!({ + "type": "session/title", + "time": created, + "data": { "title": title, "messageSeqs": [], "source": { "kind": "user" } } + })); + } + for (index, message) in transcript.body.iter().enumerate() { + if message.role == Role::User { + log.user(index, message); + } else { + log.assistant(index, message, model); + } + } + let events = log.finish(); + DshSession { header, events } +} + +/// Lays Common messages out as dsh's turn/step lifecycle, which the official +/// reader checks: a human message opens a turn and its first step, a model +/// response after tool results opens the next step, and every tool call is +/// started (`tool/call`) and answered (`tool/result`) before its step ends. +struct LogWriter<'a> { + id: &'a str, + events: Vec, + time: i64, + turn: u64, + turn_open: bool, + step: u64, + step_state: StepState, + interrupted: bool, + /// Calls advertised in the open step and not yet answered, in order: + /// `(id, name, arguments)`. + pending: Vec<(String, String, String)>, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum StepState { + Closed, + Open, + /// A tool result has answered the step since its last model response. + Answered, +} + +impl<'a> LogWriter<'a> { + fn new(id: &'a str, time: i64) -> Self { + Self { + id, + events: Vec::new(), + time, + turn: 0, + turn_open: false, + step: 0, + step_state: StepState::Closed, + interrupted: false, + pending: Vec::new(), + } + } + + fn push(&mut self, mut event: Value) -> u64 { + let seq = self.events.len() as u64; + event["seq"] = json!(seq); + self.events.push(event); + seq + } + + fn user(&mut self, index: usize, message: &Message) { + self.time = message.timestamp.timestamp_millis(); + for block in &message.content { + if let Block::ToolResult { + tool_use_id, + content, + is_error, + } = block + { + self.tool_result(index, tool_use_id, content, *is_error); + } + } + let texts: Vec = message + .content + .iter() + .filter_map(|block| match block { + Block::Text { text } => Some(json!({"type": "text", "text": text})), + _ => None, + }) + .collect(); + if texts.is_empty() { + return; + } + self.close_turn(); + self.open_turn(); + self.open_step(); + let msg_id = Uuid::new_v5(&NS, format!("{}:{index}:user", self.id).as_bytes()).to_string(); + self.push(json!({ + "type": "user/message", + "time": self.time, + "data": { + "role": "user", + "id": msg_id, + "content": texts, + "source": { "kind": "user" } + }, + "surfaceOp": "append" + })); + } + + /// A result whose call is not pending in the open step has nothing dsh + /// would accept it against, and is left out. + fn tool_result(&mut self, index: usize, call_id: &str, content: &ToolOutput, is_error: bool) { + let Some(position) = self.pending.iter().position(|(id, ..)| id == call_id) else { + return; + }; + let (_, name, arguments) = self.pending.remove(position); + let call_seq = self.push(json!({ + "type": "tool/call", + "time": self.time, + "data": { + "turn": self.turn, + "step": self.step, + "callId": call_id, + "name": name, + "arguments": arguments + } + })); + let text = match content { + ToolOutput::Text(text) => text.clone(), + ToolOutput::Json(value) => value.to_string(), + }; + let msg_id = Uuid::new_v5( + &NS, + format!("{}:{index}:result:{call_id}", self.id).as_bytes(), + ) + .to_string(); + self.push(json!({ + "type": "tool/result", + "time": self.time, + "data": { + "turn": self.turn, + "step": self.step, + "message": { + "role": "tool", + "id": msg_id, + "toolCallId": call_id, + "content": [{ "type": "text", "text": text }], + "isError": is_error, + "source": { "kind": "tool", "callId": call_id } + } + }, + "sourceEventSeqs": [call_seq], + "surfaceOp": "append" + })); + self.step_state = StepState::Answered; + } + + fn assistant(&mut self, index: usize, message: &Message, default_model: &str) { + let mut content = Vec::new(); + let mut calls = Vec::new(); + for block in &message.content { + match block { + Block::Text { text } => content.push(json!({"type": "text", "text": text})), + Block::Thinking { text, .. } => { + content.push(json!({"type": "reasoning", "text": text})); + } + Block::ToolUse { id: call_id, tool } => { + let (name, input) = tool.to_canonical(); + let arguments = + serde_json::to_string(&input).unwrap_or_else(|_| "{}".to_string()); + content.push(json!({ + "type": "tool-call", + "id": call_id, + "name": name, + "arguments": arguments + })); + calls.push((call_id.clone(), name, arguments)); + } + Block::ToolResult { .. } | Block::Image { .. } | Block::Artifact { .. } => {} + } + } + if content.is_empty() { + return; + } + self.time = message.timestamp.timestamp_millis(); + if !self.turn_open { + self.open_turn(); + } + // A response after a tool result is a new model call, so a new step. + if self.step_state == StepState::Answered { + self.close_step(); + } + self.open_step(); + let msg_id = + Uuid::new_v5(&NS, format!("{}:{index}:assistant", self.id).as_bytes()).to_string(); + let model = message.model.as_deref().unwrap_or(default_model); + let mut data = json!({ + "turn": self.turn, + "step": self.step, + "message": { + "role": "assistant", + "id": msg_id, + "content": content, + "source": { "kind": "model", "provider": IMPORTED_PROVIDER, "model": model } + }, + // The provider stream is not recorded in Common. dsh replays an + // embedded stream against the message it produced, and skips + // that check only for an empty one. + "stream": [] + }); + if let Some(usage) = &message.usage { + let mut tokens = json!({ + "inputTokens": usage.input_tokens, + "outputTokens": usage.output_tokens, + }); + if let Some(read) = usage.cache_read_input_tokens { + tokens["cacheReadTokens"] = json!(read); + } + if let Some(write) = usage.cache_creation_input_tokens { + tokens["cacheWriteTokens"] = json!(write); + } + data["usage"] = tokens; + } + // dsh only ever writes `interrupted: true`; absence means a + // completed response. + if message.stop_reason == Some(StopReason::Aborted) { + data["interrupted"] = json!(true); + self.interrupted = true; + } + self.push(json!({ + "type": "assistant/message", + "time": self.time, + "data": data, + "surfaceOp": "append" + })); + for call in calls { + if !self.pending.iter().any(|(id, ..)| *id == call.0) { + self.pending.push(call); + } + } + } + + fn open_turn(&mut self) { + self.turn += 1; + self.turn_open = true; + self.step = 0; + self.interrupted = false; + self.push(json!({ + "type": "turn/start", + "time": self.time, + "data": { "turn": self.turn } + })); + } + + fn open_step(&mut self) { + if self.step_state != StepState::Closed { + return; + } + self.step += 1; + self.step_state = StepState::Open; + self.push(json!({ + "type": "step/start", + "time": self.time, + "data": { "turn": self.turn, "step": self.step } + })); + if self.turn == 1 && self.step == 1 { + self.system_head(); + } + } + + /// dsh appends its current system prompt to the history of a session it + /// resumes, and accepts that only behind a system message heading the + /// surface; without one, the resumed log fails dsh's next load. Common has + /// no system prompt to put there, so the head is empty. + fn system_head(&mut self) { + let msg_id = Uuid::new_v5(&NS, format!("{}:system", self.id).as_bytes()).to_string(); + self.push(json!({ + "type": "system/message", + "time": self.time, + "data": { + "turn": self.turn, + "step": self.step, + "message": { + "role": "system", + "id": msg_id, + "content": [], + "source": { "kind": "system-prompt" } + } + }, + "surfaceOp": "append" + })); + } + + /// Close the open step, answering each unanswered call with dsh's own + /// not-started repair. + fn close_step(&mut self) { + if self.step_state == StepState::Closed { + return; + } + for (call_id, ..) in std::mem::take(&mut self.pending) { + let seq = self.events.len(); + self.push(json!({ + "type": "tool/result", + "time": self.time, + "data": { + "turn": self.turn, + "step": self.step, + "message": { + "role": "tool", + "id": format!("interrupted-tool-result-{call_id}-{seq}"), + "toolCallId": call_id, + "content": [{ "type": "text", "text": NOT_STARTED_TEXT }], + "isError": true, + "source": { "kind": "tool", "callId": call_id } + }, + "error": { "name": "ToolNotStartedError", "code": "TOOL_NOT_STARTED" } + }, + "surfaceOp": "append" + })); + self.interrupted = true; + } + self.push(json!({ + "type": "step/end", + "time": self.time, + "data": { "turn": self.turn, "step": self.step } + })); + self.step_state = StepState::Closed; + } + + fn close_turn(&mut self) { + if !self.turn_open { + return; + } + self.close_step(); + let reason = if self.interrupted { + "interrupted" + } else { + "completed" + }; + self.push(json!({ + "type": "turn/end", + "time": self.time, + "data": { "turn": self.turn, "reason": { "kind": reason } } + })); + self.turn_open = false; + } + + fn finish(mut self) -> Vec { + self.close_turn(); + self.events + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ts(ms: i64) -> DateTime { + DateTime::from_timestamp_millis(ms).unwrap_or(DateTime::::UNIX_EPOCH) + } + + #[test] + fn header_line_feeds_meta() { + let body = DshSession { + header: json!({ + "type": "session", + "version": 0, + "id": "session-abc", + "createdAt": 1_786_637_231_769_i64, + "cwd": "/repo", + "delegationDepth": 0, + "agentPreset": "standard" + }), + events: vec![ + json!({"type": "session/title", "seq": 0, "time": 1, "data": {"title": "hello"}}), + json!({"type": "request/context", "seq": 1, "time": 2, "data": {"provider": "deepseek-official", "model": "deepseek-v4-flash"}}), + ], + }; + let meta = meta_from_body(&body); + assert_eq!(meta.id, "session-abc"); + assert_eq!(meta.cwd.as_deref(), Some("/repo")); + assert_eq!(meta.title.as_deref(), Some("hello")); + assert_eq!(meta.model.as_deref(), Some("deepseek-v4-flash")); + assert_eq!(meta.timestamp, ts(1_786_637_231_769)); + } + + #[test] + fn surface_projection_keeps_text_thinking_and_tools() { + let events = vec![ + json!({"type": "user/message", "seq": 0, "time": 10, "data": { + "role": "user", + "content": [{"type": "text", "text": "hi"}], + "source": {"kind": "user"} + }, "surfaceOp": "append"}), + json!({"type": "assistant/chunk", "seq": 1, "time": 11, "data": {"turn": 1, "step": 1, "chunk": {"type": "block-start"}}}), + json!({"type": "reasoning-chunks", "seq0": 2, "time0": 12, "data": {"texts": ["skip"]}}), + json!({"type": "assistant/message", "seq": 3, "time": 13, "data": { + "turn": 1, "step": 1, + "message": { + "role": "assistant", + "content": [ + {"type": "reasoning", "text": "think"}, + {"type": "text", "text": "ok"}, + {"type": "tool-call", "id": "c1", "name": "bash", "arguments": "{\"command\":\"ls\"}"} + ], + "source": {"kind": "model", "model": "deepseek-v4-flash"} + } + }, "surfaceOp": "append"}), + json!({"type": "tool/result", "seq": 4, "time": 14, "data": { + "message": {"content": [{ + "type": "tool-result", + "toolCallId": "c1", + "isError": false, + "content": [{"type": "text", "text": "a.rs"}] + }]} + }, "surfaceOp": "append"}), + ]; + let messages = events_to_messages(&events, DateTime::::UNIX_EPOCH); + assert_eq!(messages.len(), 3); + assert!(matches!(&messages[0].content[0], Block::Text { text } if text == "hi")); + assert!(matches!(&messages[1].content[0], Block::Thinking { text, .. } if text == "think")); + assert!(matches!(&messages[1].content[2], Block::ToolUse { id, .. } if id == "c1")); + assert_eq!(messages[1].model.as_deref(), Some("deepseek-v4-flash")); + assert!(matches!( + &messages[2].content[0], + Block::ToolResult { tool_use_id, is_error, .. } if tool_use_id == "c1" && !is_error + )); + } + + fn user_event(seq: u64, text: &str, surface_op: &Value) -> Value { + json!({"type": "user/message", "seq": seq, "time": seq, "data": { + "content": [{"type": "text", "text": text}] + }, "surfaceOp": surface_op}) + } + + #[test] + fn surface_replace_drops_shadowed_nodes() { + // Log-only events between surface nodes push each node's seq past its + // surface position, so a replace op must be resolved by seq. + let events = vec![ + user_event(0, "old", &json!("append")), + json!({"type": "step/start", "seq": 1, "time": 1, "data": {}}), + user_event(2, "kept", &json!("append")), + user_event( + 3, + "summary", + &json!({"op": "replace", "start": 0, "end": 0}), + ), + ]; + let messages = events_to_messages(&events, DateTime::::UNIX_EPOCH); + assert_eq!(messages.len(), 2); + assert!(matches!(&messages[0].content[0], Block::Text { text } if text == "summary")); + assert!(matches!(&messages[1].content[0], Block::Text { text } if text == "kept")); + } + + #[test] + fn surface_replace_spans_the_whole_seq_range() { + // Compaction shadows a contiguous run: nodes at seq 1 and 3 must both + // go, even though seq 3 sits at surface index 1. + let events = vec![ + user_event(1, "first", &json!("append")), + json!({"type": "assistant/chunk", "seq": 2, "time": 2, "data": {}}), + user_event(3, "second", &json!("append")), + user_event(4, "kept", &json!("append")), + user_event( + 9, + "summary", + &json!({"op": "replace", "start": 1, "end": 3}), + ), + ]; + let messages = events_to_messages(&events, DateTime::::UNIX_EPOCH); + assert_eq!(messages.len(), 2); + assert!(matches!(&messages[0].content[0], Block::Text { text } if text == "summary")); + assert!(matches!(&messages[1].content[0], Block::Text { text } if text == "kept")); + } + + #[test] + fn surface_replace_with_unresolvable_range_keeps_history() { + // A range naming a seq that never reached the surface shadows nothing. + let events = vec![ + user_event(4, "kept", &json!("append")), + user_event( + 7, + "summary", + &json!({"op": "replace", "start": 2, "end": 2}), + ), + ]; + let messages = events_to_messages(&events, DateTime::::UNIX_EPOCH); + assert_eq!(messages.len(), 2); + assert!(matches!(&messages[0].content[0], Block::Text { text } if text == "kept")); + assert!(matches!(&messages[1].content[0], Block::Text { text } if text == "summary")); + } + + #[test] + fn assistant_usage_and_interruption_survive_projection() { + let events = vec![ + json!({"type": "assistant/message", "seq": 0, "time": 5, "data": { + "turn": 1, "step": 1, + "message": { + "role": "assistant", + "content": [{"type": "text", "text": "partial"}], + "source": {"kind": "model", "model": "deepseek-v4-flash"} + }, + "usage": { + "inputTokens": 13972, + "outputTokens": 205, + "cacheReadTokens": 64, + "cacheWriteTokens": 32, + "reasoningTokens": 106 + }, + "interrupted": true + }, "surfaceOp": "append"}), + ]; + let messages = events_to_messages(&events, DateTime::::UNIX_EPOCH); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].stop_reason, Some(StopReason::Aborted)); + let usage = messages[0].usage.expect("usage projected"); + assert_eq!(usage.input_tokens, 13972); + assert_eq!(usage.output_tokens, 205); + assert_eq!(usage.cache_read_input_tokens, Some(64)); + assert_eq!(usage.cache_creation_input_tokens, Some(32)); + } + + #[test] + fn assistant_without_usage_keeps_normal_stop_reason() { + let events = vec![ + json!({"type": "assistant/message", "seq": 0, "time": 5, "data": { + "message": {"role": "assistant", "content": [{"type": "text", "text": "done"}]} + }, "surfaceOp": "append"}), + ]; + let messages = events_to_messages(&events, DateTime::::UNIX_EPOCH); + assert_eq!(messages[0].stop_reason, Some(StopReason::EndTurn)); + assert!(messages[0].usage.is_none()); + } + + #[cfg(feature = "dsh_zstd")] + #[test] + fn decode_zstd_spans_concatenated_frames() { + // The backend writes the header in its own frame, then one frame per + // append batch; all of them must decode as a single JSONL stream. + let header = b"{\"type\":\"session\",\"version\":0,\"id\":\"s1\"}\n"; + let batch = b"{\"type\":\"user/message\",\"seq\":0,\"time\":1,\"data\":{},\"surfaceOp\":\"append\"}\n"; + let mut bytes = zstd::encode_all(&header[..], 0).unwrap(); + bytes.extend(zstd::encode_all(&batch[..], 0).unwrap()); + + let text = decode_zstd(&bytes).unwrap(); + assert_eq!(text.lines().count(), 2); + assert!(text.starts_with("{\"type\":\"session\"")); + assert!(text.contains("user/message")); + } + + #[test] + fn only_canonical_log_names_carry_a_generation() { + assert_eq!(log_generation("session.jsonl"), Some(0)); + assert_eq!(log_generation("session.jsonl.zstd"), Some(0)); + assert_eq!(log_generation("session.v4.jsonl.zstd"), Some(4)); + assert_eq!(log_generation("session.v12.jsonl"), Some(12)); + // dsh publishes through a temporary sibling, then links it into place. + assert_eq!( + log_generation("session.v4.jsonl.zstd.af44d7226f77.tmp"), + None + ); + assert_eq!(log_generation("session.lock"), None); + assert_eq!(log_generation("session.v.jsonl"), None); + assert_eq!(log_generation("._session.v4.jsonl.zstd"), None); + } + + #[test] + fn parse_log_requires_session_header() { + let err = parse_log("{\"type\":\"user/message\"}\n").unwrap_err(); + assert!(err.to_string().contains("session header")); + } +} diff --git a/src/harness/mod.rs b/src/harness/mod.rs index b550c47..5644fb8 100644 --- a/src/harness/mod.rs +++ b/src/harness/mod.rs @@ -17,6 +17,7 @@ mod cowork_files; pub mod cowork_remote; pub mod cursor; pub mod cursor_desktop; +pub mod dsh; pub mod fx; pub mod grok; pub mod grok_bot; diff --git a/src/lib.rs b/src/lib.rs index a340523..e01b516 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,7 +1,7 @@ //! Typed conversion for coding-agent session transcripts. //! //! Claude Code, Claude Chat, Cowork, Codex, `OpenCode`, pi, Campfire, Cursor, -//! Grok, Hermes, Amp, and Antigravity record similar conversation data in +//! Grok, `dsh`, Hermes, Amp, and Antigravity record similar conversation data in //! different stores. This crate maps each format through [`Transcript`] and //! converts with [`convert::`](convert): `A` -> [`Common`] -> `B`. //! diff --git a/src/local.rs b/src/local.rs index 4c8dd57..588a5d8 100644 --- a/src/local.rs +++ b/src/local.rs @@ -23,7 +23,7 @@ use chrono::{DateTime, Utc}; use crate::common::{ArtifactSource, Block, Meta}; use crate::harness::{ - amp, antigravity, campfire, claude_code, codex, cowork, cursor, fx, grok, grok_bot, pi, + amp, antigravity, campfire, claude_code, codex, cowork, cursor, dsh, fx, grok, grok_bot, pi, }; #[cfg(feature = "chatgpt")] @@ -131,6 +131,8 @@ pub fn discover_with(mut on_store: impl FnMut(HarnessId, usize)) -> Vec grok_bot::GrokBotStore::default_root(), &mut out, ); + on_store(HarnessId::Dsh, out.len()); + scan(HarnessId::Dsh, dsh::DshStore::default_root(), &mut out); on_store(HarnessId::Fx, out.len()); scan(HarnessId::Fx, fx::FxStore::default_root(), &mut out); on_store(HarnessId::Amp, out.len()); @@ -379,6 +381,7 @@ impl Session { (HarnessId::Cursor, Locator::Path(p)) => go(cursor::CursorStore::default_root(), p), (HarnessId::Grok, Locator::Path(p)) => go(grok::GrokStore::default_root(), p), (HarnessId::GrokBot, Locator::Path(p)) => go(grok_bot::GrokBotStore::default_root(), p), + (HarnessId::Dsh, Locator::Path(p)) => go(dsh::DshStore::default_root(), p), (HarnessId::Fx, Locator::Path(p)) => go(fx::FxStore::default_root(), p), (HarnessId::Amp, Locator::Path(p)) => go(amp::AmpStore::default_root(), p), (HarnessId::Antigravity, Locator::Path(p)) => { @@ -438,6 +441,7 @@ impl Session { (HarnessId::Cursor, Locator::Path(p)) => go(cursor::CursorStore::default_root(), p), (HarnessId::Grok, Locator::Path(p)) => go(grok::GrokStore::default_root(), p), (HarnessId::GrokBot, Locator::Path(p)) => go(grok_bot::GrokBotStore::default_root(), p), + (HarnessId::Dsh, Locator::Path(p)) => go(dsh::DshStore::default_root(), p), (HarnessId::Fx, Locator::Path(p)) => go(fx::FxStore::default_root(), p), (HarnessId::Amp, Locator::Path(p)) => go(amp::AmpStore::default_root(), p), (HarnessId::Antigravity, Locator::Path(p)) => { @@ -507,6 +511,7 @@ pub fn fingerprints(sessions: &[Session]) -> Vec { HarnessId::Cursor => group.files(cursor::CursorStore::default_root()), HarnessId::Grok => group.files(grok::GrokStore::default_root()), HarnessId::GrokBot => group.files(grok_bot::GrokBotStore::default_root()), + HarnessId::Dsh => group.files(dsh::DshStore::default_root()), HarnessId::Fx => group.files(fx::FxStore::default_root()), HarnessId::Amp => group.files(amp::AmpStore::default_root()), HarnessId::Antigravity => group.files(antigravity::AntigravityStore::default_root()), @@ -836,6 +841,13 @@ pub fn write_with( }) } } + HarnessId::Dsh => go( + dsh::DshStore::default_root(), + dsh::DshStore::new, + root, + common, + |s| s.sessions_dir, + ), HarnessId::Fx => go( fx::FxStore::default_root(), fx::FxStore::new, @@ -1119,6 +1131,10 @@ pub fn resume_command(harness: HarnessId, id: &str) -> (String, Vec) { // the session is in the Agents sidebar. HarnessId::CursorDesktop => ("cursor".into(), Vec::new()), HarnessId::Grok => ("grok".into(), vec!["--resume".into(), id]), + HarnessId::Dsh => ( + "dsh".into(), + vec!["--profile".into(), "tui".into(), "--resume".into(), id], + ), HarnessId::Fx => ("fx".into(), vec!["--resume".into(), id]), HarnessId::Hermes => ("hermes".into(), vec!["--resume".into(), id]), HarnessId::Amp => ("amp".into(), vec!["threads".into(), "continue".into(), id]), diff --git a/src/transcript.rs b/src/transcript.rs index c89ba1e..f6bbba9 100644 --- a/src/transcript.rs +++ b/src/transcript.rs @@ -456,6 +456,7 @@ pub enum HarnessId { CursorDesktop, Grok, GrokBot, + Dsh, Fx, Hermes, Amp, @@ -465,7 +466,7 @@ pub enum HarnessId { } impl HarnessId { - pub const ALL: [HarnessId; 18] = [ + pub const ALL: [HarnessId; 19] = [ HarnessId::ClaudeCode, HarnessId::ClaudeChat, HarnessId::CoworkRemote, @@ -478,6 +479,7 @@ impl HarnessId { HarnessId::CursorDesktop, HarnessId::Grok, HarnessId::GrokBot, + HarnessId::Dsh, HarnessId::Fx, HarnessId::Hermes, HarnessId::Amp, @@ -502,6 +504,7 @@ impl HarnessId { HarnessId::CursorDesktop => "cursor_desktop", HarnessId::Grok => "grok", HarnessId::GrokBot => "grok_bot", + HarnessId::Dsh => "dsh", HarnessId::Fx => "fx", HarnessId::Hermes => "hermes", HarnessId::Amp => "amp", @@ -548,6 +551,9 @@ impl FromStr for HarnessId { "grok_bot" | "grok-bot" | "grokbot" | "grok_desktop" | "grok-desktop" => { Ok(HarnessId::GrokBot) } + "dsh" | "deepseek" | "deepseek_harness" | "deepseek-harness" | "deepseekharness" => { + Ok(HarnessId::Dsh) + } "fx" | "fx_cli" | "fx-cli" | "fxcli" | "vercel_fx" | "vercel-fx" => Ok(HarnessId::Fx), "hermes" | "hermes_agent" | "hermes-agent" | "hermesagent" => Ok(HarnessId::Hermes), "amp" | "ampcode" | "amp_code" | "amp-code" => Ok(HarnessId::Amp), diff --git a/src/wasm.rs b/src/wasm.rs index e1bbb6f..3d37f8d 100644 --- a/src/wasm.rs +++ b/src/wasm.rs @@ -15,7 +15,7 @@ use wasm_bindgen::prelude::*; use crate::common; use crate::harness::{ amp, antigravity, campfire, chatgpt, claude_chat, claude_code, codex, cowork, cowork_remote, - cursor, cursor_desktop, fx, grok, grok_bot, hermes, opencode, pi, simple, + cursor, cursor_desktop, dsh, fx, grok, grok_bot, hermes, opencode, pi, simple, }; use crate::transcript::{Codec, Common, HarnessId, TextCodec, Transcript}; @@ -211,6 +211,7 @@ fn parse_to_common(harness: HarnessId, text: &str) -> crate::Result go::(text), HarnessId::Grok => go::(text), HarnessId::GrokBot => go::(text), + HarnessId::Dsh => go::(text), HarnessId::Fx => go::(text), HarnessId::Hermes => go::(text), HarnessId::Amp => go::(text), @@ -237,6 +238,7 @@ fn render_from_common(harness: HarnessId, common: &Transcript) -> crate: HarnessId::CursorDesktop => go::(common), HarnessId::Grok => go::(common), HarnessId::GrokBot => go::(common), + HarnessId::Dsh => go::(common), HarnessId::Fx => go::(common), HarnessId::Hermes => go::(common), HarnessId::Amp => go::(common), diff --git a/tests/integration/cross_harness.rs b/tests/integration/cross_harness.rs index 7a80ba5..8d72f30 100644 --- a/tests/integration/cross_harness.rs +++ b/tests/integration/cross_harness.rs @@ -6,7 +6,7 @@ use chrono::{DateTime, Utc}; use txcript::common; use txcript::harness::{ - amp, antigravity, campfire, claude_code, codex, cowork, cursor, cursor_desktop, fx, grok, + amp, antigravity, campfire, claude_code, codex, cowork, cursor, cursor_desktop, dsh, fx, grok, grok_bot, hermes, opencode, pi, simple, }; use txcript::{Codec, Common, Transcript, convert}; @@ -194,6 +194,7 @@ fn sample_with_raw_tool(tool_name: &str) -> Transcript { Transcript::new(meta, body) } +#[allow(clippy::too_many_lines)] fn assert_cycle(common: &Transcript, context: &str) { let expected = signature(common); @@ -261,7 +262,14 @@ fn assert_cycle(common: &Transcript, context: &str) { "{context}: grok_bot" ); - let fx = convert::(&grok_bot).unwrap(); + let dsh = convert::(&grok_bot).unwrap(); + assert_eq!( + signature(&dsh::Dsh::to_common(&dsh).unwrap()), + expected, + "{context}: dsh" + ); + + let fx = convert::(&dsh).unwrap(); assert_eq!( signature(&fx::Fx::to_common(&fx).unwrap()), expected, diff --git a/tests/integration/dsh.rs b/tests/integration/dsh.rs new file mode 100644 index 0000000..69bee8e --- /dev/null +++ b/tests/integration/dsh.rs @@ -0,0 +1,768 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] + +use chrono::{DateTime, Utc}; +use serde_json::json; +use txcript::common::{self, Block}; +use txcript::harness::dsh; +use txcript::{Codec, Common, Store, TextCodec}; + +fn ts(s: &str) -> DateTime { + s.parse().unwrap() +} + +fn sample_body() -> dsh::DshSession { + dsh::DshSession { + header: json!({ + "type": "session", + "version": 0, + "id": "session-abc", + "createdAt": 1_704_067_445_000_i64, + "cwd": "/repo", + "delegationDepth": 0, + "agentPreset": "standard" + }), + events: vec![ + json!({"type": "session/title", "seq": 0, "time": 1_704_067_445_000_i64, + "data": {"title": "Parser work", "source": {"kind": "fallback"}}}), + json!({"type": "request/context", "seq": 1, "time": 1_704_067_445_001_i64, + "data": {"provider": "deepseek-official", "model": "deepseek-v4-flash"}}), + json!({"type": "user/message", "seq": 2, "time": 1_704_067_445_002_i64, "data": { + "role": "user", + "id": "u1", + "content": [{"type": "text", "text": "list files"}], + "source": {"kind": "user"} + }, "surfaceOp": "append"}), + json!({"type": "reasoning-chunks", "seq0": 3, "time0": 1_704_067_445_003_i64, + "data": {"texts": ["should", " not", " appear"]}}), + json!({"type": "assistant/message", "seq": 4, "time": 1_704_067_445_004_i64, "data": { + "turn": 1, "step": 1, + "message": { + "role": "assistant", + "id": "a1", + "content": [ + {"type": "reasoning", "text": "plan"}, + {"type": "tool-call", "id": "c1", "name": "bash", + "arguments": "{\"command\":\"ls\"}"} + ], + "source": {"kind": "model", "provider": "deepseek-official", "model": "deepseek-v4-flash"} + } + }, "surfaceOp": "append"}), + json!({"type": "tool/result", "seq": 5, "time": 1_704_067_445_005_i64, "data": { + "turn": 1, "step": 1, + "message": { + "role": "user", + "content": [{ + "type": "tool-result", + "toolCallId": "c1", + "isError": false, + "content": [{"type": "text", "text": "a.rs"}] + }] + } + }, "surfaceOp": "append"}), + json!({"type": "future.dsh.event", "seq": 6, "time": 1_704_067_445_006_i64, "data": {}}), + ], + } +} + +#[test] +fn metadata_and_messages_are_converted() { + let native = txcript::Transcript::new( + common::Meta { + id: "session-abc".into(), + timestamp: ts("2024-01-01T00:00:45.000Z"), + cwd: Some("/repo".into()), + git_branch: None, + title: Some("Parser work".into()), + cli_version: None, + model: Some("deepseek-v4-flash".into()), + lineage: None, + }, + sample_body(), + ); + let text = dsh::Dsh::to_text(&native).unwrap(); + let parsed = dsh::Dsh::from_text(&text).unwrap(); + assert_eq!(parsed.meta.cwd.as_deref(), Some("/repo")); + assert_eq!(parsed.meta.title.as_deref(), Some("Parser work")); + assert_eq!(parsed.meta.model.as_deref(), Some("deepseek-v4-flash")); + + let common = dsh::Dsh::to_common(&parsed).unwrap(); + assert_eq!(common.body.len(), 3); + assert!(matches!( + common.body[1].content[0], + common::Block::Thinking { .. } + )); + assert!(matches!( + common.body[1].content[1], + common::Block::ToolUse { .. } + )); + assert!(matches!( + common.body[2].content[0], + common::Block::ToolResult { .. } + )); +} + +#[test] +fn native_text_round_trip_retains_unknown_events() { + let transcript = txcript::Transcript::new( + common::Meta { + id: "session-abc".into(), + timestamp: ts("2024-01-01T00:00:45.000Z"), + cwd: Some("/repo".into()), + git_branch: None, + title: Some("Parser work".into()), + cli_version: None, + model: Some("deepseek-v4-flash".into()), + lineage: None, + }, + sample_body(), + ); + let text = dsh::Dsh::to_text(&transcript).unwrap(); + let parsed = dsh::Dsh::from_text(&text).unwrap(); + assert_eq!(parsed.body, sample_body()); + assert!(text.contains("future.dsh.event")); +} + +#[test] +fn store_discovers_and_loads_a_session() { + let root = tempfile::tempdir().unwrap(); + let session = root.path().join("--repo--").join("session-abc"); + write_session(&session, &sample_body()); + let store = dsh::DshStore::new(root.path()); + let found = store.discover().unwrap(); + assert_eq!(found.len(), 1); + assert_eq!(found[0].meta.id, "session-abc"); + let loaded = store.load(&found[0].reference).unwrap(); + assert_eq!(loaded.body, sample_body()); +} + +fn saveable(id: &str, cwd: &str) -> txcript::Transcript { + let mut body = sample_body(); + body.header["id"] = json!(id); + body.header["cwd"] = json!(cwd); + let meta = common::Meta { + id: id.into(), + timestamp: ts("2024-01-01T00:04:05.000Z"), + cwd: Some(cwd.into()), + git_branch: None, + title: Some("Parser work".into()), + cli_version: None, + model: None, + lineage: None, + }; + txcript::Transcript::new(meta, body) +} + +#[cfg(feature = "dsh_zstd")] +#[test] +fn save_writes_the_layout_dsh_scans_for() { + // dsh finds sessions by walking `///`, and + // then asserts the header's own id and cwd name that exact path. A layout + // that disagrees is not merely skipped — it fails dsh's whole listing. + let root = tempfile::tempdir().unwrap(); + let store = dsh::DshStore::new(root.path()); + let saved = store + .save(&saveable("session-abc", "/repo/Mobile Documents/想法")) + .unwrap(); + + let expected = root + .path() + .join("--repo-Mobile~0020Documents-~60F3~6CD5--") + .join("session-abc"); + assert_eq!(saved.reference, expected); + assert!(expected.join("session.jsonl.zstd").is_file()); +} + +#[cfg(feature = "dsh_zstd")] +#[test] +fn saved_session_is_discovered_and_loads_back() { + let root = tempfile::tempdir().unwrap(); + let store = dsh::DshStore::new(root.path()); + let original = saveable("session-abc", "/repo"); + store.save(&original).unwrap(); + + let found = store.discover().unwrap(); + assert_eq!(found.len(), 1); + assert_eq!(found[0].meta.id, "session-abc"); + + let loaded = store.load(&found[0].reference).unwrap(); + assert_eq!(loaded.body, original.body); +} + +/// Decode only the first Zstandard frame of a written log, the way dsh's +/// `assertZstdHeaderFrame` does. +#[cfg(feature = "dsh_zstd")] +fn first_frame(log: &std::path::Path) -> String { + use std::io::Read as _; + + let bytes = std::fs::read(log).unwrap(); + let mut decoder = zstd::stream::read::Decoder::new(std::io::Cursor::new(bytes)) + .unwrap() + .single_frame(); + let mut first = String::new(); + decoder.read_to_string(&mut first).unwrap(); + first +} + +#[cfg(feature = "dsh_zstd")] +#[test] +fn the_first_frame_holds_exactly_the_header_line() { + // dsh reads a session's metadata by decompressing one frame and requiring + // it to be a single line. Compressing the whole log as one frame still + // decodes, and still round-trips through txcript, but dsh rejects it — + // the frame split is part of the format. + let root = tempfile::tempdir().unwrap(); + let store = dsh::DshStore::new(root.path()); + let saved = store.save(&saveable("session-abc", "/repo")).unwrap(); + + let first = first_frame(&saved.reference.join("session.jsonl.zstd")); + assert_eq!(first.matches('\n').count(), 1); + assert!(first.ends_with('\n')); + let header: serde_json::Value = serde_json::from_str(first.trim_end()).unwrap(); + assert_eq!(header["type"], json!("session")); + assert_eq!(header["id"], json!("session-abc")); +} + +#[cfg(feature = "dsh_zstd")] +#[test] +fn save_stamps_the_header_with_the_identity_that_built_the_path() { + // dsh cross-checks every log it finds against the path its own header + // would name, and a mismatch fails its entire listing rather than skipping + // the one session. A copy given a new id must not keep the old one. + let root = tempfile::tempdir().unwrap(); + let store = dsh::DshStore::new(root.path()); + let mut transcript = saveable("session-abc", "/repo"); + transcript.meta.id = "session-copy".into(); + transcript.meta.cwd = Some("/elsewhere".into()); + + let saved = store.save(&transcript).unwrap(); + assert_eq!(saved.id, "session-copy"); + assert_eq!( + saved.reference, + root.path().join("--elsewhere--").join("session-copy") + ); + + let header: serde_json::Value = + serde_json::from_str(first_frame(&saved.reference.join("session.jsonl.zstd")).trim_end()) + .unwrap(); + assert_eq!(header["id"], json!("session-copy")); + assert_eq!(header["cwd"], json!("/elsewhere")); +} + +#[test] +fn save_follows_the_encoding_the_root_already_uses() { + // A root that mixes `.jsonl` and `.jsonl.zstd` does not merely hide the + // odd session: dsh refuses to list *any* session in it, under either + // configuration. Writing the default encoding into a plaintext root would + // take the user's whole session list down, so the root decides. + let root = tempfile::tempdir().unwrap(); + write_session( + &root.path().join("--repo--").join("session-old"), + &sample_body(), + ); + + let store = dsh::DshStore::new(root.path()); + let saved = store.save(&saveable("session-abc", "/repo")).unwrap(); + assert!(saved.reference.join("session.jsonl").is_file()); + assert!(!saved.reference.join("session.jsonl.zstd").exists()); + + // And it still reads back. + let found = store.discover().unwrap(); + assert_eq!(found.len(), 2); +} + +#[cfg(feature = "dsh_zstd")] +#[test] +fn saving_under_a_new_cwd_leaves_no_duplicate_id_behind() { + // A project directory is keyed by cwd, so re-saving a session whose cwd + // changed lands it somewhere new. dsh treats one id appearing under two + // project directories as corruption and fails its whole listing, so the + // old copy has to go. + let root = tempfile::tempdir().unwrap(); + let store = dsh::DshStore::new(root.path()); + let first = store.save(&saveable("session-abc", "/a")).unwrap(); + let second = store.save(&saveable("session-abc", "/b")).unwrap(); + + assert_ne!(first.reference, second.reference); + assert!(!first.reference.exists()); + assert_eq!(store.discover().unwrap().len(), 1); +} + +#[cfg(feature = "dsh_zstd")] +#[test] +fn a_failed_save_leaves_the_previous_copy_in_place() { + // Re-saving under a new cwd retires the copy under the old project + // directory — but only once the replacement is on disk. If writing the + // replacement fails, the old session must still be there. + let root = tempfile::tempdir().unwrap(); + let store = dsh::DshStore::new(root.path()); + let first = store.save(&saveable("session-abc", "/a")).unwrap(); + + // Learn where `/b` sessions go, then block that project directory with a + // plain file so the replacement cannot be created. + let scratch = tempfile::tempdir().unwrap(); + let probe = dsh::DshStore::new(scratch.path()) + .save(&saveable("probe", "/b")) + .unwrap(); + let project = probe.reference.parent().unwrap().file_name().unwrap(); + std::fs::write(root.path().join(project), b"").unwrap(); + + assert!(store.save(&saveable("session-abc", "/b")).is_err()); + assert!( + first.reference.join("session.jsonl.zstd").is_file(), + "the previous copy must survive a failed save" + ); + assert_eq!(store.discover().unwrap().len(), 1); +} + +#[cfg(feature = "dsh_zstd")] +#[test] +fn save_refuses_an_id_that_would_escape_the_store() { + let root = tempfile::tempdir().unwrap(); + let store = dsh::DshStore::new(root.path()); + // dsh encodes rather than rejects a traversing id, so the escape attempt + // becomes a literal directory name and stays inside the store. + let saved = store.save(&saveable("../../evil", "/repo")).unwrap(); + assert!(saved.reference.starts_with(root.path())); + // A dot is a safe character; only a segment that is entirely `.` or `..` + // is special-cased. Escaping the separators is what contains the traversal. + assert_eq!(saved.reference.file_name().unwrap(), "..~002F..~002Fevil"); +} + +#[test] +fn discovery_does_not_depend_on_the_directory_name() { + let root = tempfile::tempdir().unwrap(); + write_session( + &root.path().join("workspace-1").join("conversation-7"), + &sample_body(), + ); + let found = dsh::DshStore::new(root.path()).discover().unwrap(); + assert_eq!(found.len(), 1); + assert_eq!(found[0].meta.id, "session-abc"); +} + +/// A v4 log shaped like one dsh 0.2 records: the system prompt is a surface +/// node, dsh injects its runtime context as extra `user/message`s, and tool +/// results are tool-role messages rather than `tool-result` wrappers. +fn v4_body() -> dsh::DshSession { + let t = 1_792_066_443_800_i64; + dsh::DshSession { + header: json!({ + "type": "session", "version": 4, "id": "session-v4", "createdAt": t, + "cwd": "/repo", "isSeeded": false, "delegationDepth": 0 + }), + events: vec![ + json!({"type": "turn/start", "seq": 0, "time": t, "data": {"turn": 1}}), + json!({"type": "step/start", "seq": 1, "time": t, "data": {"turn": 1, "step": 1}}), + json!({"type": "system/message", "seq": 2, "time": t, "data": {"turn": 1, "step": 1, + "message": {"role": "system", "id": "sys", "source": {"kind": "system-prompt"}, + "content": [{"type": "text", "text": "You are an AI agent."}]}}, + "surfaceOp": "append"}), + json!({"type": "user/message", "seq": 3, "time": t + 1, "data": { + "role": "user", "id": "u1", "source": {"kind": "user"}, + "content": [{"type": "text", "text": "count the lines"}]}, "surfaceOp": "append"}), + json!({"type": "user/message", "seq": 4, "time": t + 1, "data": { + "role": "user", "id": "u2", "source": {"kind": "runtime-context", "form": "snapshot"}, + "content": [{"type": "text", "text": "Current runtime context."}]}, "surfaceOp": "append"}), + json!({"type": "assistant/message", "seq": 5, "time": t + 2, "data": {"turn": 1, "step": 1, + "message": {"role": "assistant", "id": "a1", + "source": {"kind": "model", "provider": "deepseek-official", "model": "deepseek-flash"}, + "content": [ + {"type": "reasoning", "text": "read it"}, + {"type": "tool-call", "id": "c1", "name": "read", + "arguments": "{\"file_path\": \"/repo/notes.txt\"}"} + ]}}, "surfaceOp": "append"}), + json!({"type": "tool/call", "seq": 6, "time": t + 2, "data": {"turn": 1, "step": 1, + "callId": "c1", "name": "read", "arguments": "{\"file_path\": \"/repo/notes.txt\"}"}}), + json!({"type": "tool/result", "seq": 7, "time": t + 3, "data": {"turn": 1, "step": 1, + "message": {"role": "tool", "id": "r1", "source": {"kind": "tool", "callId": "c1"}, + "toolCallId": "c1", "isError": false, + "content": [{"type": "text", "text": "1: alpha\n2: beta"}]}}, + "sourceEventSeqs": [6], "surfaceOp": "append"}), + json!({"type": "step/end", "seq": 8, "time": t + 3, "data": {"turn": 1, "step": 1}}), + json!({"type": "step/start", "seq": 9, "time": t + 3, "data": {"turn": 1, "step": 2}}), + json!({"type": "assistant/message", "seq": 10, "time": t + 4, "data": {"turn": 1, "step": 2, + "message": {"role": "assistant", "id": "a2", + "source": {"kind": "model", "provider": "deepseek-official", "model": "deepseek-flash"}, + "content": [{"type": "text", "text": "It has 2 lines."}]}}, "surfaceOp": "append"}), + json!({"type": "step/end", "seq": 11, "time": t + 4, "data": {"turn": 1, "step": 2}}), + json!({"type": "turn/end", "seq": 12, "time": t + 4, + "data": {"turn": 1, "reason": {"kind": "completed"}}}), + ], + } +} + +fn write_log(dir: &std::path::Path, name: &str, body: &dsh::DshSession) { + std::fs::create_dir_all(dir).unwrap(); + let mut text = format!("{}\n", serde_json::to_string(&body.header).unwrap()); + for event in &body.events { + text.push_str(&serde_json::to_string(event).unwrap()); + text.push('\n'); + } + std::fs::write(dir.join(name), text).unwrap(); +} + +fn texts(message: &common::Message) -> Vec { + message + .content + .iter() + .map(|block| match block { + Block::Text { text } => format!("text:{text}"), + Block::Thinking { text, .. } => format!("thinking:{text}"), + Block::ToolUse { id, .. } => format!("tool_use:{id}"), + Block::ToolResult { + tool_use_id, + content, + is_error, + } => format!("tool_result:{tool_use_id}:{is_error}:{content:?}"), + other => format!("{other:?}"), + }) + .collect() +} + +#[test] +fn discovery_reads_the_newest_versioned_generation() { + // dsh 0.2 names each generation `session.vN.jsonl(.zstd)` and, when it + // migrates an old session, writes the new generation beside the original + // instead of replacing it. The highest generation is the live one. + let root = tempfile::tempdir().unwrap(); + let dir = root.path().join("--repo--").join("session-v4"); + let mut stale = sample_body(); + stale.header["id"] = json!("session-v4"); + write_log(&dir, "session.jsonl", &stale); + write_log(&dir, "session.v4.jsonl", &v4_body()); + + let store = dsh::DshStore::new(root.path()); + let found = store.discover().unwrap(); + assert_eq!(found.len(), 1); + let loaded = store.load(&found[0].reference).unwrap(); + assert_eq!(loaded.body.header["version"], json!(4)); +} + +#[test] +fn v4_surface_projects_tool_role_results_and_skips_injected_context() { + let native = txcript::Transcript::new( + common::Meta { + id: "session-v4".into(), + timestamp: ts("2026-10-15T00:00:00Z"), + cwd: Some("/repo".into()), + git_branch: None, + title: None, + cli_version: None, + model: None, + lineage: None, + }, + v4_body(), + ); + let common = dsh::Dsh::to_common(&native).unwrap(); + let got: Vec<(common::Role, Vec)> = common + .body + .iter() + .map(|message| (message.role, texts(message))) + .collect(); + assert_eq!( + got, + vec![ + (common::Role::User, vec!["text:count the lines".to_string()]), + ( + common::Role::Assistant, + vec!["thinking:read it".to_string(), "tool_use:c1".to_string()] + ), + ( + common::Role::User, + vec!["tool_result:c1:false:Text(\"1: alpha\\n2: beta\")".to_string()] + ), + ( + common::Role::Assistant, + vec!["text:It has 2 lines.".to_string()] + ), + ] + ); +} + +#[test] +fn v4_replacement_endpoints_are_start_seq_and_end_seq() { + // Since v3 a replace op names its span `startSeq`/`endSeq`, and the system + // prompt is a surface node the span is counted around. A compaction + // checkpoint replaces the conversation it summarizes. + let mut body = v4_body(); + body.events.push(json!({"type": "user/message", "seq": 13, "time": 1_792_066_443_900_i64, + "data": {"role": "user", "id": "k1", "source": {"kind": "compact-checkpoint", "compactionId": "x"}, + "content": [{"type": "text", "text": "Summary: counted 2 lines."}]}, + "sourceEventSeqs": [3, 4, 5, 7, 10], + "surfaceOp": {"op": "replace", "startSeq": 3, "endSeq": 10}})); + let native = txcript::Transcript::new( + common::Meta { + id: "session-v4".into(), + timestamp: ts("2026-10-15T00:00:00Z"), + cwd: None, + git_branch: None, + title: None, + cli_version: None, + model: None, + lineage: None, + }, + body, + ); + let common = dsh::Dsh::to_common(&native).unwrap(); + assert_eq!(common.body.len(), 1, "{:?}", common.body); + assert_eq!( + texts(&common.body[0]), + vec!["text:Summary: counted 2 lines.".to_string()] + ); +} + +#[test] +fn parent_session_becomes_lineage() { + let root = tempfile::tempdir().unwrap(); + // A subagent run records its parent and `origin: subagent`. + let mut child = v4_body(); + child.header["id"] = json!("child"); + child.header["parentSession"] = json!("session-parent"); + child.header["origin"] = json!("subagent"); + child.header["delegationDepth"] = json!(1); + write_log( + &root.path().join("--repo--").join("child"), + "session.v4.jsonl", + &child, + ); + // A user fork records only the parent, and is seeded with its history. + let mut fork = v4_body(); + fork.header["id"] = json!("session-fork"); + fork.header["parentSession"] = json!("session-parent"); + fork.header["isSeeded"] = json!(true); + write_log( + &root.path().join("--repo--").join("session-fork"), + "session.v4.jsonl", + &fork, + ); + + let mut found = dsh::DshStore::new(root.path()).discover().unwrap(); + found.sort_by(|a, b| a.meta.id.cmp(&b.meta.id)); + let lineage = |i: usize| found[i].meta.lineage.clone().expect("lineage"); + assert_eq!(found[0].meta.id, "child"); + assert_eq!(lineage(0).parent, "session-parent"); + assert_eq!(lineage(0).relation, common::Relation::Spawn); + assert_eq!(found[1].meta.id, "session-fork"); + assert_eq!(lineage(1).relation, common::Relation::Fork); +} + +fn write_session(dir: &std::path::Path, body: &dsh::DshSession) { + std::fs::create_dir_all(dir).unwrap(); + let mut text = format!("{}\n", serde_json::to_string(&body.header).unwrap()); + for event in &body.events { + text.push_str(&serde_json::to_string(event).unwrap()); + text.push('\n'); + } + std::fs::write(dir.join("session.jsonl"), text).unwrap(); +} + +fn common_sample() -> txcript::Transcript { + txcript::Transcript::new( + common::Meta { + id: "session-xyz".into(), + timestamp: ts("2024-01-01T00:00:45.000Z"), + cwd: Some("/repo".into()), + git_branch: None, + title: Some("hi".into()), + cli_version: None, + model: Some("deepseek-v4-flash".into()), + lineage: None, + }, + vec![common::Message { + role: common::Role::User, + content: vec![Block::Text { + text: "hello".into(), + }], + timestamp: ts("2024-01-01T00:00:46.000Z"), + model: None, + stop_reason: None, + usage: None, + }], + ) +} + +#[test] +fn rendering_is_deterministic() { + let a = dsh::Dsh::from_common(&common_sample()).unwrap(); + let b = dsh::Dsh::from_common(&common_sample()).unwrap(); + assert_eq!(a.body, b.body); +} + +#[test] +fn assistant_timestamps_survive_a_common_round_trip() { + let native = dsh::Dsh::from_common(&common_sample()).unwrap(); + let common = dsh::Dsh::to_common(&native).unwrap(); + assert_eq!(common.body[0].timestamp, ts("2024-01-01T00:00:46.000Z")); +} + +fn message(role: common::Role, at: &str, content: Vec) -> common::Message { + common::Message { + role, + content, + timestamp: ts(at), + model: None, + stop_reason: None, + usage: None, + } +} + +fn tool_use(id: &str) -> Block { + Block::ToolUse { + id: id.into(), + tool: common::Tool::from_canonical("bash", json!({ "command": "ls" })), + } +} + +/// Two turns: the first answers one tool call and moves to a second step; +/// the second leaves its call unanswered. +fn tool_session() -> txcript::Transcript { + use common::Role::{Assistant, User}; + let mut transcript = common_sample(); + transcript.body.extend([ + message( + Assistant, + "2024-01-01T00:00:47.000Z", + vec![tool_use("call-1")], + ), + message( + User, + "2024-01-01T00:00:48.000Z", + vec![Block::ToolResult { + tool_use_id: "call-1".into(), + content: common::ToolOutput::Text("a\nb".into()), + is_error: false, + }], + ), + message( + Assistant, + "2024-01-01T00:00:49.000Z", + vec![Block::Text { + text: "two files".into(), + }], + ), + message( + User, + "2024-01-01T00:00:50.000Z", + vec![Block::Text { + text: "again".into(), + }], + ), + message( + Assistant, + "2024-01-01T00:00:51.000Z", + vec![tool_use("call-2")], + ), + ]); + transcript +} + +fn event_types(body: &dsh::DshSession) -> Vec<&str> { + body.events + .iter() + .map(|event| event["type"].as_str().unwrap()) + .collect() +} + +#[test] +fn from_common_writes_the_v4_turn_and_step_lifecycle() { + let native = dsh::Dsh::from_common(&tool_session()).unwrap(); + assert_eq!(native.body.header["version"], json!(4)); + assert_eq!(native.body.header["isSeeded"], json!(false)); + assert_eq!( + event_types(&native.body), + [ + "session/title", + "turn/start", + "step/start", + "system/message", + "user/message", + "assistant/message", + "tool/call", + "tool/result", + "step/end", + "step/start", + "assistant/message", + "step/end", + "turn/end", + "turn/start", + "step/start", + "user/message", + "assistant/message", + "tool/result", + "step/end", + "turn/end", + ] + ); + for (seq, event) in native.body.events.iter().enumerate() { + assert_eq!(event["seq"], json!(seq)); + } + let events = &native.body.events; + // A result cites the `tool/call` that started it, and is a tool-role + // message rather than a wrapped block. + assert_eq!(events[7]["sourceEventSeqs"], json!([6])); + assert_eq!(events[7]["data"]["message"]["role"], json!("tool")); + assert_eq!(events[7]["data"]["message"]["toolCallId"], json!("call-1")); + // An unanswered call is closed with dsh's own not-started repair, and + // the turn it ends is recorded as interrupted. + assert_eq!( + events[17]["data"]["message"]["id"], + json!("interrupted-tool-result-call-2-17") + ); + assert_eq!( + events[17]["data"]["error"]["code"], + json!("TOOL_NOT_STARTED") + ); + assert!(events[17].get("sourceEventSeqs").is_none()); + assert_eq!(events[19]["data"]["reason"]["kind"], json!("interrupted")); + assert_eq!(events[12]["data"]["reason"]["kind"], json!("completed")); + // Every model source names a provider and model, and the stream is + // present but empty. + assert_eq!(events[5]["data"]["stream"], json!([])); + assert_eq!( + events[5]["data"]["message"]["source"]["model"], + json!("deepseek-v4-flash") + ); +} + +#[test] +fn from_common_round_trips_tool_calls_and_results() { + let native = dsh::Dsh::from_common(&tool_session()).unwrap(); + let common = dsh::Dsh::to_common(&native).unwrap(); + let blocks: Vec<&Block> = common.body.iter().flat_map(|m| &m.content).collect(); + assert!(matches!(blocks[1], Block::ToolUse { id, .. } if id == "call-1")); + assert!(matches!( + blocks[2], + Block::ToolResult { tool_use_id, is_error: false, .. } if tool_use_id == "call-1" + )); + assert!(matches!( + blocks.last().unwrap(), + Block::ToolResult { tool_use_id, is_error: true, .. } if tool_use_id == "call-2" + )); +} + +#[test] +fn save_names_the_log_for_its_header_version() { + let root = tempfile::tempdir().unwrap(); + let store = dsh::DshStore::new(root.path()); + // Fix the root to plaintext so the test does not depend on `dsh_zstd`. + write_session( + &root.path().join("--elsewhere--").join("session-other"), + &sample_body(), + ); + + let native = dsh::Dsh::from_common(&tool_session()).unwrap(); + let saved = store.save(&native).unwrap(); + assert!(saved.reference.join("session.v4.jsonl").is_file()); + + // A re-save replaces the session, so another generation beside it goes. + std::fs::write(saved.reference.join("session.v1.jsonl"), "stale").unwrap(); + store.save(&native).unwrap(); + assert!(!saved.reference.join("session.v1.jsonl").exists()); + + // A native body keeps its own version, and so its own file name. + let mut older = native.clone(); + older.body.header["version"] = json!(1); + store.save(&older).unwrap(); + assert!(saved.reference.join("session.v1.jsonl").is_file()); + assert!(!saved.reference.join("session.v4.jsonl").exists()); +} diff --git a/tests/integration/main.rs b/tests/integration/main.rs index 4eb9cce..09d4d3d 100644 --- a/tests/integration/main.rs +++ b/tests/integration/main.rs @@ -16,6 +16,7 @@ mod cowork; mod cross_harness; mod cursor; mod cursor_desktop; +mod dsh; mod fx; mod grok; mod grok_bot; diff --git a/tests/integration/path_safety.rs b/tests/integration/path_safety.rs index 64b82bc..c23d481 100644 --- a/tests/integration/path_safety.rs +++ b/tests/integration/path_safety.rs @@ -8,7 +8,7 @@ use std::path::{Path, PathBuf}; use chrono::{TimeZone, Utc}; use txcript::common::{Block, Message, Meta, Role}; use txcript::harness::{ - amp, antigravity, campfire, claude_code, codex, cowork, cursor, fx, grok, grok_bot, pi, + amp, antigravity, campfire, claude_code, codex, cowork, cursor, dsh, fx, grok, grok_bot, pi, }; use txcript::{Codec, Common, Store, Transcript}; @@ -84,6 +84,7 @@ fn hostile_ids_cannot_escape_any_file_backed_store() { assert_save_confined(&cursor::CursorStore::new(root.to_path_buf()), root); assert_save_confined(&grok::GrokStore::new(root.to_path_buf()), root); assert_save_confined(&grok_bot::GrokBotStore::new(root.to_path_buf()), root); + assert_save_confined(&dsh::DshStore::new(root.to_path_buf()), root); assert_save_confined(&::AmpStore::new(root.to_path_buf()), root); assert_save_confined( &antigravity::AntigravityStore::new(root.to_path_buf()), @@ -174,6 +175,59 @@ fn grok_delete_refuses_a_symlink_into_the_sessions_root() { ); } +#[test] +fn dsh_delete_refuses_paths_outside_the_sessions_root() { + let sessions = tempfile::tempdir().unwrap(); + let victim = tempfile::tempdir().unwrap(); + // Give the victim a session's shape: the guard must hold even when the + // directory looks like a real session. + std::fs::write(victim.path().join("session.jsonl"), b"").unwrap(); + + let store = dsh::DshStore::new(sessions.path()); + assert!( + store.delete(&victim.path().to_path_buf()).is_err(), + "a session-shaped directory outside the root must be refused" + ); + assert!(victim.path().is_dir(), "the foreign directory must survive"); + + // A project directory holds sessions but isn't one itself. + let project = sessions.path().join("--proj--"); + std::fs::create_dir_all(&project).unwrap(); + std::fs::write(project.join("session.jsonl"), b"").unwrap(); + assert!(store.delete(&project).is_err()); + assert!(project.is_dir()); + + // A directory with no session log isn't a session at all. + let bare = project.join("bare"); + std::fs::create_dir_all(&bare).unwrap(); + assert!(store.delete(&bare).is_err()); + assert!(bare.is_dir()); +} + +#[cfg(unix)] +#[test] +fn dsh_delete_refuses_a_symlink_into_the_sessions_root() { + let sessions = tempfile::tempdir().unwrap(); + let victim = tempfile::tempdir().unwrap(); + std::fs::write(victim.path().join("session.jsonl.zstd"), b"").unwrap(); + + // Plant a link at the exact depth delete expects a session at. + let project = sessions.path().join("--proj--"); + std::fs::create_dir_all(&project).unwrap(); + let link = project.join("linked-session"); + std::os::unix::fs::symlink(victim.path(), &link).unwrap(); + + let store = dsh::DshStore::new(sessions.path()); + assert!( + store.delete(&link).is_err(), + "a symlink resolving outside the root must be refused" + ); + assert!( + victim.path().join("session.jsonl.zstd").is_file(), + "the symlink target must survive" + ); +} + #[test] fn antigravity_delete_refuses_paths_outside_the_conversations_root() { let root = tempfile::tempdir().unwrap(); diff --git a/tests/integration/properties.rs b/tests/integration/properties.rs index 176c2da..8220953 100644 --- a/tests/integration/properties.rs +++ b/tests/integration/properties.rs @@ -18,7 +18,7 @@ use proptest::test_runner::TestCaseError; use serde_json::json; use txcript::common::{Block, Message, Meta, Role, Tool, ToolOutput}; use txcript::harness::{ - amp, antigravity, campfire, claude_code, codex, cowork, cursor, cursor_desktop, fx, grok, + amp, antigravity, campfire, claude_code, codex, cowork, cursor, cursor_desktop, dsh, fx, grok, grok_bot, hermes, opencode, pi, simple, }; use txcript::{Codec, Common, Transcript}; @@ -274,6 +274,7 @@ proptest! { assert_fixpoint::("cursor_desktop", &common)?; assert_fixpoint::("grok", &common)?; assert_fixpoint::("grok_bot", &common)?; + assert_fixpoint::("dsh", &common)?; assert_fixpoint::("fx", &common)?; assert_fixpoint::("hermes", &common)?; assert_fixpoint::("amp", &common)?; diff --git a/tests/integration/store_delete.rs b/tests/integration/store_delete.rs index 7dd918d..7b1cb64 100644 --- a/tests/integration/store_delete.rs +++ b/tests/integration/store_delete.rs @@ -185,6 +185,18 @@ fn grok_bot_delete_removes_the_session_directory() { assert!(leftovers.is_empty(), "no files left behind: {leftovers:?}"); } +#[cfg(feature = "dsh_zstd")] +#[test] +fn dsh_delete_removes_the_session_directory() { + let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let store = txcript::harness::dsh::DshStore::new(dir.path().to_path_buf()); + roundtrip(&store); + // dsh scans project directories for session directories, so an empty + // project directory is harmless — a leftover log is not. + let leftovers: Vec<_> = walk_files(dir.path()); + assert!(leftovers.is_empty(), "no log left behind: {leftovers:?}"); +} + #[cfg(feature = "opencode")] #[test] fn cursor_delete_removes_the_whole_session_dir() {