diff --git a/bt-daemon/src/transcript_import/cursor.rs b/bt-daemon/src/transcript_import/cursor.rs index 3a274ce2..7a8cdfbb 100644 --- a/bt-daemon/src/transcript_import/cursor.rs +++ b/bt-daemon/src/transcript_import/cursor.rs @@ -1,4 +1,5 @@ use super::{envelope, validate_session_id}; +use crate::translate::cursor::{IMPORT_CHECKPOINT, IMPORT_START, IMPORT_STOP}; use crate::wire::Envelope; use anyhow::{bail, Context}; use serde_json::{json, Value}; @@ -144,7 +145,7 @@ pub(super) fn envelopes_with_snapshot( "start_time_estimated": true, "historical_fidelity": "user_and_assistant_text_and_terminal_status" }); - if event == "stop" { + if event == IMPORT_STOP { if let Some(status) = final_terminal.and_then(|record| record.get("status")) { payload["status"] = status.clone(); } @@ -164,8 +165,8 @@ pub(super) fn envelopes_with_snapshot( envelope("cursor", None, &session_id, event, ts_ms, payload) }; Ok(vec![ - boundary("sessionStart", None), - boundary("ImportCheckpoint", Some(through)), - boundary("stop", Some(through)), + boundary(IMPORT_START, None), + boundary(IMPORT_CHECKPOINT, Some(through)), + boundary(IMPORT_STOP, Some(through)), ]) } diff --git a/bt-daemon/src/transcript_import/mod.rs b/bt-daemon/src/transcript_import/mod.rs index 6a6006ed..c8f258ca 100644 --- a/bt-daemon/src/transcript_import/mod.rs +++ b/bt-daemon/src/transcript_import/mod.rs @@ -705,7 +705,7 @@ mod tests { .iter() .map(|event| event.event.as_str()) .collect::>(), - vec!["sessionStart", "ImportCheckpoint", "stop"] + vec!["_bt_importStart", "_bt_importCheckpoint", "_bt_importStop"] ); assert!(events.iter().all(|event| event.source == "cursor")); assert_eq!( @@ -734,7 +734,10 @@ mod tests { .unwrap(); let events = transcript_envelopes(&transcript, ImportSource::Cursor).unwrap(); - let stop = events.iter().find(|event| event.event == "stop").unwrap(); + let stop = events + .iter() + .find(|event| event.event == "_bt_importStop") + .unwrap(); assert_eq!(stop.payload["status"], "error"); assert_eq!(stop.payload["error_message"], "WritableIterable is closed"); } @@ -755,7 +758,7 @@ mod tests { .iter() .map(|event| event.event.as_str()) .collect::>(), - vec!["sessionStart", "ImportCheckpoint"] + vec!["_bt_importStart", "_bt_importCheckpoint"] ); let snapshot = first[1].payload["_bt_transcript_mirror"]["mirror"] .as_str() @@ -781,7 +784,7 @@ mod tests { .iter() .map(|event| event.event.as_str()) .collect::>(), - vec!["ImportCheckpoint"] + vec!["_bt_importCheckpoint"] ); assert_eq!( grown[0].payload["_bt_transcript_mirror"]["mirror"], @@ -797,7 +800,7 @@ mod tests { .iter() .map(|event| event.event.as_str()) .collect::>(), - vec!["stop"] + vec!["_bt_importStop"] ); std::fs::write(&transcript, b"replacement transcript\n").unwrap(); let snapshot_contents = std::fs::read_to_string(&snapshot).unwrap(); @@ -830,8 +833,8 @@ mod tests { ) .unwrap(); let replacement = tail.poll(false).unwrap(); - assert_eq!(replacement[0].event, "sessionStart"); - assert_eq!(replacement[1].event, "ImportCheckpoint"); + assert_eq!(replacement[0].event, "_bt_importStart"); + assert_eq!(replacement[1].event, "_bt_importCheckpoint"); let replacement_snapshot = replacement[1].payload["_bt_transcript_mirror"]["mirror"] .as_str() .unwrap(); @@ -856,7 +859,7 @@ mod tests { tail.allow_incomplete_final_record_on_shutdown(); assert_eq!( tail.poll(false).unwrap().last().unwrap().event, - "ImportCheckpoint" + "_bt_importCheckpoint" ); use std::io::Write; std::fs::OpenOptions::new() @@ -872,7 +875,7 @@ mod tests { .iter() .map(|event| event.event.as_str()) .collect::>(), - vec!["ImportCheckpoint", "stop"] + vec!["_bt_importCheckpoint", "_bt_importStop"] ); } diff --git a/bt-daemon/src/translate/cursor/hooks.rs b/bt-daemon/src/translate/cursor/hooks.rs new file mode 100644 index 00000000..34c31888 --- /dev/null +++ b/bt-daemon/src/translate/cursor/hooks.rs @@ -0,0 +1,547 @@ +//! Typed Cursor hook payloads and transcript records. +//! +//! The journal keeps every native payload verbatim. The translator decodes it +//! here once per event and works with these types from then on. The shapes +//! follow the payloads the Cursor CLI builds for each hook: a field it always +//! sends is required, and a missing field or one with an unexpected type +//! fails the event with a [`DecodeError`]. Only fields Cursor omits in some +//! cases are optional. Fields the translator does not read are not modelled. +//! +//! Transcript import synthesizes its own `_bt_importStart`, +//! `_bt_importCheckpoint` and `_bt_importStop` events, which carry only what a +//! saved transcript records. The `_bt_` prefix is the daemon's reserved +//! namespace, so these names cannot collide with a native Cursor hook. + +use crate::translate::decode::{decode, DecodeError}; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; + +/// One decoded Cursor hook. +pub(super) struct Hook { + pub kind: CursorHook, + /// Present on every live agent hook; absent on import events. + pub common: Option, + /// Added by the daemon when it mirrors the session transcript. + pub transcript_mirror: Option, +} + +impl Hook { + pub fn decode(event: &str, payload: &Value) -> Result { + let what = format!("{event} hook"); + let kind = CursorHook::decode(event, &what, payload)?; + let common = match kind { + CursorHook::ImportStart(_) + | CursorHook::ImportCheckpoint + | CursorHook::ImportStop(_) + | CursorHook::Other => None, + _ => Some(decode("cursor", &what, payload)?), + }; + let daemon: DaemonFields = decode("cursor", &what, payload)?; + Ok(Self { + kind, + common, + transcript_mirror: daemon.transcript_mirror, + }) + } + + pub fn generation_id(&self) -> Option<&str> { + self.common + .as_ref() + .map(|common| common.generation_id.as_str()) + } + + /// The model the agent reported, preferring its precise id. + pub fn model(&self) -> Option<&str> { + let common = self.common.as_ref()?; + Some(common.model_id.as_deref().unwrap_or(&common.model)) + } + + pub fn model_params(&self) -> Option<&Value> { + self.common.as_ref()?.model_params.as_ref() + } + + /// The working directory, falling back to the first workspace root. + pub fn cwd(&self) -> Option<&str> { + let cwd = match &self.kind { + CursorHook::Tool(_, tool) => tool.cwd.as_deref().filter(|cwd| !cwd.is_empty()), + _ => None, + }; + cwd.or_else(|| { + self.common + .as_ref()? + .workspace_roots + .first() + .map(String::as_str) + }) + } + + /// The native tool call this hook reports, if any. + pub fn tool_use_id(&self) -> Option<&str> { + match &self.kind { + CursorHook::Tool(_, tool) => Some(&tool.tool_use_id), + CursorHook::Specialized(specialized) => specialized.tool_use_id(), + _ => None, + } + } +} + +/// Fields the Cursor CLI sends with every agent hook. +#[derive(Deserialize)] +pub(super) struct HookCommon { + /// Empty for background agents. + pub generation_id: String, + /// `unknown` when the CLI cannot name the model. + pub model: String, + pub model_id: Option, + pub model_params: Option, + pub workspace_roots: Vec, +} + +#[derive(Deserialize)] +struct DaemonFields { + #[serde(rename = "_bt_transcript_mirror")] + transcript_mirror: Option, +} + +/// The daemon's reference to its copy of the transcript at this hook. +#[derive(Deserialize)] +#[serde(untagged)] +pub(super) enum TranscriptMirror { + Captured { + mirror: String, + through: u64, + }, + /// No transcript was available at this hook. + Unavailable(EmptyObject), +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct EmptyObject {} + +/// One Cursor hook, decoded by event name. +pub(super) enum CursorHook { + SessionStart, + BeforeSubmitPrompt(PromptHook), + Tool(ToolPhase, ToolHook), + AfterAgentThought(ThoughtHook), + AfterAgentResponse(ResponseHook), + Stop(StopHook), + SessionEnd(SessionEndHook), + SubagentStart(SubagentStartHook), + SubagentStop(SubagentStopHook), + PreCompact, + Specialized(Specialized), + ImportStart(ImportBoundary), + ImportCheckpoint, + ImportStop(ImportStopHook), + /// An event this translator does not reduce. + Other, +} + +impl CursorHook { + fn decode(event: &str, what: &str, payload: &Value) -> Result { + Ok(match event { + "sessionStart" => Self::SessionStart, + "beforeSubmitPrompt" => Self::BeforeSubmitPrompt(decode("cursor", what, payload)?), + "preToolUse" => Self::Tool(ToolPhase::Pre, decode("cursor", what, payload)?), + "postToolUse" => { + let hook: PostToolHook = decode("cursor", what, payload)?; + Self::Tool(ToolPhase::Post(hook.result), hook.tool) + } + "postToolUseFailure" => { + let hook: ToolFailureHook = decode("cursor", what, payload)?; + Self::Tool(ToolPhase::Failure(hook.failure), hook.tool) + } + "afterAgentThought" => Self::AfterAgentThought(decode("cursor", what, payload)?), + "afterAgentResponse" => Self::AfterAgentResponse(decode("cursor", what, payload)?), + "stop" => Self::Stop(decode("cursor", what, payload)?), + "sessionEnd" => Self::SessionEnd(decode("cursor", what, payload)?), + "subagentStart" => Self::SubagentStart(decode("cursor", what, payload)?), + "subagentStop" => Self::SubagentStop(decode("cursor", what, payload)?), + "preCompact" => Self::PreCompact, + "afterShellExecution" => { + Self::Specialized(Specialized::Shell(decode("cursor", what, payload)?)) + } + "afterMCPExecution" => { + Self::Specialized(Specialized::Mcp(decode("cursor", what, payload)?)) + } + "afterFileEdit" => { + Self::Specialized(Specialized::FileEdit(decode("cursor", what, payload)?)) + } + "beforeReadFile" => { + Self::Specialized(Specialized::ReadFile(decode("cursor", what, payload)?)) + } + IMPORT_START => Self::ImportStart(decode("cursor", what, payload)?), + IMPORT_CHECKPOINT => Self::ImportCheckpoint, + IMPORT_STOP => Self::ImportStop(decode("cursor", what, payload)?), + _ => Self::Other, + }) + } +} + +pub(crate) const IMPORT_START: &str = "_bt_importStart"; +pub(crate) const IMPORT_CHECKPOINT: &str = "_bt_importCheckpoint"; +pub(crate) const IMPORT_STOP: &str = "_bt_importStop"; + +#[derive(Deserialize)] +pub(super) struct PromptHook { + pub prompt: String, +} + +pub(super) enum ToolPhase { + Pre, + Post(ToolResult), + Failure(ToolFailure), +} + +/// Fields every tool hook carries. +#[derive(Deserialize)] +pub(super) struct ToolHook { + pub tool_use_id: String, + pub tool_name: String, + pub tool_input: Map, + /// Sent by pre- and post-use hooks when the tool has a working directory. + pub cwd: Option, + /// Set when the tool runs inside another tool call, such as a subagent. + pub parent_tool_call_id: Option, +} + +#[derive(Deserialize)] +struct PostToolHook { + #[serde(flatten)] + tool: ToolHook, + #[serde(flatten)] + result: ToolResult, +} + +#[derive(Deserialize)] +pub(super) struct ToolResult { + pub tool_output: String, + /// Native execution time in milliseconds. + pub duration: f64, +} + +#[derive(Deserialize)] +struct ToolFailureHook { + #[serde(flatten)] + tool: ToolHook, + #[serde(flatten)] + failure: ToolFailure, +} + +#[derive(Deserialize)] +pub(super) struct ToolFailure { + pub error_message: String, + pub failure_type: String, + pub is_interrupt: bool, + /// Native execution time in milliseconds. + pub duration: f64, +} + +impl ToolPhase { + pub fn duration(&self) -> Option { + match self { + Self::Pre => None, + Self::Post(result) => Some(result.duration), + Self::Failure(failure) => Some(failure.duration), + } + } +} + +#[derive(Deserialize)] +pub(super) struct ThoughtHook { + pub text: String, +} + +#[derive(Deserialize)] +pub(super) struct ResponseHook { + pub text: String, + #[serde(flatten)] + pub usage: TurnUsage, +} + +#[derive(Deserialize)] +pub(super) struct StopHook { + pub status: String, + #[serde(flatten)] + pub usage: TurnUsage, +} + +/// Token totals for one user generation, repeated on response and stop. The +/// CLI omits counts it does not know. +#[derive(Deserialize)] +pub(super) struct TurnUsage { + pub input_tokens: Option, + pub output_tokens: Option, + pub cache_read_tokens: Option, + pub cache_write_tokens: Option, +} + +#[derive(Deserialize)] +pub(super) struct SessionEndHook { + pub reason: String, + pub final_status: String, + /// Documented as optional; the CLI does not send it. + pub error_message: Option, +} + +#[derive(Deserialize)] +pub(super) struct SubagentStartHook { + pub subagent_id: String, + pub subagent_type: String, + pub task: String, + /// The tool call that spawned the subagent. + pub tool_call_id: Option, + pub subagent_model: Option, +} + +#[derive(Deserialize)] +pub(super) struct SubagentStopHook { + pub subagent_id: String, + pub subagent_type: String, + pub status: String, + pub summary: Option, +} + +/// Hooks that carry a richer result for a tool already reported by the +/// generic tool hooks. None of them names the tool call; the CLI does not +/// send `tool_use_id` here, so it is matched only when present. +pub(super) enum Specialized { + Shell(ShellHook), + Mcp(McpHook), + FileEdit(FileEditHook), + ReadFile(ReadFileHook), +} + +#[derive(Deserialize)] +pub(super) struct ShellHook { + pub tool_use_id: Option, + pub command: String, + pub output: String, +} + +#[derive(Deserialize)] +pub(super) struct McpHook { + pub tool_use_id: Option, + pub tool_name: String, + /// The tool arguments, JSON-encoded. + pub tool_input: String, + /// The tool result, JSON-encoded. + pub result_json: String, +} + +#[derive(Deserialize)] +pub(super) struct FileEditHook { + pub tool_use_id: Option, + pub file_path: String, + pub edits: Vec, +} + +#[derive(Deserialize, Serialize)] +pub(super) struct FileEdit { + pub old_string: String, + pub new_string: String, +} + +#[derive(Deserialize)] +pub(super) struct ReadFileHook { + pub tool_use_id: Option, + pub file_path: String, + /// File contents observed before the read. + pub content: String, +} + +impl Specialized { + pub fn tool_use_id(&self) -> Option<&str> { + match self { + Self::Shell(hook) => hook.tool_use_id.as_deref(), + Self::Mcp(hook) => hook.tool_use_id.as_deref(), + Self::FileEdit(hook) => hook.tool_use_id.as_deref(), + Self::ReadFile(hook) => hook.tool_use_id.as_deref(), + } + } + + /// Whether this hook can enrich a generic tool span with this name. + pub fn accepts(&self, tool_name: &str) -> bool { + match self { + Self::Shell(_) => tool_name == "Shell", + Self::Mcp(hook) => { + tool_name.strip_prefix("MCP:").unwrap_or(tool_name) == hook.tool_name + } + Self::ReadFile(_) => tool_name == "Read", + Self::FileEdit(_) => tool_name == "Write", + } + } + + /// Whether this hook describes the call with these generic tool inputs. + pub fn matches_input(&self, input: &Value) -> bool { + let file_path = || { + input + .get("file_path") + .or_else(|| input.get("path")) + .and_then(Value::as_str) + }; + match self { + Self::Shell(hook) => { + input.get("command").and_then(Value::as_str) == Some(hook.command.as_str()) + } + Self::Mcp(hook) => { + let arguments = serde_json::from_str::(&hook.tool_input) + .unwrap_or_else(|_| Value::String(hook.tool_input.clone())); + input == &arguments + } + Self::FileEdit(hook) => file_path() == Some(hook.file_path.as_str()), + Self::ReadFile(hook) => file_path() == Some(hook.file_path.as_str()), + } + } + + /// The native result to record on the tool span. Reads report contents + /// observed before the tool ran instead. + pub fn result(&self) -> Option { + match self { + Self::Shell(hook) => Some(Value::String(hook.output.clone())), + Self::Mcp(hook) => Some(Value::String(hook.result_json.clone())), + Self::FileEdit(hook) => { + Some(serde_json::to_value(&hook.edits).expect("file edits serialize")) + } + Self::ReadFile(_) => None, + } + } +} + +/// `sessionStart` synthesized by transcript import. +#[derive(Deserialize)] +pub(super) struct ImportBoundary { + /// Always true: a saved transcript has no session start time. + pub start_time_estimated: bool, +} + +/// `stop` synthesized by transcript import from the transcript's final +/// `turn_ended` marker, when it has one. +#[derive(Deserialize)] +pub(super) struct ImportStopHook { + pub status: Option, + pub error_message: Option, +} + +/// One line of Cursor's saved transcript. +#[derive(Deserialize)] +#[serde(untagged)] +pub(super) enum TranscriptRecord { + Message { + role: Role, + message: TranscriptMessage, + }, + /// Marks the end of a turn. The translator reads turn status from hooks, + /// so only the marker's shape is checked. + TurnEnded { + #[serde(rename = "type")] + _kind: TurnEndedKind, + #[serde(rename = "status")] + _status: String, + #[serde(rename = "error", default)] + _error: Option, + }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "lowercase")] +pub(super) enum Role { + User, + Assistant, +} + +impl Role { + pub fn as_str(self) -> &'static str { + match self { + Self::User => "user", + Self::Assistant => "assistant", + } + } +} + +#[derive(Deserialize)] +pub(super) struct TranscriptMessage { + pub content: MessageContent, +} + +#[derive(Deserialize)] +#[serde(rename_all = "snake_case")] +pub(super) enum TurnEndedKind { + TurnEnded, +} + +/// Message content: plain text, or content blocks. +#[derive(Deserialize)] +#[serde(untagged)] +pub(super) enum MessageContent { + Text(String), + Blocks(Vec), +} + +#[derive(Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub(super) enum ContentBlock { + Text { + text: String, + }, + /// Tool requests are reported by hooks; transcripts add nothing to them. + ToolUse, +} + +impl MessageContent { + /// The concatenated text of the content. + pub fn text(&self) -> String { + match self { + Self::Text(text) => text.clone(), + Self::Blocks(blocks) => blocks + .iter() + .filter_map(|block| match block { + ContentBlock::Text { text } => Some(text.as_str()), + ContentBlock::ToolUse => None, + }) + .collect(), + } + } +} + +impl TranscriptRecord { + pub fn decode(record: &Value) -> Result { + decode("cursor", "transcript record", record) + } + + pub fn role(&self) -> Option { + match self { + Self::Message { role, .. } => Some(*role), + Self::TurnEnded { .. } => None, + } + } + + pub fn text(&self) -> Option { + match self { + Self::Message { message, .. } => Some(message.content.text()), + Self::TurnEnded { .. } => None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn specialized_hooks_require_their_kind_s_fields() { + let common = json!({"generation_id":"g","model":"m","workspace_roots":[]}); + let mut shell = common.clone(); + shell["command"] = json!("ls"); + let error = Hook::decode("afterShellExecution", &shell) + .err() + .unwrap() + .to_string(); + assert!(error.contains("missing field `output`"), "{error}"); + shell["output"] = json!("a.txt"); + assert!(Hook::decode("afterShellExecution", &shell).is_ok()); + } +} diff --git a/bt-daemon/src/translate/cursor.rs b/bt-daemon/src/translate/cursor/mod.rs similarity index 70% rename from bt-daemon/src/translate/cursor.rs rename to bt-daemon/src/translate/cursor/mod.rs index a2047406..7b46e387 100644 --- a/bt-daemon/src/translate/cursor.rs +++ b/bt-daemon/src/translate/cursor/mod.rs @@ -3,6 +3,8 @@ //! Transcripts add observable messages only. Model inputs/times are reconstructed, //! and neither missing tool contents nor unverified token counts are invented. +mod hooks; + use super::git::GitMetadataCache; use super::recent::{RecentMap, RecentSet}; use super::{ @@ -10,6 +12,12 @@ use super::{ TranslatorFactory, }; use crate::{ids, wire::Envelope}; +use hooks::{ + CursorHook, Hook, ImportStopHook, PromptHook, ResponseHook, Role, SessionEndHook, Specialized, + StopHook, SubagentStartHook, SubagentStopHook, ToolHook, ToolPhase, TranscriptMirror, + TranscriptRecord, TurnUsage, +}; +pub(crate) use hooks::{IMPORT_CHECKPOINT, IMPORT_START, IMPORT_STOP}; use serde_json::{json, Value}; use std::collections::{BTreeMap, VecDeque}; use std::io::{Read, Seek, SeekFrom}; @@ -214,7 +222,7 @@ struct CursorTranslator { started: bool, last_ms: i64, cwd: Option, - model: Option, + model: Option, model_params: Option, git: Arc, turn_seq: u64, @@ -276,28 +284,22 @@ impl CursorTranslator { ..Default::default() } } - fn root(&mut self, e: &Envelope, ctx: &SessionCtx, ops: &mut Vec) { + fn root(&mut self, e: &Envelope, hook: &Hook, ctx: &SessionCtx, ops: &mut Vec) { self.last_ms = self.last_ms.max(e.ts_ms); let previous_cwd = self.cwd.clone(); - if let Some(cwd) = e - .payload - .get("cwd") - .and_then(Value::as_str) - .filter(|cwd| !cwd.is_empty()) - .or_else(|| e.payload.get("workspace_roots")?.get(0)?.as_str()) - { + if let Some(cwd) = hook.cwd() { self.cwd = Some(cwd.into()); } - if let Some(model) = e.payload.get("model_id").or_else(|| e.payload.get("model")) { - if self.model.as_ref() != Some(model) { + if let Some(model) = hook.model() { + if self.model.as_deref() != Some(model) { self.model_params = None; } - self.model = Some(model.clone()); + self.model = Some(model.to_owned()); } - if e.event == "beforeSubmitPrompt" { + if matches!(hook.kind, CursorHook::BeforeSubmitPrompt(_)) { self.model_params = None; } - if let Some(params) = e.payload.get("model_params") { + if let Some(params) = hook.model_params() { self.model_params = Some(params.clone()); } if self.started { @@ -325,8 +327,8 @@ impl CursorTranslator { let mut metadata = json!({"source":"cursor", "session_id":ctx.session_id, "username":local_username(), "os":std::env::consts::OS, "workspace":self.cwd, "trace_cursor_version":e.source_version, "trace_plugin_version":e.plugin_version}); - if let Some(estimated) = e.payload.get("start_time_estimated") { - metadata["start_time_estimated"] = estimated.clone(); + if let CursorHook::ImportStart(start) = &hook.kind { + metadata["start_time_estimated"] = json!(start.start_time_estimated); } ops.push(SpanOp::Insert(SpanRow { span_id: self.root.clone(), @@ -388,11 +390,9 @@ impl CursorTranslator { self.history.push(json!({"role":"user","content":p})); } } - fn prompt(&mut self, e: &Envelope, ops: &mut Vec) { - let Some(prompt) = e.payload.get("prompt").and_then(Value::as_str) else { - return; - }; - let generation = e.payload.get("generation_id").and_then(Value::as_str); + fn prompt(&mut self, e: &Envelope, hook: &Hook, prompt: &PromptHook, ops: &mut Vec) { + let prompt = prompt.prompt.as_str(); + let generation = hook.generation_id(); if generation.is_some_and(|g| self.prompt_generations.get(g).is_some()) { return; } @@ -595,12 +595,17 @@ impl CursorTranslator { } } } - fn tool(&mut self, e: &Envelope, ops: &mut Vec) { - let Some(call_id) = e.payload.get("tool_use_id").and_then(Value::as_str) else { - return; - }; + fn tool( + &mut self, + e: &Envelope, + hook: &Hook, + phase: &ToolPhase, + tool_hook: &ToolHook, + ops: &mut Vec, + ) { + let call_id = tool_hook.tool_use_id.as_str(); let key = call_id.to_string(); - let terminal = e.event != "preToolUse"; + let terminal = !matches!(phase, ToolPhase::Pre); let late_incomplete = self .completed .get(&key) @@ -622,42 +627,32 @@ impl CursorTranslator { } if !self.tools.contains_key(&key) { - let native_turn = e - .payload - .get("generation_id") - .and_then(Value::as_str) + let native_turn = hook + .generation_id() .and_then(|g| self.prompt_generations.get(g)) .map(|t| t.id.clone()); if native_turn.is_none() { self.ensure_turn(e.ts_ms, None, None, "hook_order", ops); } let turn = native_turn.unwrap_or_else(|| self.turn.as_ref().unwrap().id.clone()); - let name = e - .payload - .get("tool_name") - .and_then(Value::as_str) - .unwrap_or("Cursor tool"); + let name = tool_hook.tool_name.as_str(); let mut row = self.row( ids::span_id(&self.namespace, &format!("tool:{call_id}")), turn.clone(), name, SpanType::Tool, ); - let duration = e - .payload - .get("duration") - .and_then(Value::as_f64) - .filter(|n| n.is_finite() && *n >= 0.0); + let duration = phase.duration().filter(|n| n.is_finite() && *n >= 0.0); row.start_ms = Some(if terminal { e.ts_ms .saturating_sub(duration.unwrap_or(0.0).min(i64::MAX as f64) as i64) } else { e.ts_ms }); - row.input = e.payload.get("tool_input").cloned(); + row.input = Some(Value::Object(tool_hook.tool_input.clone())); row.metadata = Some( - json!({"tool_use_id":call_id,"generation_id":e.payload.get("generation_id"),"turn_attribution":"active_prompt_boundary", - "start_time_estimated":terminal,"result_completeness":"unknown","parent_tool_call_id":e.payload.get("parent_tool_call_id")}), + json!({"tool_use_id":call_id,"generation_id":hook.generation_id(),"turn_attribution":"active_prompt_boundary", + "start_time_estimated":terminal,"result_completeness":"unknown","parent_tool_call_id":tool_hook.parent_tool_call_id}), ); ops.push(SpanOp::Insert(row.clone())); self.insert_open_tool( @@ -731,6 +726,10 @@ impl CursorTranslator { let mut tool = self.remove_open_tool(&key).unwrap(); tool.ended = true; tool.row.end_ms = Some(e.ts_ms.max(tool.row.start_ms.unwrap_or(e.ts_ms))); + let output = match phase { + ToolPhase::Post(result) => Some(json!(result.tool_output)), + _ => None, + }; if tool .row .metadata @@ -738,40 +737,24 @@ impl CursorTranslator { .is_some_and(|m| m.get("result_source").is_some()) { tool.row.metadata.as_mut().unwrap()["native_tool_result"] = - e.payload.get("tool_output").cloned().unwrap_or(Value::Null); + output.unwrap_or(Value::Null); } else { - tool.row.output = e.payload.get("tool_output").cloned(); + tool.row.output = output; } - let failed = e.event == "postToolUseFailure"; let mut metadata = tool.row.metadata.take().unwrap_or(json!({})); - metadata["status"] = json!(if failed { "error" } else { "completed" }); - if let Some(duration) = e - .payload - .get("duration") - .and_then(Value::as_f64) - .filter(|n| n.is_finite() && *n >= 0.0) - { + metadata["status"] = json!(if matches!(phase, ToolPhase::Failure(_)) { + "error" + } else { + "completed" + }); + if let Some(duration) = phase.duration().filter(|n| n.is_finite() && *n >= 0.0) { metadata["native_duration_ms"] = json!(duration); } - if failed { - tool.row.error = Some( - e.payload - .get("error_message") - .and_then(Value::as_str) - .unwrap_or("Cursor tool failed") - .into(), - ); - metadata["failure_type"] = e - .payload - .get("failure_type") - .cloned() - .unwrap_or(Value::Null); - metadata["is_interrupt"] = e - .payload - .get("is_interrupt") - .cloned() - .unwrap_or(Value::Null); - if e.payload.get("failure_type").and_then(Value::as_str) == Some("permission_denied") { + if let ToolPhase::Failure(failure) = phase { + tool.row.error = Some(failure.error_message.clone()); + metadata["failure_type"] = json!(failure.failure_type); + metadata["is_interrupt"] = json!(failure.is_interrupt); + if failure.failure_type == "permission_denied" { metadata["tool_approval"] = json!("denied"); } } @@ -823,33 +806,14 @@ impl CursorTranslator { self.open_tool_order.retain(|candidate| candidate != key); self.tools.remove(key) } - fn specialized(&mut self, e: &Envelope, ops: &mut Vec) { - if !matches!( - e.event.as_str(), - "afterShellExecution" | "afterMCPExecution" | "afterFileEdit" | "beforeReadFile" - ) { - return; - } - let explicit_id = e.payload.get("tool_use_id").and_then(Value::as_str); + fn specialized(&mut self, e: &Envelope, hook: &Specialized, ops: &mut Vec) { + let explicit_id = hook.tool_use_id(); let candidates: Vec = self .tools .iter() .chain(self.completed.iter()) .filter_map(|(id, tool)| { - let compatible = match e.event.as_str() { - "afterShellExecution" => tool.row.name == "Shell", - "afterMCPExecution" => { - tool.row.name.strip_prefix("MCP:").unwrap_or(&tool.row.name) - == e.payload - .get("tool_name") - .and_then(Value::as_str) - .unwrap_or("") - } - "beforeReadFile" => tool.row.name == "Read", - "afterFileEdit" => tool.row.name == "Write", - _ => false, - }; - if !compatible { + if !hook.accepts(&tool.row.name) { return None; } if let Some(native) = explicit_id { @@ -859,30 +823,7 @@ impl CursorTranslator { return None; } let input = tool.row.input.as_ref()?; - let matches = match e.event.as_str() { - "afterShellExecution" => { - input.get("command") == e.payload.get("command") - && e.payload.get("command").is_some() - } - "afterMCPExecution" => { - tool.row.name.strip_prefix("MCP:").unwrap_or(&tool.row.name) - == e.payload - .get("tool_name") - .and_then(Value::as_str) - .unwrap_or("") - && input - == &parse_json_string( - e.payload.get("tool_input").cloned().unwrap_or(Value::Null), - ) - } - "beforeReadFile" | "afterFileEdit" => { - input.get("file_path").or_else(|| input.get("path")) - == e.payload.get("file_path") - && e.payload.get("file_path").is_some() - } - _ => false, - }; - matches.then(|| id.clone()) + hook.matches_input(input).then(|| id.clone()) }) .collect(); if candidates.len() != 1 { @@ -895,26 +836,20 @@ impl CursorTranslator { .cloned() .or_else(|| self.completed.get(id).cloned()) .unwrap(); - if e.event == "beforeReadFile" { - if let Some(content) = e.payload.get("content") { - copy.row.metadata.as_mut().unwrap()["pre_read_content"] = content.clone(); - copy.row.metadata.as_mut().unwrap()["pre_read_content_source"] = - json!("beforeReadFile"); - ops.push(SpanOp::Merge(copy.row.clone())); - if copy.ended { - self.completed.insert(id.clone(), copy); - } else { - self.tools.insert(id.clone(), copy); - } + if let Specialized::ReadFile(read) = hook { + copy.row.metadata.as_mut().unwrap()["pre_read_content"] = json!(read.content); + copy.row.metadata.as_mut().unwrap()["pre_read_content_source"] = + json!("beforeReadFile"); + ops.push(SpanOp::Merge(copy.row.clone())); + if copy.ended { + self.completed.insert(id.clone(), copy); + } else { + self.tools.insert(id.clone(), copy); } return; } - let output = e - .payload - .get("output") - .or_else(|| e.payload.get("result_json")) - .or_else(|| e.payload.get("edits")); - if let Some(output) = output { + if let Some(output) = hook.result() { + let output = &output; copy.row.output = Some(output.clone()); copy.row.metadata.as_mut().unwrap()["result_source"] = json!(e.event); copy.row.metadata.as_mut().unwrap()["result_completeness"] = @@ -949,16 +884,20 @@ impl CursorTranslator { } } } - fn transcript_batch(&mut self, e: &Envelope, ops: &mut Vec) -> anyhow::Result { - let Some(reference) = e.payload.get("_bt_transcript_mirror") else { - return Ok(true); - }; - let (Some(path), Some(through)) = ( - reference.get("mirror").and_then(Value::as_str), - reference.get("through").and_then(Value::as_u64), - ) else { + fn transcript_batch( + &mut self, + e: &Envelope, + hook: &Hook, + ops: &mut Vec, + ) -> anyhow::Result { + let Some(TranscriptMirror::Captured { + mirror: path, + through, + }) = &hook.transcript_mirror + else { return Ok(true); }; + let (path, through) = (path.as_str(), *through); if self.transcript.path != path { self.transcript = TranscriptCursor { path: path.into(), @@ -973,42 +912,48 @@ impl CursorTranslator { } let (records, truncated, complete) = self.transcript.read_batch(path, through)?; self.history.truncated |= truncated; - for record in records { - let role = record.get("role").and_then(Value::as_str); - if role == Some("user") { + for raw in records { + let record = TranscriptRecord::decode(&raw)?; + let role = record.role(); + if role == Some(Role::User) { self.transcript.assistant_records = 0; // User ownership must be resolved again after a rewrite: // a truncated view may put a later native turn at line 1. - self.transcript_record(&record, e.ts_ms, ops); + self.transcript_record(&raw, &record, e.ts_ms, ops); } else { - if role == Some("assistant") { + if role == Some(Role::Assistant) { self.transcript.assistant_records += 1; } let key = format!( "transcript:{:?}:{}:{}:{}", - role, + role.map(Role::as_str), self.transcript_owner.as_deref().unwrap_or("unattributed"), self.transcript.assistant_records, - ids::span_id(&self.namespace, &record.to_string()) + ids::span_id(&self.namespace, &raw.to_string()) ); if self.seen.insert(key) { - self.transcript_record(&record, e.ts_ms, ops); + self.transcript_record(&raw, &record, e.ts_ms, ops); } } } Ok(complete) } - fn transcript_record(&mut self, record: &Value, ts: i64, ops: &mut Vec) { - let role = record.get("role").and_then(Value::as_str).unwrap_or(""); - let content = record.get("message").and_then(|m| m.get("content")); - if role == "user" { - if content.is_none() { - self.history.truncated = true; + /// `raw` is the record as saved, which keys late merges for idempotence. + fn transcript_record( + &mut self, + raw: &Value, + record: &TranscriptRecord, + ts: i64, + ops: &mut Vec, + ) { + let role = record.role(); + let content = record.text(); + if role == Some(Role::User) { + let Some(full_prompt) = content else { return; - } + }; self.transcript_users += 1; self.transcript_unattributed = false; - let full_prompt = content.map(text_content).unwrap_or_default(); let prompt = transcript_prompt(&full_prompt).to_string(); if let Some((known_prompt, id)) = self .transcript @@ -1103,13 +1048,12 @@ impl CursorTranslator { .prompt_owners .get(&self.transcript_users) .map(|(_, id)| id.clone()); - } else if role == "assistant" { + } else if role == Some(Role::Assistant) { if self.transcript_unattributed { self.history.truncated = true; return; } - if let Some(content) = content { - let text = text_content(content); + if let Some(text) = content { if text.is_empty() { return; } @@ -1141,7 +1085,7 @@ impl CursorTranslator { ); row.late_merge_key = Some(format!( "cursor-transcript:{}", - ids::span_id(&self.namespace, &record.to_string()) + ids::span_id(&self.namespace, &raw.to_string()) )); ops.push(SpanOp::Merge(row)); self.mark_history_truncated(); @@ -1181,88 +1125,72 @@ impl CursorTranslator { )); ops.push(SpanOp::Merge(row)); } - fn subagent(&mut self, e: &Envelope, ops: &mut Vec) { - let Some(id) = e.payload.get("subagent_id").and_then(Value::as_str) else { + fn subagent_start(&mut self, e: &Envelope, hook: &SubagentStartHook, ops: &mut Vec) { + let id = hook.subagent_id.as_str(); + if self.subagents.contains_key(id) { return; - }; - if e.event == "subagentStart" { - if self.subagents.contains_key(id) { - return; - } - self.ensure_turn(e.ts_ms, None, None, "hook_order", ops); - let spawn = e.payload.get("tool_call_id").and_then(Value::as_str); - let parent = spawn - .and_then(|s| self.tools.get(s).or_else(|| self.completed.get(s))) - .map(|t| t.turn.clone()) - .unwrap_or_else(|| self.turn.as_ref().unwrap().id.clone()); - let mut row = self.row( - ids::span_id(&self.namespace, &format!("subagent:{id}")), - parent, - "Cursor subagent", - SpanType::Task, - ); - row.start_ms = Some(e.ts_ms); - row.input = e.payload.get("task").cloned(); - row.metadata = Some( - json!({"subagent_id":id,"spawning_tool_call_id":spawn,"model":e.payload.get("subagent_model"),"subagent_type":e.payload.get("subagent_type"),"recursive_activity_verified":false}), - ); - if self.subagents.len() >= MAX_OPEN_TOOLS { - return; - } - self.subagents.insert(id.into(), row.clone()); - ops.push(SpanOp::Insert(row)); - } else if let Some(mut row) = self.subagents.remove(id) { + } + self.ensure_turn(e.ts_ms, None, None, "hook_order", ops); + let spawn = hook.tool_call_id.as_deref(); + let parent = spawn + .and_then(|s| self.tools.get(s).or_else(|| self.completed.get(s))) + .map(|t| t.turn.clone()) + .unwrap_or_else(|| self.turn.as_ref().unwrap().id.clone()); + let mut row = self.row( + ids::span_id(&self.namespace, &format!("subagent:{id}")), + parent, + "Cursor subagent", + SpanType::Task, + ); + row.start_ms = Some(e.ts_ms); + row.input = Some(json!(hook.task)); + row.metadata = Some( + json!({"subagent_id":id,"spawning_tool_call_id":spawn,"model":hook.subagent_model,"subagent_type":hook.subagent_type,"recursive_activity_verified":false}), + ); + if self.subagents.len() >= MAX_OPEN_TOOLS { + return; + } + self.subagents.insert(id.into(), row.clone()); + ops.push(SpanOp::Insert(row)); + } + fn subagent_stop(&mut self, e: &Envelope, hook: &SubagentStopHook, ops: &mut Vec) { + let id = hook.subagent_id.as_str(); + let error = + matches!(hook.status.as_str(), "error" | "aborted").then(|| hook.status.clone()); + if let Some(mut row) = self.subagents.remove(id) { row.end_ms = Some(e.ts_ms.max(row.start_ms.unwrap_or(e.ts_ms))); - row.output = e.payload.get("summary").cloned(); - row.metadata.as_mut().unwrap()["status"] = - e.payload.get("status").cloned().unwrap_or(Value::Null); - if matches!( - e.payload.get("status").and_then(Value::as_str), - Some("error" | "aborted") - ) { - row.error = Some(e.payload["status"].as_str().unwrap().into()); - } + row.output = hook.summary.as_ref().map(|summary| json!(summary)); + row.metadata.as_mut().unwrap()["status"] = json!(hook.status); + row.error = error; ops.push(SpanOp::Merge(row)); - } else { - // Cursor only supports a stop hook in the shipped integration. - // Keep the terminal observation as a useful, zero-duration span - // and mark the unavailable start time explicitly. - self.ensure_turn(e.ts_ms, None, None, "hook_order", ops); - let spawn = e.payload.get("tool_call_id").and_then(Value::as_str); - let parent = spawn - .and_then(|s| self.tools.get(s).or_else(|| self.completed.get(s))) - .map(|t| t.turn.clone()) - .unwrap_or_else(|| self.turn.as_ref().unwrap().id.clone()); - let mut row = self.row( - ids::span_id(&self.namespace, &format!("subagent:{id}")), - parent, - "Cursor subagent", - SpanType::Task, - ); - row.start_ms = Some(e.ts_ms); - row.end_ms = Some(e.ts_ms); - row.output = e.payload.get("summary").cloned(); - row.metadata = Some(json!({ - "subagent_id": id, - "spawning_tool_call_id": spawn, - "model": e.payload.get("subagent_model"), - "subagent_type": e.payload.get("subagent_type"), - "recursive_activity_verified": false, - "start_time_estimated": true, - "result_completeness": "terminal_observation_only", - "status": e.payload.get("status"), - })); - if matches!( - e.payload.get("status").and_then(Value::as_str), - Some("error" | "aborted") - ) { - row.error = e.payload["status"].as_str().map(str::to_owned); - } - ops.push(SpanOp::Insert(row)); + return; } + // The shipped plugin registers only the stop hook. Keep the terminal + // observation as a useful, zero-duration span and mark the + // unavailable start time explicitly. + self.ensure_turn(e.ts_ms, None, None, "hook_order", ops); + let parent = self.turn.as_ref().unwrap().id.clone(); + let mut row = self.row( + ids::span_id(&self.namespace, &format!("subagent:{id}")), + parent, + "Cursor subagent", + SpanType::Task, + ); + row.start_ms = Some(e.ts_ms); + row.end_ms = Some(e.ts_ms); + row.output = hook.summary.as_ref().map(|summary| json!(summary)); + row.metadata = Some(json!({ + "subagent_id": id, + "subagent_type": hook.subagent_type, + "recursive_activity_verified": false, + "start_time_estimated": true, + "result_completeness": "terminal_observation_only", + "status": hook.status, + })); + row.error = error; + ops.push(SpanOp::Insert(row)); } - fn turn_usage(&mut self, e: &Envelope, ops: &mut Vec) { - let generation = e.payload.get("generation_id").and_then(Value::as_str); + fn turn_usage(&mut self, generation: Option<&str>, usage: &TurnUsage, ops: &mut Vec) { // Verified interactive CLI/desktop emissions are totals for one user // generation, repeated on response and stop. Merge, never sum. let Some((generation, mut turn)) = @@ -1271,13 +1199,13 @@ impl CursorTranslator { return; }; let mut metrics = serde_json::Map::new(); - for (native, metric) in [ - ("input_tokens", "prompt_tokens"), - ("output_tokens", "completion_tokens"), - ("cache_read_tokens", "prompt_cached_tokens"), - ("cache_write_tokens", "prompt_cache_creation_tokens"), + for (count, metric) in [ + (usage.input_tokens, "prompt_tokens"), + (usage.output_tokens, "completion_tokens"), + (usage.cache_read_tokens, "prompt_cached_tokens"), + (usage.cache_write_tokens, "prompt_cache_creation_tokens"), ] { - if let Some(count) = e.payload.get(native).and_then(Value::as_u64) { + if let Some(count) = count { metrics.insert(metric.into(), json!(count)); } } @@ -1314,29 +1242,31 @@ impl CursorTranslator { )); ops.push(SpanOp::Merge(row)); } - fn reduce_hook(&mut self, e: &Envelope, ops: &mut Vec) { + fn reduce_hook(&mut self, e: &Envelope, hook: &Hook, ops: &mut Vec) { + // Deduplicate on the native payload as received, without daemon fields. let mut native = e.payload.clone(); if let Some(map) = native.as_object_mut() { map.retain(|k, _| !k.starts_with("_bt_")); } - let generation = native.get("generation_id").and_then(Value::as_str); - let scope = if native.get("tool_use_id").is_some() { + let generation = hook.generation_id(); + let scope = if hook.tool_use_id().is_some() { "native_tool".to_string() } else if matches!( - e.event.as_str(), - "beforeSubmitPrompt" | "afterAgentResponse" | "stop" + hook.kind, + CursorHook::BeforeSubmitPrompt(_) + | CursorHook::AfterAgentResponse(_) + | CursorHook::Stop(_) + | CursorHook::ImportStop(_) ) { generation .map(str::to_string) .unwrap_or_else(|| format!("capture:{}", e.ts_ms)) } else if matches!( - e.event.as_str(), - "sessionStart" - | "sessionEnd" - | "beforeReadFile" - | "afterShellExecution" - | "afterMCPExecution" - | "afterFileEdit" + hook.kind, + CursorHook::SessionStart + | CursorHook::ImportStart(_) + | CursorHook::SessionEnd(_) + | CursorHook::Specialized(_) ) { format!("capture:{}", e.ts_ms) } else { @@ -1350,159 +1280,169 @@ impl CursorTranslator { if !self.seen.insert(fingerprint) { return; } - match e.event.as_str() { - "beforeSubmitPrompt" => self.prompt(e, ops), - "preToolUse" | "postToolUse" | "postToolUseFailure" => self.tool(e, ops), - "afterAgentThought" => { - if let Some(text) = e.payload.get("text") { - self.model_content(json!({"type":"thinking","text":text}), e.ts_ms, ops); - } - } - "afterAgentResponse" => { - let owner = e - .payload - .get("generation_id") - .and_then(Value::as_str) - .and_then(|g| self.prompt_generations.get(g)) - .map(|t| t.id.clone()); - if let Some(owner) = - owner.filter(|id| self.turn.as_ref().is_none_or(|t| &t.id != id)) - { - let mut row = self.row(owner, self.root.clone(), "", SpanType::Task); - row.output = e.payload.get("text").cloned(); - row.metadata = Some( - json!({"late_output_enrichment":true,"historical_model_grouping_available":false}), - ); - row.late_merge_key = Some(format!( - "cursor-response:{}", - ids::span_id(&self.namespace, &native.to_string()) - )); - ops.push(SpanOp::Merge(row)); - self.mark_history_truncated(); - self.turn_usage(e, ops); - } else { - if let Some(text) = e.payload.get("text").and_then(Value::as_str) { - if self - .turn - .as_ref() - .is_none_or(|t| t.output.as_deref() != Some(text)) - { - let unseen = self - .turn - .as_ref() - .and_then(|t| text.strip_prefix(&t.emitted_text)) - .unwrap_or(text); - if !unseen.is_empty() { - self.model_content( - json!({"type":"text","text":unseen}), - e.ts_ms, - ops, - ); - } - if let Some(t) = &mut self.turn { - t.output = Some(text.into()); - } - } - } - self.turn_usage(e, ops); - self.close_model(e.ts_ms, ops); - self.merge_late_answer(ops); - } + match &hook.kind { + CursorHook::BeforeSubmitPrompt(prompt) => self.prompt(e, hook, prompt, ops), + CursorHook::Tool(phase, tool) => self.tool(e, hook, phase, tool, ops), + CursorHook::AfterAgentThought(thought) => { + self.model_content(json!({"type":"thinking","text":thought.text}), e.ts_ms, ops); } - "stop" => { - self.turn_usage(e, ops); - let prior = e - .payload - .get("generation_id") - .and_then(Value::as_str) - .and_then(|g| self.prompt_generations.get(g)) - .filter(|t| self.turn.as_ref().is_none_or(|current| current.id != t.id)) - .cloned(); - if let Some(prior) = prior { - let mut row = self.row(prior.id, self.root.clone(), "", SpanType::Task); - row.end_ms = Some(e.ts_ms.max(prior.start)); - row.metadata = Some(json!({"status":e.payload.get("status")})); - if matches!( - e.payload.get("status").and_then(Value::as_str), - Some("error" | "aborted") - ) { - row.error = Some(e.payload["status"].as_str().unwrap().into()); - } - row.late_merge_key = Some(format!( - "cursor-stop:{}", - ids::span_id(&self.namespace, &native.to_string()) - )); - ops.push(SpanOp::Merge(row)); - } else { - self.close_turn( - e.ts_ms, - e.payload.get("status").and_then(Value::as_str), - e.payload.get("error_message").and_then(Value::as_str), - ops, - ); - } + CursorHook::AfterAgentResponse(response) => { + self.response(e, generation, response, &native, ops) } - "sessionEnd" => { - self.close_turn( - e.ts_ms, - e.payload.get("reason").and_then(Value::as_str), - e.payload.get("error_message").and_then(Value::as_str), - ops, - ); - let mut row = self.row( - self.root.clone(), - self.root.clone(), - "Cursor session", - SpanType::Task, - ); - row.parent_span_ids = self.root_parents.clone(); - row.end_ms = Some(e.ts_ms); - row.metadata = Some( - json!({"status":e.payload.get("reason"),"final_status":e.payload.get("final_status")}), - ); - row.output = self - .turn - .as_ref() - .and_then(|t| t.output.as_ref()) - .map(|s| json!(s)); - if e.payload.get("reason").and_then(Value::as_str) == Some("error") { - row.error = Some( - e.payload - .get("error_message") - .and_then(Value::as_str) - .unwrap_or("Cursor session failed") - .into(), - ); - } - ops.push(SpanOp::Merge(row)); + CursorHook::Stop(stop) => self.stop(e, generation, stop, &native, ops), + CursorHook::ImportStop(stop) => self.import_stop(e, stop, ops), + CursorHook::SessionEnd(end) => self.session_end(e, end, ops), + CursorHook::SubagentStart(start) => self.subagent_start(e, start, ops), + CursorHook::SubagentStop(stop) => self.subagent_stop(e, stop, ops), + CursorHook::PreCompact => self.compact(e, native, ops), + CursorHook::Specialized(specialized) => self.specialized(e, specialized, ops), + CursorHook::SessionStart + | CursorHook::ImportStart(_) + | CursorHook::ImportCheckpoint + | CursorHook::Other => {} + } + } + fn response( + &mut self, + e: &Envelope, + generation: Option<&str>, + response: &ResponseHook, + native: &Value, + ops: &mut Vec, + ) { + let owner = generation + .and_then(|g| self.prompt_generations.get(g)) + .map(|t| t.id.clone()); + if let Some(owner) = owner.filter(|id| self.turn.as_ref().is_none_or(|t| &t.id != id)) { + let mut row = self.row(owner, self.root.clone(), "", SpanType::Task); + row.output = Some(json!(response.text)); + row.metadata = Some( + json!({"late_output_enrichment":true,"historical_model_grouping_available":false}), + ); + row.late_merge_key = Some(format!( + "cursor-response:{}", + ids::span_id(&self.namespace, &native.to_string()) + )); + ops.push(SpanOp::Merge(row)); + self.mark_history_truncated(); + self.turn_usage(generation, &response.usage, ops); + return; + } + let text = response.text.as_str(); + if self + .turn + .as_ref() + .is_none_or(|t| t.output.as_deref() != Some(text)) + { + let unseen = self + .turn + .as_ref() + .and_then(|t| text.strip_prefix(&t.emitted_text)) + .unwrap_or(text); + if !unseen.is_empty() { + self.model_content(json!({"type":"text","text":unseen}), e.ts_ms, ops); } - "subagentStart" | "subagentStop" => self.subagent(e, ops), - "preCompact" => { - self.close_model(e.ts_ms, ops); - self.compact_seq += 1; - let mut row = self.row( - ids::span_id(&self.namespace, &format!("compaction:{}", self.compact_seq)), - self.root.clone(), - "Cursor compaction requested", - SpanType::Task, - ); - row.start_ms = Some(e.ts_ms); - row.input = Some(native); - row.end_ms = Some(e.ts_ms); - row.tags = Some(vec!["compaction".into()]); - row.metadata = Some( - json!({"observation_only":true,"completion_observed":false,"replacement_context_available":false}), - ); - ops.push(SpanOp::Insert(row)); - // No observable replacement context: do not carry stale precompact - // history forward as if it were the new model request. - self.history = History { - truncated: true, - ..Default::default() - }; + if let Some(t) = &mut self.turn { + t.output = Some(text.into()); } - _ => self.specialized(e, ops), } + self.turn_usage(generation, &response.usage, ops); + self.close_model(e.ts_ms, ops); + self.merge_late_answer(ops); + } + fn stop( + &mut self, + e: &Envelope, + generation: Option<&str>, + stop: &StopHook, + native: &Value, + ops: &mut Vec, + ) { + self.turn_usage(generation, &stop.usage, ops); + let prior = generation + .and_then(|g| self.prompt_generations.get(g)) + .filter(|t| self.turn.as_ref().is_none_or(|current| current.id != t.id)) + .cloned(); + let Some(prior) = prior else { + self.close_turn(e.ts_ms, Some(&stop.status), None, ops); + return; + }; + let mut row = self.row(prior.id, self.root.clone(), "", SpanType::Task); + row.end_ms = Some(e.ts_ms.max(prior.start)); + row.metadata = Some(json!({"status":stop.status})); + if matches!(stop.status.as_str(), "error" | "aborted") { + row.error = Some(stop.status.clone()); + } + row.late_merge_key = Some(format!( + "cursor-stop:{}", + ids::span_id(&self.namespace, &native.to_string()) + )); + ops.push(SpanOp::Merge(row)); + } + /// An imported transcript has no generation ids, so its end closes the + /// active turn with the transcript's final status. + fn import_stop(&mut self, e: &Envelope, stop: &ImportStopHook, ops: &mut Vec) { + self.close_turn( + e.ts_ms, + stop.status.as_deref(), + stop.error_message.as_deref(), + ops, + ); + } + fn session_end(&mut self, e: &Envelope, end: &SessionEndHook, ops: &mut Vec) { + self.close_turn( + e.ts_ms, + Some(&end.reason), + end.error_message.as_deref(), + ops, + ); + let mut row = self.row( + self.root.clone(), + self.root.clone(), + "Cursor session", + SpanType::Task, + ); + row.parent_span_ids = self.root_parents.clone(); + row.end_ms = Some(e.ts_ms); + row.metadata = Some(json!({"status":end.reason,"final_status":end.final_status})); + row.output = self + .turn + .as_ref() + .and_then(|t| t.output.as_ref()) + .map(|s| json!(s)); + if end.reason == "error" { + row.error = Some( + end.error_message + .as_deref() + .unwrap_or("Cursor session failed") + .into(), + ); + } + ops.push(SpanOp::Merge(row)); + } + fn compact(&mut self, e: &Envelope, native: Value, ops: &mut Vec) { + self.close_model(e.ts_ms, ops); + self.compact_seq += 1; + let mut row = self.row( + ids::span_id(&self.namespace, &format!("compaction:{}", self.compact_seq)), + self.root.clone(), + "Cursor compaction requested", + SpanType::Task, + ); + row.start_ms = Some(e.ts_ms); + row.input = Some(native); + row.end_ms = Some(e.ts_ms); + row.tags = Some(vec!["compaction".into()]); + row.metadata = Some( + json!({"observation_only":true,"completion_observed":false,"replacement_context_available":false}), + ); + ops.push(SpanOp::Insert(row)); + // No observable replacement context: do not carry stale precompact + // history forward as if it were the new model request. + self.history = History { + truncated: true, + ..Default::default() + }; } fn enrich(&self, ops: &mut [SpanOp], ctx: &SessionCtx) { self.git.enrich_rows(self.cwd.as_deref(), ops); @@ -1549,10 +1489,13 @@ impl AgentTranslator for CursorTranslator { self.pending.is_none(), "Cursor pending transcript work must be drained before the next hook" ); + // Decode before touching any state, so a payload in an unexpected + // format leaves the translator where it was. + let hook = Hook::decode(&e.event, &e.payload)?; let mut ops = Vec::new(); - self.root(e, ctx, &mut ops); - if self.transcript_batch(e, &mut ops)? { - self.reduce_hook(e, &mut ops); + self.root(e, &hook, ctx, &mut ops); + if self.transcript_batch(e, &hook, &mut ops)? { + self.reduce_hook(e, &hook, &mut ops); } else { self.pending = Some(e.clone()); } @@ -1563,9 +1506,11 @@ impl AgentTranslator for CursorTranslator { let Some(e) = self.pending.take() else { return Ok(None); }; + // `handle` already decoded this payload; decoding again is cheap. + let hook = Hook::decode(&e.event, &e.payload)?; let mut ops = Vec::new(); - if self.transcript_batch(&e, &mut ops)? { - self.reduce_hook(&e, &mut ops); + if self.transcript_batch(&e, &hook, &mut ops)? { + self.reduce_hook(&e, &hook, &mut ops); } else { self.pending = Some(e); } @@ -1645,24 +1590,6 @@ fn tool_result_content(value: Option<&Value>) -> String { } } -fn parse_json_string(value: Value) -> Value { - value - .as_str() - .and_then(|s| serde_json::from_str(s).ok()) - .unwrap_or(value) -} -fn text_content(content: &Value) -> String { - if let Some(text) = content.as_str() { - return text.into(); - } - content - .as_array() - .into_iter() - .flatten() - .filter_map(|b| b.get("text").and_then(Value::as_str)) - .collect::() -} - /// CLI saved messages wrap the submitted prompt in native timestamp/query tags. /// Use the exact observed wrapper only for correlation with the prompt hook. fn transcript_prompt(text: &str) -> &str { diff --git a/bt-daemon/src/translate/decode.rs b/bt-daemon/src/translate/decode.rs new file mode 100644 index 00000000..627120b9 --- /dev/null +++ b/bt-daemon/src/translate/decode.rs @@ -0,0 +1,49 @@ +//! Decoding native payloads into typed structs. +//! +//! A field with an unexpected JSON type is an error, never silently dropped. +//! It usually means the agent changed its format and the translator needs an +//! update, so [`DecodeError`] keeps these failures distinguishable from other +//! translation errors. + +use serde::de::DeserializeOwned; +use serde_json::Value; +use std::fmt; + +/// A native payload that does not have the shape its translator expects. +#[derive(Debug)] +pub(crate) struct DecodeError { + /// The agent source whose payload failed to decode. + pub source: &'static str, + /// What was being decoded, such as an event name. + pub what: String, + pub error: serde_json::Error, +} + +impl fmt::Display for DecodeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "unexpected {} {} format: {}", + self.source, self.what, self.error + ) + } +} + +impl std::error::Error for DecodeError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&self.error) + } +} + +/// Decode `value`, attributing a failure to `source` and `what`. +pub(crate) fn decode( + source: &'static str, + what: impl fmt::Display, + value: &Value, +) -> Result { + T::deserialize(value).map_err(|error| DecodeError { + source, + what: what.to_string(), + error, + }) +} diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index 7f8782d3..07d95bda 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -10,8 +10,9 @@ mod antigravity; mod claude; mod codex; -mod cursor; +pub(crate) mod cursor; mod debug; +mod decode; mod git; mod grok; mod opencode; diff --git a/bt-daemon/tests/cursor_pipeline.rs b/bt-daemon/tests/cursor_pipeline.rs index 7e93e2df..dacbf00e 100644 --- a/bt-daemon/tests/cursor_pipeline.rs +++ b/bt-daemon/tests/cursor_pipeline.rs @@ -287,6 +287,15 @@ async fn send_with_flush_mode( ) { payload["conversation_id"] = json!("cursor-recovery"); payload["hook_event_name"] = json!(event); + // Fields the Cursor CLI sends with every agent hook. + let object = payload.as_object_mut().unwrap(); + object + .entry("generation_id") + .or_insert_with(|| json!(format!("generation-{event}-{ts_ms}"))); + object.entry("model").or_insert_with(|| json!("default")); + object + .entry("workspace_roots") + .or_insert_with(|| json!(["/cursor-project"])); let env = Envelope { source: "cursor".into(), source_version: Some("2.5.0".into()), @@ -552,7 +561,7 @@ async fn cursor_mirror_generations_and_capture_bounds_survive_deleted_native_rec 2400, json!({ "generation_id":"g1", "tool_use_id":"shell-1", "tool_name":"Shell", - "tool_input":{"command":"printf cursor"}, "tool_output":{"stdout":"cursor"}, + "tool_input":{"command":"printf cursor"}, "tool_output":"cursor", "duration":200 }), ) @@ -695,7 +704,7 @@ async fn cursor_mirror_generations_and_capture_bounds_survive_deleted_native_rec &socket, "sessionEnd", 6000, - json!({"transcript_path":native,"reason":"completed"}), + json!({"transcript_path":native,"reason":"completed","final_status":"completed"}), ) .await; assert!( diff --git a/bt-daemon/tests/cursor_translator.rs b/bt-daemon/tests/cursor_translator.rs index 8482e95f..4d6c6c5e 100644 --- a/bt-daemon/tests/cursor_translator.rs +++ b/bt-daemon/tests/cursor_translator.rs @@ -11,9 +11,22 @@ use span_identity::IdentityLedger; use std::collections::HashMap; use std::path::{Path, PathBuf}; +/// A Cursor hook envelope. Fields the CLI sends with every agent hook are +/// filled in when the test does not set them; a test that needs a hook to +/// share a generation with an earlier one sets `generation_id` explicitly. fn event(kind: &str, ts: i64, mut payload: Value) -> Envelope { payload["hook_event_name"] = json!(kind); payload["conversation_id"] = json!("test-session"); + if !kind.starts_with("_bt_") { + let object = payload.as_object_mut().unwrap(); + object + .entry("generation_id") + .or_insert_with(|| json!(format!("generation-{kind}-{ts}"))); + object.entry("model").or_insert_with(|| json!("default")); + object + .entry("workspace_roots") + .or_insert_with(|| json!(["/workspace"])); + } Envelope { source: "cursor".into(), source_version: Some("2026.10.01-14929f9".into()), @@ -216,7 +229,7 @@ fn synthetic_success_is_parented_and_duplicate_response_stop_usage_is_not_summed h.handle(&event( "postToolUse", 130, - json!({"tool_name":"Read","tool_use_id":"r1","tool_output":"metadata only","duration":10}), + json!({"tool_name":"Read","tool_use_id":"r1","tool_input":{"file_path":"/tmp/a"},"tool_output":"metadata only","duration":10}), )); let response = event( "afterAgentResponse", @@ -304,7 +317,7 @@ fn terminal_tool_hook_does_not_close_a_later_completion() { h.handle(&event( "postToolUse", 150, - json!({"tool_name":"Shell","tool_use_id":"slow","tool_output":"done","duration":30}), + json!({"tool_name":"Shell","tool_use_id":"slow","tool_input":{"command":"sleep 1"},"tool_output":"done","duration":30}), )); assert_eq!(h.inserted(SpanType::Llm).len(), 1); h.handle(&event("stop", 160, json!({"status":"completed"}))); @@ -370,7 +383,7 @@ fn synthetic_denial_and_missing_tool_end_preserve_native_outcome() { 110, json!({"tool_name":"Shell","tool_use_id":"denied","tool_input":{"command":"rm x"}}), )); - h.handle(&event("postToolUseFailure",120,json!({"tool_name":"Shell","tool_use_id":"denied","failure_type":"permission_denied","error_message":"Denied by policy","is_interrupt":false}))); + h.handle(&event("postToolUseFailure",120,json!({"tool_name":"Shell","tool_use_id":"denied","tool_input":{"command":"rm x"},"failure_type":"permission_denied","error_message":"Denied by policy","is_interrupt":false,"duration":0}))); h.handle(&event( "preToolUse", 130, @@ -418,9 +431,13 @@ fn synthetic_attached_session_preserves_external_identity_across_all_merges() { h.handle(&event( "afterAgentResponse", 110, - json!({"generation_id":"t","text":"hi","cwd":"","workspace_roots":["/audit/second-workspace"]}), + json!({"generation_id":"t","text":"hi","model":"audit-model","workspace_roots":["/audit/second-workspace"]}), + )); + h.handle(&event( + "sessionEnd", + 120, + json!({"reason":"completed","final_status":"completed","model":"audit-model","workspace_roots":["/audit/second-workspace"]}), )); - h.handle(&event("sessionEnd", 120, json!({"reason":"completed"}))); h.finish(); let root = h .inserted(SpanType::Task) @@ -456,7 +473,7 @@ fn synthetic_attached_session_preserves_external_identity_across_all_merges() { fn imported_session_root_marks_its_start_timestamp_as_estimated() { let mut h = Harness::new("test-session"); h.handle(&event( - "sessionStart", + "_bt_importStart", 100, json!({"start_time_estimated":true}), )); @@ -473,17 +490,36 @@ fn imported_session_root_marks_its_start_timestamp_as_estimated() { #[test] fn imported_terminal_error_message_is_preserved_on_the_turn_span() { + let dir = tempfile::tempdir().unwrap(); + let transcript = dir.path().join("transcript.jsonl"); + std::fs::write( + &transcript, + format!( + "{}\n", + json!({"role":"user","message":{"content":[{"type":"text","text":"hello"}]}}) + ), + ) + .unwrap(); + let through = std::fs::metadata(&transcript).unwrap().len(); let mut h = Harness::new("test-session"); - h.handle(&event("sessionStart", 100, json!({}))); h.handle(&event( - "beforeSubmitPrompt", + "_bt_importStart", + 100, + json!({"start_time_estimated":true}), + )); + h.handle(&mirrored( + "_bt_importCheckpoint", 110, - json!({"generation_id":"turn-1","prompt":"hello"}), + &transcript, + through, + json!({}), )); - h.handle(&event( - "stop", + h.handle(&mirrored( + "_bt_importStop", 120, - json!({"generation_id":"turn-1","status":"error","error_message":"WritableIterable is closed"}), + &transcript, + through, + json!({"status":"error","error_message":"WritableIterable is closed"}), )); let turn = h.turns().into_iter().next().unwrap(); assert_eq!( @@ -566,7 +602,7 @@ fn synthetic_transcript_partial_records_and_large_batches_are_bounded_and_option h.handle(&event( "sessionEnd", 120, - json!({"reason":"completed","transcript_path":"/unavailable"}), + json!({"reason":"completed","final_status":"completed","transcript_path":"/unavailable"}), )); h.finish(); assert_eq!(h.turns().len(), 1); @@ -658,7 +694,7 @@ fn synthetic_delayed_transcript_after_closed_native_turn_does_not_create_phantom 130, &path, std::fs::metadata(&path).unwrap().len(), - json!({"reason":"completed"}), + json!({"reason":"completed","final_status":"completed"}), )); h.finish(); assert_eq!( @@ -703,7 +739,7 @@ fn synthetic_late_tool_completion_retains_old_turn_and_does_not_split_new_model( h.handle(&event( "postToolUse", 150, - json!({"tool_name":"Shell","tool_use_id":"late","tool_output":"old result"}), + json!({"tool_name":"Shell","tool_use_id":"late","tool_input":{"command":"slow"},"tool_output":"old result","duration":0}), )); h.handle(&event( "afterAgentResponse", @@ -923,7 +959,7 @@ fn synthetic_specialized_native_call_id_wins_over_ambiguous_arguments() { h.handle(&event( "postToolUse", 140, - json!({"tool_name":"Shell","tool_use_id":"shell-1","tool_output":"generic metadata"}), + json!({"tool_name":"Shell","tool_use_id":"shell-1","tool_input":{"command":"pwd"},"tool_output":"generic metadata","duration":0}), )); h.finish(); let target = h @@ -1026,14 +1062,13 @@ fn synthetic_subagents_parent_to_spawning_turn_and_unmatched_stop_stays_incomple h.handle(&event( "subagentStop", 140, - json!({"subagent_id":"child-a","status":"completed","summary":"available summary"}), + json!({"subagent_id":"child-a","subagent_type":"explore","status":"completed","summary":"available summary"}), )); h.handle(&event( - "subagentStop", - 150, - json!({"status":"completed","summary":"unmatched summary"}), + "sessionEnd", + 160, + json!({"reason":"completed","final_status":"completed"}), )); - h.handle(&event("sessionEnd", 160, json!({"reason":"completed"}))); h.finish(); let children: Vec<_> = h .inserted(SpanType::Task) @@ -1084,8 +1119,7 @@ fn terminal_only_subagent_stop_synthesizes_estimated_span() { "subagent_id":"child-terminal-only", "status":"completed", "summary":"finished work", - "subagent_type":"explore", - "subagent_model":"auto" + "subagent_type":"explore" }), )); h.finish(); @@ -1103,7 +1137,8 @@ fn terminal_only_subagent_stop_synthesizes_estimated_span() { assert_eq!(metadata["result_completeness"], "terminal_observation_only"); assert_eq!(metadata["status"], "completed"); assert_eq!(metadata["subagent_type"], "explore"); - assert_eq!(metadata["model"], "auto"); + // subagentStop does not report the subagent's model. + assert!(metadata.get("model").is_none()); assert_eq!(child.parent_span_ids, vec![h.turns()[0].span_id.clone()]); } #[test] @@ -1227,7 +1262,7 @@ fn synthetic_late_old_turn_response_cannot_end_current_turn_model() { h.handle(&event( "postToolUse", 146, - json!({"generation_id":"t2","tool_name":"Read","tool_use_id":"r1","tool_output":"file content"}), + json!({"generation_id":"t2","tool_name":"Read","tool_use_id":"r1","tool_input":{"file_path":"/tmp/a"},"tool_output":"file content","duration":0}), )); h.handle(&event( "afterAgentResponse", @@ -1594,7 +1629,7 @@ fn synthetic_truncated_mirror_after_stop_maps_its_first_user_record_to_native_tu 140, &mirror, second.len() as u64, - json!({"reason":"completed"}), + json!({"reason":"completed","final_status":"completed"}), )); h.finish(); assert_eq!( @@ -1637,7 +1672,7 @@ fn ambiguous_rewritten_prompt_does_not_overwrite_either_identical_native_turn() 300, &mirror, transcript.len() as u64, - json!({"reason":"completed"}), + json!({"reason":"completed","final_status":"completed"}), )); h.finish(); let rows = h.rows(); @@ -1690,7 +1725,7 @@ fn synthetic_rewrite_removing_terminal_marker_does_not_reemit_shifted_assistant_ 140, &rewritten_mirror, rewritten.len() as u64, - json!({"reason":"completed"}), + json!({"reason":"completed","final_status":"completed"}), )); h.finish(); assert_eq!(h.turns().len(), 2); @@ -1725,7 +1760,7 @@ fn synthetic_partial_usage_merges_known_counts_without_inventing_missing_values( 110, json!({ "generation_id":"t", "text":"hello", "input_tokens":20, - "cache_read_tokens":7, "cache_write_tokens":-1 + "cache_read_tokens":7, "model":"composer-2.5" }), )); let id = h.turns()[0].span_id.clone(); @@ -1739,8 +1774,8 @@ fn synthetic_partial_usage_merges_known_counts_without_inventing_missing_values( "stop", 120, json!({ - "generation_id":"t", "status":"completed", "input_tokens":-2, - "output_tokens":4 + "generation_id":"t", "status":"completed", "output_tokens":4, + "model":"composer-2.5" }), ); h.handle(&stop); @@ -1760,7 +1795,7 @@ fn synthetic_partial_usage_merges_known_counts_without_inventing_missing_values( } #[test] -fn synthetic_missing_tool_result_does_not_invent_an_empty_model_input_message() { +fn synthetic_failed_tool_without_result_does_not_invent_an_empty_model_input_message() { let mut h = Harness::new("test-session"); h.handle(&event( "beforeSubmitPrompt", @@ -1777,9 +1812,13 @@ fn synthetic_missing_tool_result_does_not_invent_an_empty_model_input_message() }), )); h.handle(&event( - "postToolUse", + "postToolUseFailure", 120, - json!({"tool_use_id":"r", "tool_name":"Read"}), + json!({ + "tool_use_id":"r", "tool_name":"Read", "tool_input":{"file_path":"/tmp/a"}, + "error_message":"read failed", "failure_type":"error", "is_interrupt":false, + "duration":5 + }), )); h.handle(&event( "afterAgentResponse", @@ -1796,3 +1835,75 @@ fn synthetic_missing_tool_result_does_not_invent_an_empty_model_input_message() true ); } + +#[test] +fn malformed_hooks_fail_the_event_instead_of_dropping_fields() { + let ctx = SessionCtx { + session_id: "test-session".into(), + config: None, + }; + for (kind, payload) in [ + ( + "afterAgentResponse", + json!({"generation_id":"t", "text":"hello", "input_tokens":-1}), + ), + ( + "preToolUse", + json!({"tool_name":5, "tool_use_id":"call-1", "tool_input":{}}), + ), + ("stop", json!({"status":{"code":"completed"}})), + ("sessionStart", json!({"workspace_roots":"/not/a/list"})), + // Fields the CLI always sends are required. + ("beforeSubmitPrompt", json!({})), + ("preToolUse", json!({"tool_name":"Shell", "tool_input":{}})), + ("sessionEnd", json!({"reason":"completed"})), + ( + "subagentStop", + json!({"subagent_type":"explore", "status":"completed"}), + ), + ] { + let mut translator = Registry::default_agents().create("cursor", "test-session"); + let error = translator + .handle(&event(kind, 100, payload), &ctx) + .unwrap_err() + .to_string(); + assert!( + error.starts_with(&format!("unexpected cursor {kind} hook format")), + "{error}" + ); + } +} + +#[test] +fn unexpected_transcript_record_types_fail_the_event() { + let dir = tempfile::tempdir().unwrap(); + let transcript = dir.path().join("transcript.jsonl"); + std::fs::write( + &transcript, + format!("{}\n", json!({"role":"user","message":{"content":42}})), + ) + .unwrap(); + let through = std::fs::metadata(&transcript).unwrap().len(); + let ctx = SessionCtx { + session_id: "test-session".into(), + config: None, + }; + let mut translator = Registry::default_agents().create("cursor", "test-session"); + let error = translator + .handle( + &mirrored( + "beforeSubmitPrompt", + 100, + &transcript, + through, + json!({"prompt":"hi"}), + ), + &ctx, + ) + .unwrap_err() + .to_string(); + assert!( + error.starts_with("unexpected cursor transcript record format"), + "{error}" + ); +}