From e8607f4d74cebe2b7ff9db2a1ab59e0fd4c29df5 Mon Sep 17 00:00:00 2001 From: zzyking Date: Sat, 29 Aug 2026 23:34:40 +0800 Subject: [PATCH 1/4] feat: add read-only DeepSeek Harness (dsh) Discover, list, search, view, export, and convert from dsh sessions under $DSH_HOME/sessions (default ~/.dsh/sessions). The store is read-only: dsh has no documented session import command and its persistence seam does not delete logs. Logs are append-only JSONL, Zstandard-framed by default. The first line is a type:session header; the Common projection rebuilds the ordered surface (user/message, assistant/message, tool/result) and leaves packed chunk rows and unknown events in the native body. --- Cargo.toml | 6 + README.md | 1 + cli/src/lib.rs | 8 +- docs/formats/README.md | 1 + docs/formats/dsh.md | 77 +++ docs/usage.md | 3 +- examples/search_bench.rs | 9 +- src/harness/dsh.rs | 817 +++++++++++++++++++++++++++++ src/harness/mod.rs | 1 + src/lib.rs | 2 +- src/local.rs | 15 +- src/transcript.rs | 8 +- src/wasm.rs | 4 +- tests/integration/cross_harness.rs | 12 +- tests/integration/dsh.rs | 199 +++++++ tests/integration/main.rs | 1 + tests/integration/properties.rs | 3 +- 17 files changed, 1157 insertions(+), 10 deletions(-) create mode 100644 docs/formats/dsh.md create mode 100644 src/harness/dsh.rs create mode 100644 tests/integration/dsh.rs diff --git a/Cargo.toml b/Cargo.toml index 5d9f64d..57ba833 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,6 +63,12 @@ wasm-bindgen = { version = "0.2.126", optional = true } # Decoding the rollouts Codex compresses; the library Codex writes them with. zstd = { version = "0.13.3", default-features = false, optional = true } +# DeepSeek Harness stores session logs as concatenated Zstandard frames by +# default. The codec's text form is uncompressed JSON; only the native store +# decoder needs this, and wasm builds never open those files. +[target.'cfg(not(target_arch = "wasm32"))'.dependencies] +zstd = "0.13.3" + [target.'cfg(target_os = "macos")'.dependencies] aes = { version = "0.8.4", optional = true } cbc = { version = "0.1.2", features = ["alloc", "block-padding"], optional = true } diff --git a/README.md b/README.md index 9727819..aee94e7 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 | No | | [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..05e4c08 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. @@ -1318,6 +1318,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 @@ -1919,6 +1920,11 @@ fn ensure_resumable_source(source: HarnessId, target: HarnessId) -> Result<(), S "ChatGPT is pull-only: choose another --with harness; txcript never continues conversations in ChatGPT" .to_string(), ) + } else if source == HarnessId::Dsh && target == HarnessId::Dsh { + Err( + "DeepSeek Harness is read-only: choose another --with harness; txcript never writes dsh session logs" + .to_string(), + ) } else { Ok(()) } 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..c5d54e9 --- /dev/null +++ b/docs/formats/dsh.md @@ -0,0 +1,77 @@ +# DeepSeek Harness (`dsh`) + +DeepSeek Harness stores sessions under: + +```text +$DSH_HOME/sessions/ # default ~/.dsh/sessions + ----/ # or _no-cwd/ + / + session.jsonl.zstd # default: concatenated checksummed Zstandard frames + session.jsonl # only when compression: 'none' +``` + +Home resolution is configured path, then `$DSH_HOME`, then `~/.dsh`. An empty +`$DSH_HOME` is treated as unset. + +## 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`) 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. + +The official on-disk format version is `0`. There is no migration; txcript +still loads the native body when the version field differs. + +### Header + +The first line is tagged `type: "session"` and carries `version`, `id`, +`createdAt` (epoch milliseconds), optional `cwd`, `parentSession`, +`seedLength`, `origin`, `delegationDepth`, and `agentPreset`. + +### Events + +Each event is `{ type, seq, time, data, ... }`. Surface events +(`user/message`, `assistant/message`, `tool/result`) may also carry +`surfaceOp` (`"append"` or `{ op: "replace", start, end }`) and +`sourceEventSeqs`. Log-only events (turn/step markers, `assistant/chunk`, +`request/header`, packed chunk rows, …) never enter the Common conversation. + +## Common projection + +| dsh event | Common representation | +| --- | --- | +| `user/message` with text parts | user text message | +| `assistant/message` `reasoning` / `text` / `tool-call` | thinking / text / tool-use | +| `tool/result` | user tool-result (`isError` kept) | + +The ordered surface is rebuilt before projection: a `replace` `surfaceOp` +truncates earlier surface nodes. Packed chunk rows and `assistant/chunk` +stream events are ignored for Common because the assembled `assistant/message` +already carries the step. + +## Store capabilities + +DeepSeek Harness has no documented session-import CLI. The persistence seam +also has no delete API. The txcript dsh store is therefore **read-only**: + +- `list`, `query`, `view`, and `export` work; +- a dsh session can be converted into any writable target harness; +- `save` and `delete` return an explicit read-only error; +- txcript never writes directly into `~/.dsh/sessions`. + +The native resume command documented for a TUI profile is +`dsh --profile tui --resume `. Cross-harness continuation is supported; +native dsh continuation is refused. + +## Provenance + +Open source. Layout and event vocabulary follow the DeepSeek Harness +packages `@deepseek-ai/dsh-session` and +`@deepseek-ai/dsh-session-persistence-jsonl` (session format version 0, +developer preview; the project warns of compatibility-breaking changes). +The reader was also checked against a local `session.jsonl.zstd` written by +dsh around 2026-08-14. + +Last verified: 2026-08-29 diff --git a/docs/usage.md b/docs/usage.md index e9ad6c8..7de83f4 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -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..983ef56 --- /dev/null +++ b/src/harness/dsh.rs @@ -0,0 +1,817 @@ +//! `dsh` sessions: `$DSH_HOME/sessions` (default `~/.dsh/sessions`). +//! +//! Official persistence writes one append-only JSONL log per session, Zstandard +//! framed by default (`session.jsonl.zstd`). The first line is a `type: session` +//! header; the rest are `SessionEvent` records (and packed `*-chunks` rows). +//! There is no documented session-import CLI, the on-disk format is version 0 +//! with no migration, and the persistence seam has no delete API, so +//! [`DshStore`] is read-only: sessions can be discovered, loaded, searched, +//! exported, and converted into another harness, but txcript never writes +//! `~/.dsh/sessions`. +//! +//! 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 (`user/message`, `assistant/message`, `tool/result`). + +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, Message, Meta, Role, StopReason, Tool, ToolOutput}; +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 newly-written `SessionHeader`. +/// Official readers refuse any other value; txcript still loads the native +/// body so unknown future logs are not silently dropped. +const SESSION_FORMAT_VERSION: u64 = 0; + +/// 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-only 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.jsonl 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> { + Err(read_only_error()) + } + + fn delete(&self, _reference: &PathBuf) -> Result<()> { + Err(read_only_error()) + } + + 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) + } +} + +fn read_only_error() -> Error { + Error::Unconvertible { + harness: Dsh::NAME, + detail: "dsh session storage is read-only in txcript; dsh has no documented session import command and its persistence seam does not delete logs".to_string(), + } +} + +fn log_path(session_dir: &Path) -> Option { + let zstd = session_dir.join("session.jsonl.zstd"); + if zstd.is_file() { + return Some(zstd); + } + let plain = session_dir.join("session.jsonl"); + plain.is_file().then_some(plain) +} + +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(not(target_arch = "wasm32"))] +fn decode_zstd(bytes: &[u8]) -> Result { + use std::io::Read; + + // Official JSONL backend concatenates independent Zstandard frames + // (header frame, then one frame per append batch). Decode each. + let starts = zstd_frame_starts(bytes); + let starts = if starts.is_empty() { vec![0] } else { starts }; + let mut out = Vec::new(); + for (index, start) in starts.iter().enumerate() { + let end = starts.get(index + 1).copied().unwrap_or(bytes.len()); + let mut decoder = + zstd::stream::read::Decoder::new(std::io::Cursor::new(&bytes[*start..end]))?; + decoder.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(target_arch = "wasm32"))] +fn zstd_frame_starts(bytes: &[u8]) -> Vec { + const MAGIC: [u8; 4] = [0x28, 0xB5, 0x2F, 0xFD]; + let mut starts = Vec::new(); + let mut index = 0; + while index + 4 <= bytes.len() { + if bytes[index..index + 4] == MAGIC { + starts.push(index); + index += 4; + } else { + index += 1; + } + } + starts +} + +#[cfg(target_arch = "wasm32")] +fn decode_zstd(_bytes: &[u8]) -> Result { + Err(Error::Malformed { + harness: Dsh::NAME, + detail: "zstd session logs cannot be decoded in wasm".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, + // Not yet read from the header's `parentSession`. + lineage: None, + } +} + +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) +} + +fn json_usize(value: Option<&Value>) -> Option { + value + .and_then(Value::as_u64) + .and_then(|n| usize::try_from(n).ok()) +} + +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) +} + +/// 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 { + let mut surface: Vec<&Value> = Vec::new(); + for event in events { + let Some(kind) = event.get("type").and_then(Value::as_str) else { + continue; + }; + if !matches!(kind, "user/message" | "assistant/message" | "tool/result") { + continue; + } + match event.get("surfaceOp") { + Some(Value::Object(op)) if op.get("op").and_then(Value::as_str) == Some("replace") => { + let start = json_usize(op.get("start")).unwrap_or(0); + let end = json_usize(op.get("end")).unwrap_or(start); + if start < surface.len() { + let end = end.min(surface.len().saturating_sub(1)).max(start); + surface.drain(start..=end); + surface.insert(start, event); + } else { + surface.push(event); + } + } + _ => surface.push(event), + } + } + + 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") => { + 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 { .. })); + messages.push(Message { + role: Role::Assistant, + content: blocks, + timestamp, + model, + stop_reason: Some(if has_tool { + StopReason::ToolUse + } else { + StopReason::EndTurn + }), + usage: None, + }); + } + Some("tool/result") => { + if let Some(block) = tool_result_block(event.get("data")) { + messages.push(user_message(vec![block], timestamp)); + } + } + _ => {} + } + } + messages +} + +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"))?; + let parts = message.get("content")?.as_array()?; + let result = parts + .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, +]); + +#[allow(clippy::too_many_lines)] +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, + "delegationDepth": 0, + }); + if let Some(cwd) = &transcript.meta.cwd { + header["cwd"] = json!(cwd); + } + let mut events = Vec::new(); + let mut seq = 0_u64; + if let Some(title) = &transcript.meta.title { + events.push(json!({ + "type": "session/title", + "seq": seq, + "time": created, + "data": { "title": title, "source": { "kind": "fallback" } } + })); + seq += 1; + } + for (index, message) in transcript.body.iter().enumerate() { + let time = message.timestamp.timestamp_millis(); + if message.role == Role::User { + 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() { + let msg_id = Uuid::new_v5(&NS, format!("{id}:{index}:user").as_bytes()).to_string(); + events.push(json!({ + "type": "user/message", + "seq": seq, + "time": time, + "data": { + "role": "user", + "id": msg_id, + "content": texts, + "source": { "kind": "user" } + }, + "surfaceOp": "append" + })); + seq += 1; + } + for block in &message.content { + if let Block::ToolResult { + tool_use_id, + content, + is_error, + } = block + { + let text = match content { + ToolOutput::Text(text) => text.clone(), + ToolOutput::Json(value) => value.to_string(), + }; + events.push(json!({ + "type": "tool/result", + "seq": seq, + "time": time, + "data": { + "turn": 1, + "step": 1, + "message": { + "role": "user", + "source": { "kind": "tool", "callId": tool_use_id }, + "content": [{ + "type": "tool-result", + "toolCallId": tool_use_id, + "isError": is_error, + "content": [{ "type": "text", "text": text }] + }] + } + }, + "sourceEventSeqs": [], + "surfaceOp": "append" + })); + seq += 1; + } + } + } else { + let mut content = 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 + })); + } + Block::ToolResult { .. } | Block::Image { .. } | Block::Artifact { .. } => {} + } + } + if content.is_empty() { + continue; + } + let msg_id = + Uuid::new_v5(&NS, format!("{id}:{index}:assistant").as_bytes()).to_string(); + let mut source = json!({ "kind": "model" }); + if let Some(model) = &message.model { + source["model"] = json!(model); + } + events.push(json!({ + "type": "assistant/message", + "seq": seq, + "time": time, + "data": { + "turn": 1, + "step": index + 1, + "message": { + "role": "assistant", + "id": msg_id, + "content": content, + "source": source + } + }, + "surfaceOp": "append" + })); + seq += 1; + } + } + let _ = seq; + DshSession { header, 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 + )); + } + + #[test] + fn surface_replace_drops_shadowed_nodes() { + let events = vec![ + json!({"type": "user/message", "seq": 0, "time": 1, "data": { + "content": [{"type": "text", "text": "old"}] + }, "surfaceOp": "append"}), + json!({"type": "user/message", "seq": 1, "time": 2, "data": { + "content": [{"type": "text", "text": "kept"}] + }, "surfaceOp": "append"}), + json!({"type": "user/message", "seq": 2, "time": 3, "data": { + "content": [{"type": "text", "text": "summary"}] + }, "surfaceOp": {"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 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..6e5e88d 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,10 @@ pub fn write_with( }) } } + HarnessId::Dsh => Err(Error::Unconvertible { + harness: "dsh", + detail: "DeepSeek Harness session storage is read-only and dsh has no documented session import command; sessions can be converted from dsh, but not continued into it".to_string(), + }), HarnessId::Fx => go( fx::FxStore::default_root(), fx::FxStore::new, @@ -1119,6 +1128,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..7353b91 --- /dev/null +++ b/tests/integration/dsh.rs @@ -0,0 +1,199 @@ +#![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 read_only_store_discovers_and_refuses_writes() { + 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()); + assert!(store.save(&loaded).is_err()); + assert!(store.delete(&found[0].reference).is_err()); +} + +#[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"); +} + +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")); +} 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/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)?; From 71e37ed4a969e8fd59d95b6e3a3d7cc6bec7b624 Mon Sep 17 00:00:00 2001 From: zzyking Date: Mon, 7 Sep 2026 19:58:59 +0800 Subject: [PATCH 2/4] fix(dsh): resolve surface replacements by seq, keep usage and interruption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four review findings on the dsh harness, all confirmed against the official `@deepseek-ai/dsh-session` sources: - `surfaceOp` replace endpoints are event seqs, not surface-vector indexes. Log-only events sit between surface nodes, so the two diverge immediately and compaction shadowed the wrong range. The surface now tracks each node's seq and resolves both endpoints by seq, matching `replacementRange` in dsh's own `surface.ts`. The previous test masked this by using seqs that happened to equal vector indexes. - Durable `assistant/message` `usage` and `interrupted` were dropped. Map `interrupted: true` to `StopReason::Aborted` and translate the `TokenUsage` counts, in both directions. `reasoningTokens` has no Common counterpart and stays in the native body. - `std::io::Read` was imported unconditionally but only used by the native `decode_zstd`, failing the wasm32 job under `-D warnings`. - Zstandard frames were split by scanning for frame magic, which also occurs inside compressed payloads and checksums. The streaming decoder already walks frame structure and continues across boundaries, so decode in one pass. Also fixes an invariant violation in the writer: `tool/result` emitted `sourceEventSeqs: []`, which dsh's `assertProvenance` rejects — empty arrays are legal only on `assistant/message`. Verified by feeding a regenerated log from a real session to the official `foldSurface`, which now passes. --- docs/formats/dsh.md | 18 ++- src/harness/dsh.rs | 283 +++++++++++++++++++++++++++++++++----------- 2 files changed, 227 insertions(+), 74 deletions(-) diff --git a/docs/formats/dsh.md b/docs/formats/dsh.md index c5d54e9..549acf8 100644 --- a/docs/formats/dsh.md +++ b/docs/formats/dsh.md @@ -44,12 +44,22 @@ Each event is `{ type, seq, time, data, ... }`. Surface events | --- | --- | | `user/message` with text parts | user text message | | `assistant/message` `reasoning` / `text` / `tool-call` | thinking / text / tool-use | +| `assistant/message` `usage` / `interrupted` | `Usage` / `StopReason::Aborted` | | `tool/result` | user tool-result (`isError` kept) | -The ordered surface is rebuilt before projection: a `replace` `surfaceOp` -truncates earlier surface nodes. Packed chunk rows and `assistant/chunk` -stream events are ignored for Common because the assembled `assistant/message` -already carries the step. +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` `surfaceOp` names the inclusive **seq** range +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. Packed chunk rows and `assistant/chunk` stream events are ignored for +Common because the assembled `assistant/message` already carries the step. + +`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. ## Store capabilities diff --git a/src/harness/dsh.rs b/src/harness/dsh.rs index 983ef56..bdf1299 100644 --- a/src/harness/dsh.rs +++ b/src/harness/dsh.rs @@ -22,7 +22,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use uuid::Uuid; -use crate::common::{Block, Message, Meta, Role, StopReason, Tool, ToolOutput}; +use crate::common::{Block, Message, Meta, 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}; @@ -226,41 +226,23 @@ fn read_log_text(path: &Path) -> Result { #[cfg(not(target_arch = "wasm32"))] fn decode_zstd(bytes: &[u8]) -> Result { - use std::io::Read; - - // Official JSONL backend concatenates independent Zstandard frames - // (header frame, then one frame per append batch). Decode each. - let starts = zstd_frame_starts(bytes); - let starts = if starts.is_empty() { vec![0] } else { starts }; + 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(); - for (index, start) in starts.iter().enumerate() { - let end = starts.get(index + 1).copied().unwrap_or(bytes.len()); - let mut decoder = - zstd::stream::read::Decoder::new(std::io::Cursor::new(&bytes[*start..end]))?; - decoder.read_to_end(&mut out)?; - } + 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(target_arch = "wasm32"))] -fn zstd_frame_starts(bytes: &[u8]) -> Vec { - const MAGIC: [u8; 4] = [0x28, 0xB5, 0x2F, 0xFD]; - let mut starts = Vec::new(); - let mut index = 0; - while index + 4 <= bytes.len() { - if bytes[index..index + 4] == MAGIC { - starts.push(index); - index += 4; - } else { - index += 1; - } - } - starts -} - #[cfg(target_arch = "wasm32")] fn decode_zstd(_bytes: &[u8]) -> Result { Err(Error::Malformed { @@ -352,10 +334,17 @@ fn first_event_time(events: &[Value]) -> Option> { .and_then(DateTime::from_timestamp_millis) } -fn json_usize(value: Option<&Value>) -> Option { - value - .and_then(Value::as_u64) - .and_then(|n| usize::try_from(n).ok()) +/// `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 { @@ -366,10 +355,31 @@ fn event_time(event: &Value, fallback: DateTime) -> DateTime { .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)) + }; + let start = position(op.get("start"))?; + let end = position(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 { - let mut surface: Vec<&Value> = Vec::new(); + // 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; @@ -377,24 +387,25 @@ fn events_to_messages(events: &[Value], fallback: DateTime) -> Vec if !matches!(kind, "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") => { - let start = json_usize(op.get("start")).unwrap_or(0); - let end = json_usize(op.get("end")).unwrap_or(start); - if start < surface.len() { - let end = end.min(surface.len().saturating_sub(1)).max(start); - surface.drain(start..=end); - surface.insert(start, event); - } else { - surface.push(event); + 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(event), + _ => surface.push(node), } } let mut messages = Vec::new(); - for event in surface { + for (_, event) in surface { let timestamp = event_time(event, fallback); match event.get("type").and_then(Value::as_str) { Some("user/message") => { @@ -421,17 +432,26 @@ fn events_to_messages(events: &[Value], fallback: DateTime) -> Vec 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(if has_tool { - StopReason::ToolUse - } else { - StopReason::EndTurn - }), - usage: None, + stop_reason: Some(stop_reason), + usage: usage_from_data(data), }); } Some("tool/result") => { @@ -654,7 +674,8 @@ fn body_from_common(transcript: &Transcript) -> DshSession { }] } }, - "sourceEventSeqs": [], + // `sourceEventSeqs` is omitted, not empty: dsh rejects + // an empty array on anything but `assistant/message`. "surfaceOp": "append" })); seq += 1; @@ -691,20 +712,39 @@ fn body_from_common(transcript: &Transcript) -> DshSession { if let Some(model) = &message.model { source["model"] = json!(model); } + let mut data = json!({ + "turn": 1, + "step": index + 1, + "message": { + "role": "assistant", + "id": msg_id, + "content": content, + "source": source + } + }); + 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 turn. + if message.stop_reason == Some(StopReason::Aborted) { + data["interrupted"] = json!(true); + } events.push(json!({ "type": "assistant/message", "seq": seq, "time": time, - "data": { - "turn": 1, - "step": index + 1, - "message": { - "role": "assistant", - "id": msg_id, - "content": content, - "source": source - } - }, + "data": data, "surfaceOp": "append" })); seq += 1; @@ -790,18 +830,25 @@ mod tests { )); } + 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![ - json!({"type": "user/message", "seq": 0, "time": 1, "data": { - "content": [{"type": "text", "text": "old"}] - }, "surfaceOp": "append"}), - json!({"type": "user/message", "seq": 1, "time": 2, "data": { - "content": [{"type": "text", "text": "kept"}] - }, "surfaceOp": "append"}), - json!({"type": "user/message", "seq": 2, "time": 3, "data": { - "content": [{"type": "text", "text": "summary"}] - }, "surfaceOp": {"op": "replace", "start": 0, "end": 0}}), + 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); @@ -809,6 +856,102 @@ mod tests { 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(not(target_arch = "wasm32"))] + #[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 parse_log_requires_session_header() { let err = parse_log("{\"type\":\"user/message\"}\n").unwrap_err(); From d9197fbc533f0559e0c902dff558149123c1644f Mon Sep 17 00:00:00 2001 From: zzyking Date: Thu, 15 Oct 2026 14:53:43 +0800 Subject: [PATCH 3/4] build(dsh): gate zstd decoding behind an optional `dsh_zstd` feature MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dsh store pulled zstd in as an unconditional non-wasm dependency. Codex now declares the same crate as optional behind `codex_zstd`, and Cargo merges the two declarations, so the unconditional one would turn zstd on for every native build and make that feature meaningless. Follow the Codex pattern instead: `dsh_zstd = ["dep:zstd"]`, on by default. Without it — including the wasm build — `session.jsonl.zstd` logs fail to load and drop out of discovery, while `session.jsonl` logs read as before. --- Cargo.toml | 15 +++++++-------- docs/formats/dsh.md | 4 ++++ docs/usage.md | 2 +- src/harness/dsh.rs | 8 ++++---- 4 files changed, 16 insertions(+), 13 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 57ba833..b65795a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,15 +60,10 @@ 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 } -# DeepSeek Harness stores session logs as concatenated Zstandard frames by -# default. The codec's text form is uncompressed JSON; only the native store -# decoder needs this, and wasm builds never open those files. -[target.'cfg(not(target_arch = "wasm32"))'.dependencies] -zstd = "0.13.3" - [target.'cfg(target_os = "macos")'.dependencies] aes = { version = "0.8.4", optional = true } cbc = { version = "0.1.2", features = ["alloc", "block-padding"], optional = true } @@ -110,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"] @@ -146,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/docs/formats/dsh.md b/docs/formats/dsh.md index 549acf8..dbb9e0c 100644 --- a/docs/formats/dsh.md +++ b/docs/formats/dsh.md @@ -13,6 +13,10 @@ $DSH_HOME/sessions/ # default ~/.dsh/sessions Home resolution is configured path, then `$DSH_HOME`, then `~/.dsh`. An empty `$DSH_HOME` is treated as unset. +Reading `session.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 `session.jsonl` logs still work. + ## Native representation `txcript::harness::dsh::DshSession` retains the first JSONL line as `header` diff --git a/docs/usage.md b/docs/usage.md index 7de83f4..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: diff --git a/src/harness/dsh.rs b/src/harness/dsh.rs index bdf1299..12bd7cb 100644 --- a/src/harness/dsh.rs +++ b/src/harness/dsh.rs @@ -224,7 +224,7 @@ fn read_log_text(path: &Path) -> Result { }) } -#[cfg(not(target_arch = "wasm32"))] +#[cfg(feature = "dsh_zstd")] fn decode_zstd(bytes: &[u8]) -> Result { use std::io::Read as _; @@ -243,11 +243,11 @@ fn decode_zstd(bytes: &[u8]) -> Result { }) } -#[cfg(target_arch = "wasm32")] +#[cfg(not(feature = "dsh_zstd"))] fn decode_zstd(_bytes: &[u8]) -> Result { Err(Error::Malformed { harness: Dsh::NAME, - detail: "zstd session logs cannot be decoded in wasm".to_string(), + detail: "a compressed dsh session log needs txcript's `dsh_zstd` feature".to_string(), }) } @@ -936,7 +936,7 @@ mod tests { assert!(messages[0].usage.is_none()); } - #[cfg(not(target_arch = "wasm32"))] + #[cfg(feature = "dsh_zstd")] #[test] fn decode_zstd_spans_concatenated_frames() { // The backend writes the header in its own frame, then one frame per From be489cbf2ced0f63236ba9833ceb17be86a8af02 Mon Sep 17 00:00:00 2001 From: zzyking Date: Thu, 15 Oct 2026 20:30:41 +0800 Subject: [PATCH 4/4] feat(dsh): read session format v4 and lineage dsh 0.2 publishes each format generation as its own file (`session[.vN].jsonl[.zstd]`) and keeps older generations beside it. Discovery now picks the highest generation, and the reader understands v0 through v4 directly: `startSeq`/`endSeq` replace endpoints, the `system/message` and `developer/message` surface nodes, and tool-role `tool/result` messages. Environment context dsh injects as user messages (runtime snapshot, skill catalog, instruction files, clocks, ...) is no longer projected into Common. `parentSession` becomes a Fork lineage, or Spawn for a subagent run. Checked against sessions recorded by dsh 0.2.0-rc.2, and against a v0 session read both as-is and after the official catalog migrated it to v4. --- docs/formats/dsh.md | 107 ++++++++++++++------ src/harness/dsh.rs | 170 +++++++++++++++++++++++++++----- tests/integration/dsh.rs | 206 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 428 insertions(+), 55 deletions(-) diff --git a/docs/formats/dsh.md b/docs/formats/dsh.md index dbb9e0c..1d3f43f 100644 --- a/docs/formats/dsh.md +++ b/docs/formats/dsh.md @@ -3,61 +3,92 @@ DeepSeek Harness stores sessions under: ```text -$DSH_HOME/sessions/ # default ~/.dsh/sessions - ----/ # or _no-cwd/ +$DSH_HOME/sessions/ # default ~/.dsh/sessions + ----/ # or _no-cwd/ / - session.jsonl.zstd # default: concatenated checksummed Zstandard frames - session.jsonl # only when compression: 'none' + 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. -Reading `session.jsonl.zstd` needs txcript's `dsh_zstd` feature, on by default. +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 `session.jsonl` logs still work. +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`) 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. +(`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: -The official on-disk format version is `0`. There is no migration; txcript -still loads the native body when the version field differs. +| 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), optional `cwd`, `parentSession`, -`seedLength`, `origin`, `delegationDepth`, and `agentPreset`. +`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 -(`user/message`, `assistant/message`, `tool/result`) may also carry -`surfaceOp` (`"append"` or `{ op: "replace", start, end }`) and -`sourceEventSeqs`. Log-only events (turn/step markers, `assistant/chunk`, -`request/header`, packed chunk rows, …) never enter the Common conversation. +(`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 | +| `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) | +| `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` `surfaceOp` names the inclusive **seq** range -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. Packed chunk rows and `assistant/chunk` stream events are ignored for -Common because the assembled `assistant/message` already carries the step. +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` @@ -65,6 +96,14 @@ 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 DeepSeek Harness has no documented session-import CLI. The persistence seam @@ -81,11 +120,17 @@ native dsh continuation is refused. ## Provenance -Open source. Layout and event vocabulary follow the DeepSeek Harness -packages `@deepseek-ai/dsh-session` and -`@deepseek-ai/dsh-session-persistence-jsonl` (session format version 0, -developer preview; the project warns of compatibility-breaking changes). -The reader was also checked against a local `session.jsonl.zstd` written by -dsh around 2026-08-14. +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. -Last verified: 2026-08-29 +Last verified: 2026-10-15, against `@deepseek-ai/dsh` 0.2.0-rc.2. diff --git a/src/harness/dsh.rs b/src/harness/dsh.rs index 12bd7cb..4d7ee74 100644 --- a/src/harness/dsh.rs +++ b/src/harness/dsh.rs @@ -1,17 +1,20 @@ //! `dsh` sessions: `$DSH_HOME/sessions` (default `~/.dsh/sessions`). //! //! Official persistence writes one append-only JSONL log per session, Zstandard -//! framed by default (`session.jsonl.zstd`). The first line is a `type: session` -//! header; the rest are `SessionEvent` records (and packed `*-chunks` rows). -//! There is no documented session-import CLI, the on-disk format is version 0 -//! with no migration, and the persistence seam has no delete API, so -//! [`DshStore`] is read-only: sessions can be discovered, loaded, searched, -//! exported, and converted into another harness, but txcript never writes -//! `~/.dsh/sessions`. +//! 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. There is no documented session-import CLI and the +//! persistence seam has no delete API, so [`DshStore`] is read-only: sessions +//! can be discovered, loaded, searched, exported, and converted into another +//! harness, but txcript never writes `~/.dsh/sessions`. //! //! 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 (`user/message`, `assistant/message`, `tool/result`). +//! 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; @@ -22,7 +25,9 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use uuid::Uuid; -use crate::common::{Block, Message, Meta, Role, StopReason, Tool, ToolOutput, Usage}; +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}; @@ -142,7 +147,7 @@ impl Store for DshStore { fn load(&self, reference: &PathBuf) -> Result> { let log = log_path(reference).ok_or_else(|| Error::Malformed { harness: Dsh::NAME, - detail: format!("no session.jsonl in {}", reference.display()), + detail: format!("no session log in {}", reference.display()), })?; let body = load_body(&log)?; let mut meta = meta_from_body(&body); @@ -178,13 +183,39 @@ fn read_only_error() -> Error { } } +/// 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 { - let zstd = session_dir.join("session.jsonl.zstd"); - if zstd.is_file() { - return Some(zstd); + 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). +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 plain = session_dir.join("session.jsonl"); - plain.is_file().then_some(plain) + 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 { @@ -322,11 +353,30 @@ fn meta_from_body(body: &DshSession) -> Meta { title, cli_version: None, model, - // Not yet read from the header's `parentSession`. - lineage: None, + 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() @@ -366,8 +416,9 @@ fn replacement_range( let seq = seq.and_then(Value::as_u64)?; surface.iter().position(|(node, _)| *node == Some(seq)) }; - let start = position(op.get("start"))?; - let end = position(op.get("end"))?; + // 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)) } @@ -384,7 +435,16 @@ fn events_to_messages(events: &[Value], fallback: DateTime) -> Vec let Some(kind) = event.get("type").and_then(Value::as_str) else { continue; }; - if !matches!(kind, "user/message" | "assistant/message" | "tool/result") { + // 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); @@ -409,6 +469,9 @@ fn events_to_messages(events: &[Value], fallback: DateTime) -> Vec 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)); @@ -465,6 +528,42 @@ fn events_to_messages(events: &[Value], fallback: DateTime) -> Vec 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, @@ -553,10 +652,17 @@ fn assistant_blocks(content: Option<&Value>) -> Vec { fn tool_result_block(data: Option<&Value>) -> Option { let message = data.and_then(|value| value.get("message"))?; - let parts = message.get("content")?.as_array()?; - let result = parts - .iter() - .find(|part| part.get("type").and_then(Value::as_str) == Some("tool-result"))?; + // 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) @@ -952,6 +1058,22 @@ mod tests { 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(); diff --git a/tests/integration/dsh.rs b/tests/integration/dsh.rs index 7353b91..5b65bf8 100644 --- a/tests/integration/dsh.rs +++ b/tests/integration/dsh.rs @@ -149,6 +149,212 @@ fn discovery_does_not_depend_on_the_directory_name() { 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());