From 0abefce62ce7cea7aed5ca1ae24a723c6d2608e9 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 30 Sep 2026 07:40:51 +0000 Subject: [PATCH 1/3] chore: sync public mirror from internal --- .repository-projection.json | 6 +- vendor/dex-loop/src/engine.rs | 127 +++++++++++++++++++++++++++------- vendor/dex-loop/src/ports.rs | 6 ++ 3 files changed, 110 insertions(+), 29 deletions(-) diff --git a/.repository-projection.json b/.repository-projection.json index dd9c18c2a..80f2915a0 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "e93942ba70f7585b80a55a24065ebbd31e57708d", + "sourceSha": "ad05c012e3f7a89a8726239fde8c1cc5d15e6ba6", "destinationRepository": "dx-corp/code", - "priorProjectedBase": "a6d447c5960e7b738a3c3a27fa564f8dcae7f448", + "priorProjectedBase": "358d47f82eb9a64f88ecfa2865002aa704e0baa5", "definitionDigest": "82936441c776e3e8edb5d215a75007ec9714a233f489d460075d79d5ef5ba32f", "toolDigest": "c244d99199a7ae3eb8ff644a99462163c23b0bb6a83ef50af01efbdca0b81d04", - "contentDigest": "13746ab7cd6f508ce691afd179ce7d3f11b2bb8d901b5db0ba91153d55c25366", + "contentDigest": "0bacaa89b0f054be3bedcf83d92a22969a884722d56c2258c40574fcad4d3a65", "publicationEligible": true } diff --git a/vendor/dex-loop/src/engine.rs b/vendor/dex-loop/src/engine.rs index 6fb68322e..804bc352f 100644 --- a/vendor/dex-loop/src/engine.rs +++ b/vendor/dex-loop/src/engine.rs @@ -18,6 +18,8 @@ use std::collections::{HashMap, HashSet}; use std::future::Future; use std::pin::{Pin, pin}; +use std::sync::{Arc, Mutex, PoisonError}; +use std::task::Poll; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use futures_util::StreamExt; @@ -137,15 +139,83 @@ enum StreamStep { /// in the step's calls. type PrefetchFuture<'e> = Pin + Send + 'e>>; -/// Reads started while the model streamed (see the module doc). Lives in -/// `Engine::run` for one step: filled by `model_step`, drained by -/// `dispatch`, and dropped whole when the attempt does not commit. -struct Prefetch<'e> { +/// The reads that are running or finished ahead of their step's commit. +struct Reads<'e> { /// Still running. Each carries its own deadline, the one a wave would /// have given it. pending: FuturesUnordered>, /// Returned before the step committed. done: HashMap, +} + +/// The reads' shared handle. A read is only polled while the engine polls +/// it, and a read can be suspended inside a log write of its own (a client +/// request, a progress label) holding the thread's row lock. If the engine +/// then awaited a log write without polling the reads, the two would wait on +/// each other forever. [`ReadsHandle::drive`] polls the reads alongside any +/// engine await, so that never happens. +#[derive(Clone)] +struct ReadsHandle<'e>(Arc>>); + +impl<'e> ReadsHandle<'e> { + fn new() -> Self { + Self(Arc::new(Mutex::new(Reads { + pending: FuturesUnordered::new(), + done: HashMap::new(), + }))) + } + + /// Never held across an await: every use is one short synchronous step. + fn lock(&self) -> std::sync::MutexGuard<'_, Reads<'e>> { + self.0.lock().unwrap_or_else(PoisonError::into_inner) + } + + fn push(&self, read: PrefetchFuture<'e>) { + self.lock().pending.push(read); + } + + fn has_pending(&self) -> bool { + !self.lock().pending.is_empty() + } + + fn take_done(&self, index: usize) -> Option { + self.lock().done.remove(&index) + } + + fn take_pending(&self) -> FuturesUnordered> { + std::mem::take(&mut self.lock().pending) + } + + /// The next read to return, if any is running. Cancel-safe: dropping it + /// loses nothing, the read stays in the set. + async fn next_done(&self) -> Option<(usize, ToolResult)> { + std::future::poll_fn(|cx| self.lock().pending.poll_next_unpin(cx)).await + } + + /// Awaits `work` while polling the running reads, so a read suspended in + /// a log write can finish it. Results that arrive meanwhile are kept in + /// `done`. + async fn drive(&self, work: impl Future) -> T { + let mut work = pin!(work); + std::future::poll_fn(|cx| { + if let Poll::Ready(output) = work.as_mut().poll(cx) { + return Poll::Ready(output); + } + let mut reads = self.lock(); + while let Poll::Ready(Some((index, result))) = reads.pending.poll_next_unpin(cx) { + reads.done.insert(index, result); + } + Poll::Pending + }) + .await + } +} + +/// Reads started while the model streamed (see the module doc). Lives in +/// `Engine::run` for one step: filled by `model_step`, drained by +/// `dispatch`, and dropped whole when the attempt does not commit. +struct Prefetch<'e> { + reads: ReadsHandle<'e>, /// Indices whose `ToolStarted` is already on the log. started: HashSet, /// The `ToolStarted` rows appended mid-stream. The model stream borrows @@ -156,8 +226,7 @@ struct Prefetch<'e> { impl Prefetch<'_> { fn new() -> Self { Self { - pending: FuturesUnordered::new(), - done: HashMap::new(), + reads: ReadsHandle::new(), started: HashSet::new(), unobserved: Vec::new(), } @@ -294,7 +363,11 @@ where return self.interrupt(ctx).await; } if ctx.open_step().is_some() { - if let Some(exit) = self.dispatch(ctx, cancel, started, &mut prefetch).await? { + let reads = prefetch.reads.clone(); + if let Some(exit) = reads + .drive(self.dispatch(ctx, cancel, started, &mut prefetch)) + .await? + { return Ok(exit); } continue; @@ -407,7 +480,7 @@ where // `FuturesUnordered::next` on an empty set resolves at once // with `None`; the guard keeps it out of the race until a // read is actually running. - let has_pending = !prefetch.pending.is_empty(); + let has_pending = prefetch.reads.has_pending(); let outcome = tokio::select! { biased; () = cancel.cancelled() => StreamStep::Cancelled, @@ -416,7 +489,7 @@ where Some(chunk) => StreamStep::Chunk(chunk), None => StreamStep::Ended, }, - done = prefetch.pending.next(), if has_pending => match done { + done = prefetch.reads.next_done(), if has_pending => match done { Some((index, result)) => StreamStep::Prefetched(index, result), None => continue, }, @@ -424,21 +497,19 @@ where match outcome { StreamStep::Chunk(Ok(ModelChunk::Thinking(delta))) => { if let Some(summary) = thinking.push(&delta, !text.is_empty()) { - self.log - .append(&[Event::ThinkingDelta { text: summary }]) - .await?; + let event = [Event::ThinkingDelta { text: summary }]; + prefetch.reads.drive(self.log.append(&event)).await?; } } StreamStep::Chunk(Ok(ModelChunk::Text(delta))) => { if let Some(summary) = thinking.flush() { - self.log - .append(&[Event::ThinkingDelta { text: summary }]) - .await?; + let event = [Event::ThinkingDelta { text: summary }]; + prefetch.reads.drive(self.log.append(&event)).await?; } let safe = filter.push(&delta); if !safe.is_empty() { text.push_str(&safe); - self.log.append_text(safe).await?; + prefetch.reads.drive(self.log.append_text(safe)).await?; } } StreamStep::Chunk(Ok(ModelChunk::ToolCall { name, args })) => { @@ -454,7 +525,7 @@ where calls.push(call); } StreamStep::Prefetched(index, result) => { - prefetch.done.insert(index, result); + prefetch.reads.lock().done.insert(index, result); } StreamStep::Chunk(Ok(ModelChunk::Usage(usage))) => { pending_usage.push(Event::Usage(usage)); @@ -478,14 +549,13 @@ where } if failure.is_none() && !wall_exceeded { if let Some(summary) = thinking.flush() { - self.log - .append(&[Event::ThinkingDelta { text: summary }]) - .await?; + let event = [Event::ThinkingDelta { text: summary }]; + prefetch.reads.drive(self.log.append(&event)).await?; } let tail = filter.finish(); if !tail.is_empty() { text.push_str(&tail); - self.log.append_text(tail).await?; + prefetch.reads.drive(self.log.append_text(tail)).await?; } } } @@ -599,7 +669,9 @@ where reasoning, served, }); - self.emit(ctx, events).await?; + // The reads started during the stream are still running: keep + // polling them through the commit (see `ReadsHandle`). + prefetch.reads.drive(self.emit(ctx, events)).await?; Ok(None) } @@ -636,7 +708,10 @@ where } // Appended without `ctx.observe`: the stream still borrows `ctx`. let event = started(call, &spec); - let cursors = self.log.append(std::slice::from_ref(&event)).await?; + let cursors = prefetch + .reads + .drive(self.log.append(std::slice::from_ref(&event))) + .await?; let [cursor] = cursors[..] else { return Err(Fenced::new(format!( "log returned {} cursors for 1 event", @@ -648,7 +723,7 @@ where let deadline = self.call_deadline(run_started); let thread = ctx.thread().clone(); let call = call.clone(); - prefetch.pending.push(Box::pin(async move { + prefetch.reads.push(Box::pin(async move { let run = self.tools.run(&thread, &call, cancel); let result = match tokio::time::timeout(deadline, run).await { Ok(result) => result, @@ -1063,7 +1138,7 @@ where let deadline = self.call_deadline(run_started); let mut running: FuturesUnordered> = FuturesUnordered::new(); for &index in &wave { - if let Some(result) = prefetch.done.remove(&index) { + if let Some(result) = prefetch.reads.take_done(index) { self.finish(ctx, &calls[index], result).await?; continue; } @@ -1083,7 +1158,7 @@ where (index, result) })); } - for future in std::mem::take(&mut prefetch.pending) { + for future in prefetch.reads.take_pending() { running.push(future); } // On `Fenced` the remaining reads are dropped: a stale owner must not diff --git a/vendor/dex-loop/src/ports.rs b/vendor/dex-loop/src/ports.rs index 1759c55ed..0d65bfe19 100644 --- a/vendor/dex-loop/src/ports.rs +++ b/vendor/dex-loop/src/ports.rs @@ -90,6 +90,12 @@ pub trait Model: Send + Sync { ctx: &'a Context, tools: &'a [&'a ToolSpec], ) -> impl Stream> + Send + 'a; + + /// Called once when a turn is admitted, before its first step, with the + /// context that step will see. An implementation may start work the + /// first request needs (it must not block: spawn it) so that request + /// finds it done. The default does nothing. + fn prepare_turn(&self, _ctx: &Context) {} } /// Which governance a tool falls under. Read by `Tools::policy`, not by the From ea207e893f2111f9aee0b41265d5ec1240dbc5e1 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 30 Sep 2026 08:42:34 +0000 Subject: [PATCH 2/3] chore: sync public mirror from internal --- .repository-projection.json | 4 ++-- packages/dex-host-rs/src/model.rs | 1 + packages/dex-host-rs/src/tools.rs | 2 ++ vendor/dex-loop/src/context.rs | 6 ++++++ vendor/dex-loop/src/engine.rs | 11 ++++++++--- vendor/dex-loop/src/lib.rs | 2 +- vendor/dex-loop/src/ports.rs | 13 +++++++++++++ vendor/dex-loop/tests/scenarios.rs | 6 +++--- vendor/dex-loop/tests/sim.rs | 1 + vendor/dex-loop/tests/sim/scenario.rs | 1 + vendor/dex-loop/tests/support/mod.rs | 3 +++ 11 files changed, 41 insertions(+), 9 deletions(-) diff --git a/.repository-projection.json b/.repository-projection.json index 80f2915a0..a6284f57b 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "ad05c012e3f7a89a8726239fde8c1cc5d15e6ba6", + "sourceSha": "55a41e5720cd66a51382058e7dde2531497304d8", "destinationRepository": "dx-corp/code", "priorProjectedBase": "358d47f82eb9a64f88ecfa2865002aa704e0baa5", "definitionDigest": "82936441c776e3e8edb5d215a75007ec9714a233f489d460075d79d5ef5ba32f", "toolDigest": "c244d99199a7ae3eb8ff644a99462163c23b0bb6a83ef50af01efbdca0b81d04", - "contentDigest": "0bacaa89b0f054be3bedcf83d92a22969a884722d56c2258c40574fcad4d3a65", + "contentDigest": "65f67c0b8492843ebdaf00a9aa205f2492e4f06d9523e41f76df5c6b3164b50d", "publicationEligible": true } diff --git a/packages/dex-host-rs/src/model.rs b/packages/dex-host-rs/src/model.rs index 560e241c1..fb6428b85 100644 --- a/packages/dex-host-rs/src/model.rs +++ b/packages/dex-host-rs/src/model.rs @@ -291,6 +291,7 @@ mod tests { )], ); let spec = ToolSpec { + description: String::new(), name: ToolName::new("fs.read_file"), label: "Read a file".into(), schema: serde_json::json!({"type": "object"}), diff --git a/packages/dex-host-rs/src/tools.rs b/packages/dex-host-rs/src/tools.rs index 0fb441224..246f0bfea 100644 --- a/packages/dex-host-rs/src/tools.rs +++ b/packages/dex-host-rs/src/tools.rs @@ -23,6 +23,7 @@ pub const WRITE_FILE: &str = "fs.write_file"; fn catalog() -> Vec { vec![ ToolSpec { + description: "Read a text file from the workspace.".into(), name: ToolName::new(READ_FILE), label: "Read a workspace file".into(), schema: serde_json::json!({ @@ -42,6 +43,7 @@ fn catalog() -> Vec { executor: ExecutorKind::InProcess, }, ToolSpec { + description: "Write text content to a file in the workspace.".into(), name: ToolName::new(WRITE_FILE), label: "Write a workspace file".into(), schema: serde_json::json!({ diff --git a/vendor/dex-loop/src/context.rs b/vendor/dex-loop/src/context.rs index 7a5f4740a..4e25df820 100644 --- a/vendor/dex-loop/src/context.rs +++ b/vendor/dex-loop/src/context.rs @@ -306,6 +306,12 @@ impl Context { self.control = self.control.max(floor); } + /// Whether a turn is running (or a queued one just began). A host that + /// finished a turn checks this before doing work between turns. + pub fn turn_running(&self) -> bool { + self.status == Status::Running + } + pub(crate) fn status(&self) -> Status { self.status } diff --git a/vendor/dex-loop/src/engine.rs b/vendor/dex-loop/src/engine.rs index 804bc352f..e43dce1c3 100644 --- a/vendor/dex-loop/src/engine.rs +++ b/vendor/dex-loop/src/engine.rs @@ -35,7 +35,7 @@ use crate::event::{ }; use crate::ports::{ Claim, Effects, ExecutorKind, Fenced, GovernanceClass, Log, Model, ModelChunk, ModelError, - ToolSpec, Tools, Verdict, + ToolSpec, Tools, Verdict, model_tool_name, }; use crate::sanitize::{DeltaFilter, Sanitizer}; @@ -1281,7 +1281,7 @@ where } else { matches .iter() - .map(|spec| format!("{}: {}", spec.name, spec.label)) + .map(|spec| format!("{}: {}", model_tool_name(spec.name.as_str()), spec.label)) .collect::>() .join("\n") }; @@ -1500,6 +1500,10 @@ where fn search_spec() -> ToolSpec { ToolSpec { + description: "Find tools this conversation does not have yet. Say what you need to do; \ + matching tools are added to your tools from the next step, and the \ + result lists their names." + .into(), name: ToolName::new(TOOLS_SEARCH), label: "Finding the right tools".into(), schema: serde_json::json!({ @@ -1613,7 +1617,8 @@ fn validate_args(spec: &ToolSpec, args: &serde_json::Value) -> Result<(), String fn unknown_tool(name: &ToolName) -> ToolResult { ToolResult::error(format!( - "unknown tool: {name}; use {TOOLS_SEARCH} to find tools" + "unknown tool: {name}; use {} to find tools", + model_tool_name(TOOLS_SEARCH), )) } diff --git a/vendor/dex-loop/src/lib.rs b/vendor/dex-loop/src/lib.rs index 014730014..31a88302c 100644 --- a/vendor/dex-loop/src/lib.rs +++ b/vendor/dex-loop/src/lib.rs @@ -40,7 +40,7 @@ pub use event::{ }; pub use ports::{ Claim, Effects, ExecutorKind, Fenced, GovernanceClass, Log, Model, ModelChunk, ModelError, - ToolSpec, Tools, Verdict, + ToolSpec, Tools, Verdict, model_tool_name, }; pub use rehydrate::rehydrate; pub use sanitize::{DeltaFilter, Lexicon, LexiconFilter, Sanitizer}; diff --git a/vendor/dex-loop/src/ports.rs b/vendor/dex-loop/src/ports.rs index 0d65bfe19..18026232b 100644 --- a/vendor/dex-loop/src/ports.rs +++ b/vendor/dex-loop/src/ports.rs @@ -127,12 +127,25 @@ pub enum ExecutorKind { Client, } +/// How model-visible text names a tool: registry names use dots +/// (`dex.read`), and providers reject dots in function names, so the model +/// sees `dex_read`. `dex-model` declares tools under this same name whenever +/// it is a valid function name, so prose and declarations agree. +pub fn model_tool_name(name: &str) -> String { + name.replace('.', "_") +} + /// One registry entry. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct ToolSpec { pub name: ToolName, /// The only tool text a surface may show. pub label: String, + /// What the model is told the tool does. Never shown on a surface (`label` + /// is). Empty means the model falls back to the schema description, then + /// the label. `Registry` fills it from `Entry::description`. + #[serde(default)] + pub description: String, pub schema: serde_json::Value, /// Read-only calls that policy allows run together in one parallel wave, /// and may run again after a restart. Every other call is a mutation and diff --git a/vendor/dex-loop/tests/scenarios.rs b/vendor/dex-loop/tests/scenarios.rs index e7c5eef60..e70a45888 100644 --- a/vendor/dex-loop/tests/scenarios.rs +++ b/vendor/dex-loop/tests/scenarios.rs @@ -1286,9 +1286,9 @@ async fn unknown_and_denied_calls_are_visible_to_the_model() { assert_eq!( view(&model.seen()[1])[2..], strings(&[ - "tool:t1-1-0:err:unknown tool: no_such_tool; use tools.search to find tools", + "tool:t1-1-0:err:unknown tool: no_such_tool; use tools_search to find tools", "tool:t1-1-1:err:denied: destructive calls are off", - "tool:t1-1-2:err:unknown tool: crm.lookup; use tools.search to find tools", + "tool:t1-1-2:err:unknown tool: crm.lookup; use tools_search to find tools", "tool:t1-1-3:ok:out/t1-1-3", ]) ); @@ -1589,7 +1589,7 @@ async fn tools_search_exposes_schemas_for_the_next_step() { assert_eq!(tools.run_ids(), strings(&["t1-2-0"])); assert_eq!( view(&model.seen()[1])[2..], - strings(&["tool:t1-1-0:ok:crm.lookup: Label for crm.lookup"]) + strings(&["tool:t1-1-0:ok:crm_lookup: Label for crm.lookup"]) ); assert_eq!(log.rehydrate(), ctx); } diff --git a/vendor/dex-loop/tests/sim.rs b/vendor/dex-loop/tests/sim.rs index 9459cc33f..0e4eb2919 100644 --- a/vendor/dex-loop/tests/sim.rs +++ b/vendor/dex-loop/tests/sim.rs @@ -260,6 +260,7 @@ fn thread_for(name: &str) -> ThreadId { fn mutation_catalog() -> Vec { vec![ToolSpec { + description: String::new(), name: ToolName::new("mutator"), label: "Mutator".into(), schema: serde_json::json!({"type": "object"}), diff --git a/vendor/dex-loop/tests/sim/scenario.rs b/vendor/dex-loop/tests/sim/scenario.rs index 94587ab61..41399a030 100644 --- a/vendor/dex-loop/tests/sim/scenario.rs +++ b/vendor/dex-loop/tests/sim/scenario.rs @@ -58,6 +58,7 @@ fn catalog() -> Vec { let spec = |name: &str, read_only: bool, governance: GovernanceClass, executor: ExecutorKind| { ToolSpec { + description: String::new(), name: ToolName::new(name), label: format!("Label for {name}"), schema: serde_json::json!({"type": "object"}), diff --git a/vendor/dex-loop/tests/support/mod.rs b/vendor/dex-loop/tests/support/mod.rs index 45fa76f91..64d68b421 100644 --- a/vendor/dex-loop/tests/support/mod.rs +++ b/vendor/dex-loop/tests/support/mod.rs @@ -362,6 +362,7 @@ pub fn read_tool(name: &str) -> ToolSpec { #[allow(dead_code)] // used by tests/prefetch.rs only pub fn strict_read_tool(name: &str) -> ToolSpec { ToolSpec { + description: String::new(), schema: serde_json::json!({ "type": "object", "properties": {"key": {"type": "string"}}, @@ -398,6 +399,7 @@ pub fn client_tool(name: &str, read_only: bool) -> ClientToolSpec { /// would offer one already resolved from a client's declaration. pub fn client_executed_tool(name: &str, read_only: bool) -> ToolSpec { ToolSpec { + description: String::new(), name: ToolName::new(name), label: format!("Label for {name}"), schema: serde_json::json!({"type": "object"}), @@ -419,6 +421,7 @@ pub fn hidden_read_tool(name: &str) -> ToolSpec { fn spec(name: &str, read_only: bool, core: bool, executor: ExecutorKind) -> ToolSpec { ToolSpec { + description: String::new(), name: ToolName::new(name), label: format!("Label for {name}"), schema: serde_json::json!({"type": "object"}), From ef9ef026eb85fc60c28dcc17d30b488449635e9f Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 30 Sep 2026 15:12:20 +0000 Subject: [PATCH 3/3] chore: sync public mirror from internal --- .repository-projection.json | 4 +- packages/dex-host-rs/src/model.rs | 27 +- .../runtime-rs/src/agent/credential_store.rs | 62 +++- packages/runtime-rs/src/agent/native.rs | 46 ++- .../src/agent/native/provider_payload.rs | 324 ++++++++++++++++++ packages/runtime-rs/src/agent/native/tests.rs | 15 + vendor/dex-loop/src/budget.rs | 1 + vendor/dex-loop/src/compaction.rs | 1 + vendor/dex-loop/src/engine.rs | 123 ++++++- vendor/dex-loop/src/event.rs | 233 +++++++++++++ vendor/dex-loop/src/lib.rs | 4 +- vendor/dex-loop/src/ports.rs | 21 +- vendor/dex-loop/tests/client_tool_replay.rs | 1 + vendor/dex-loop/tests/prefetch.rs | 74 +++- vendor/dex-loop/tests/read_wave_progress.rs | 211 ++++++++++++ vendor/dex-loop/tests/scenarios.rs | 125 +++++++ vendor/dex-loop/tests/support/mod.rs | 2 + 17 files changed, 1207 insertions(+), 67 deletions(-) create mode 100644 packages/runtime-rs/src/agent/native/provider_payload.rs create mode 100644 vendor/dex-loop/tests/read_wave_progress.rs diff --git a/.repository-projection.json b/.repository-projection.json index a6284f57b..cb609fa12 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "55a41e5720cd66a51382058e7dde2531497304d8", + "sourceSha": "c30dc2f64dce3300144e0e0d0dc97eb97b8744ee", "destinationRepository": "dx-corp/code", "priorProjectedBase": "358d47f82eb9a64f88ecfa2865002aa704e0baa5", "definitionDigest": "82936441c776e3e8edb5d215a75007ec9714a233f489d460075d79d5ef5ba32f", "toolDigest": "c244d99199a7ae3eb8ff644a99462163c23b0bb6a83ef50af01efbdca0b81d04", - "contentDigest": "65f67c0b8492843ebdaf00a9aa205f2492e4f06d9523e41f76df5c6b3164b50d", + "contentDigest": "fc3edcce20cf8d94ea1cd95fce135752740dc77c7bd79715e242599e385cb806", "publicationEligible": true } diff --git a/packages/dex-host-rs/src/model.rs b/packages/dex-host-rs/src/model.rs index fb6428b85..b2f708737 100644 --- a/packages/dex-host-rs/src/model.rs +++ b/packages/dex-host-rs/src/model.rs @@ -192,10 +192,13 @@ impl ChunkTranslator { StreamEvent::Usage { input_tokens, output_tokens, - .. + cache_read_tokens, + cache_creation_tokens, } => vec![Ok(ModelChunk::Usage(Usage { input_tokens, output_tokens, + cache_read_input_tokens: cache_read_tokens.unwrap_or_default(), + cache_creation_input_tokens: cache_creation_tokens.unwrap_or_default(), // Not attributed here; see the module doc comment. cost_micros: 0, }))], @@ -255,6 +258,28 @@ mod tests { } } + #[test] + fn retains_provider_reported_cache_usage() { + let chunks = ChunkTranslator::default().translate(StreamEvent::Usage { + input_tokens: 10, + output_tokens: 3, + cache_read_tokens: Some(20), + cache_creation_tokens: Some(7), + }); + assert!(matches!(&chunks[0], Ok(ModelChunk::Usage(usage)) + if usage.input_tokens == 10 && usage.output_tokens == 3 + && usage.cache_read_input_tokens == 20 + && usage.cache_creation_input_tokens == 7)); + let absent = ChunkTranslator::default().translate(StreamEvent::Usage { + input_tokens: 10, + output_tokens: 3, + cache_read_tokens: None, + cache_creation_tokens: None, + }); + assert!(matches!(&absent[0], Ok(ModelChunk::Usage(usage)) + if usage.cache_read_input_tokens == 0 && usage.cache_creation_input_tokens == 0)); + } + #[tokio::test] async fn translates_text_then_tool_call_then_usage() { let scripted = ScriptedClient::new( diff --git a/packages/runtime-rs/src/agent/credential_store.rs b/packages/runtime-rs/src/agent/credential_store.rs index b1ca8436a..c5bc5382a 100644 --- a/packages/runtime-rs/src/agent/credential_store.rs +++ b/packages/runtime-rs/src/agent/credential_store.rs @@ -829,6 +829,22 @@ impl CredentialVault { Ok(state.attestation()) } + /// Protocol identifiers are not prose: a one-character password inside + /// a longer identifier is an incidental spelling overlap. Exact values, + /// standalone characters, longer credentials and unowned references still + /// fail. Content values retain the stricter full substring check. + pub(crate) fn attest_provider_identifier( + &self, + text: &str, + ) -> Result { + let state = self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + attest_provider_identifier_with_store(&state.store, text)?; + Ok(state.attestation()) + } + /// Check JSON before serialization can escape credential bytes. Object /// keys are provider content too, even though vault_in_json leaves them /// intact to preserve the schema. @@ -848,7 +864,7 @@ impl CredentialVault { } serde_json::Value::Object(entries) => { for (key, value) in entries { - attest_provider_text_with_store(store, key)?; + attest_provider_identifier_with_store(store, key)?; visit(store, value)?; } } @@ -1004,6 +1020,50 @@ fn attest_provider_text_with_store( Ok(()) } +fn attest_provider_identifier_with_store( + store: &CredentialStore, + text: &str, +) -> Result<(), &'static str> { + let references = store.references(); + for (start, end) in credential_reference_like_ranges(text) { + if !references.contains(&text[start..end]) { + return Err("provider identifier contains an unowned credential reference"); + } + } + let protected = REFERENCE_PATTERN + .find_iter(text) + .map(|reference| (reference.start(), reference.end())) + .collect::>(); + for credential in store.credentials.values() { + let value = credential.value.as_str(); + if value.is_empty() { + continue; + } + for (start, _) in text.match_indices(value) { + let end = start + value.len(); + if protected + .iter() + .any(|(left, right)| start >= *left && end <= *right) + { + continue; + } + let single_character = value.chars().count() == 1; + let within_word = text[..start] + .chars() + .next_back() + .is_some_and(char::is_alphanumeric) + || text[end..] + .chars() + .next() + .is_some_and(char::is_alphanumeric); + if !single_character || !within_word { + return Err("provider identifier contains a plaintext vaulted credential"); + } + } + } + Ok(()) +} + impl fmt::Debug for CredentialStore { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter diff --git a/packages/runtime-rs/src/agent/native.rs b/packages/runtime-rs/src/agent/native.rs index c9613ad7b..7e17ec558 100644 --- a/packages/runtime-rs/src/agent/native.rs +++ b/packages/runtime-rs/src/agent/native.rs @@ -447,6 +447,7 @@ mod deferred_tool_tests; mod model_dynamics; mod provider_history; mod provider_loop; +mod provider_payload; mod read_only_tools; mod side_questions; #[cfg(test)] @@ -3794,12 +3795,7 @@ fn vault_provider_history( messages: &[Message], credential_vault: &CredentialVault, ) -> Result> { - // Scan the final provider projection as a last boundary check. This also - // catches raw material introduced by prompts, hooks, or extensions after - // tool output was first vaulted. Existing references remain opaque. - let serialized = serde_json::to_value(messages).context("serialize provider history")?; - let safe = credential_vault.vault_in_json(&serialized); - serde_json::from_value(safe).context("deserialize vaulted provider history") + provider_payload::vault_history(messages, credential_vault) } fn vault_provider_history_shared( @@ -3857,10 +3853,7 @@ impl ProviderSafeRequest { // Config preparation may discover a credential present in history. let messages = vault_provider_history_shared(&messages, vault)?; config.system = config.system.map(|system| vault.vault_in_text(&system)); - let history_json = serde_json::to_value(messages.as_ref())?; - let attestation = vault - .attest_provider_json(&history_json) - .map_err(anyhow::Error::msg)?; + let attestation = provider_payload::attest_history(&messages, vault)?; if let Some(system) = &config.system { let system_attestation = vault .attest_provider_text(system) @@ -3870,10 +3863,7 @@ impl ProviderSafeRequest { "credential vault changed during provider request preparation" ); } - let tools_json = serde_json::to_value(config.tools.as_ref())?; - let tools_attestation = vault - .attest_provider_json(&tools_json) - .map_err(anyhow::Error::msg)?; + let tools_attestation = provider_payload::attest_tools(&config.tools, vault)?; anyhow::ensure!( attestation == tools_attestation, "credential vault changed during provider request preparation" @@ -3900,9 +3890,9 @@ fn vault_provider_tools( ) -> Result>> { let mut safe_tools = Vec::with_capacity(tools.len()); for tool in tools { - if credential_vault.vault_in_text(&tool.name) != tool.name { - anyhow::bail!("credential detected in provider tool name"); - } + credential_vault + .attest_provider_identifier(&tool.name) + .map_err(anyhow::Error::msg)?; let mut safe = tool.clone(); safe.description = credential_vault.vault_in_text(&tool.description); safe.input_schema = vault_provider_schema(&tool.input_schema, credential_vault)?; @@ -3912,9 +3902,9 @@ fn vault_provider_tools( // only as a schema key or tool name. Validate structural names again with // the complete vault before handing the definitions to the provider. for safe in &mut safe_tools { - if credential_vault.vault_in_text(&safe.name) != safe.name { - anyhow::bail!("credential detected in provider tool name"); - } + credential_vault + .attest_provider_identifier(&safe.name) + .map_err(anyhow::Error::msg)?; safe.description = credential_vault.vault_in_text(&safe.description); safe.input_schema = vault_provider_schema(&safe.input_schema, credential_vault)?; } @@ -3933,10 +3923,18 @@ fn vault_provider_schema(value: &Value, credential_vault: &CredentialVault) -> R Value::Object(entries) => { let mut safe = serde_json::Map::new(); for (key, value) in entries { - if credential_vault.vault_in_text(key) != *key { - anyhow::bail!("credential detected in provider tool schema key"); - } - safe.insert(key.clone(), vault_provider_schema(value, credential_vault)?); + credential_vault + .attest_provider_identifier(key) + .map_err(anyhow::Error::msg)?; + safe.insert( + key.clone(), + if provider_payload::schema_protocol_keyword(key) { + provider_payload::attest_protocol_json(value, credential_vault)?; + value.clone() + } else { + vault_provider_schema(value, credential_vault)? + }, + ); } Ok(Value::Object(safe)) } diff --git a/packages/runtime-rs/src/agent/native/provider_payload.rs b/packages/runtime-rs/src/agent/native/provider_payload.rs new file mode 100644 index 000000000..b52c8d077 --- /dev/null +++ b/packages/runtime-rs/src/agent/native/provider_payload.rs @@ -0,0 +1,324 @@ +//! Vault data fields without rewriting protocol tags or continuation bytes. + +use super::*; + +pub(super) fn vault_history(messages: &[Message], vault: &CredentialVault) -> Result> { + let mut safe = messages.to_vec(); + for message in &mut safe { + match &mut message.content { + MessageContent::Text(text) => *text = vault.vault_in_text(text), + MessageContent::Blocks(blocks) => { + for block in blocks { + match block { + ContentBlock::Text { text } => *text = vault.vault_in_text(text), + ContentBlock::Thinking { thinking, .. } => { + *thinking = vault.vault_in_text(thinking); + } + ContentBlock::ToolUse { input, .. } => *input = vault.vault_in_json(input), + ContentBlock::ToolResult { content, .. } => { + *content = vault.vault_in_text(content); + } + ContentBlock::Image { + source: ImageSource::Url { url }, + } => { + *url = vault.vault_in_text(url); + } + ContentBlock::Image { + source: ImageSource::Base64 { .. }, + } => {} + } + } + } + } + } + Ok(safe) +} + +pub(super) fn attest_history( + messages: &[Message], + vault: &CredentialVault, +) -> Result { + let attestation = vault.attest_provider_text("").map_err(anyhow::Error::msg)?; + let text = |value: &str| check(attestation, vault.attest_provider_text(value)); + let identifier = |value: &str| check(attestation, vault.attest_provider_identifier(value)); + for message in messages { + match &message.content { + MessageContent::Text(value) => text(value)?, + MessageContent::Blocks(blocks) => { + for block in blocks { + match block { + ContentBlock::Text { text: value } => text(value)?, + ContentBlock::Thinking { + thinking, + signature, + } => { + text(thinking)?; + if let Some(signature) = signature { + identifier(signature)?; + } + } + ContentBlock::ToolUse { + id, + name, + input, + gemini_context, + } => { + identifier(id)?; + identifier(name)?; + check(attestation, vault.attest_provider_json(input))?; + if let Some(context) = gemini_context { + identifier(&context.native_name)?; + if let Some(id) = &context.native_id { + identifier(id)?; + } + if let Some(signature) = &context.thought_signature { + identifier(signature)?; + } + } + } + ContentBlock::ToolResult { + tool_use_id, + content, + .. + } => { + identifier(tool_use_id)?; + text(content)?; + } + ContentBlock::Image { source } => match source { + ImageSource::Base64 { media_type, data } => { + identifier(media_type)?; + identifier(data)?; + } + ImageSource::Url { url } => text(url)?, + }, + } + } + } + } + } + Ok(attestation) +} + +fn check( + expected: CredentialAttestation, + actual: std::result::Result, +) -> Result<()> { + anyhow::ensure!( + expected == actual.map_err(anyhow::Error::msg)?, + "credential vault changed during provider request preparation" + ); + Ok(()) +} + +pub(super) fn schema_protocol_keyword(key: &str) -> bool { + matches!( + key, + "type" + | "required" + | "enum" + | "const" + | "format" + | "pattern" + | "$ref" + | "$schema" + | "$id" + | "$anchor" + ) +} + +pub(super) fn attest_protocol_json(value: &Value, vault: &CredentialVault) -> Result<()> { + match value { + Value::String(text) => { + vault + .attest_provider_identifier(text) + .map_err(anyhow::Error::msg)?; + } + Value::Array(values) => { + for value in values { + attest_protocol_json(value, vault)?; + } + } + Value::Object(entries) => { + for (key, value) in entries { + vault + .attest_provider_identifier(key) + .map_err(anyhow::Error::msg)?; + attest_protocol_json(value, vault)?; + } + } + _ => {} + } + Ok(()) +} + +fn attest_schema(value: &Value, vault: &CredentialVault) -> Result<()> { + match value { + Value::String(text) => { + vault + .attest_provider_text(text) + .map_err(anyhow::Error::msg)?; + } + Value::Array(values) => { + for value in values { + attest_schema(value, vault)?; + } + } + Value::Object(entries) => { + for (key, value) in entries { + vault + .attest_provider_identifier(key) + .map_err(anyhow::Error::msg)?; + if schema_protocol_keyword(key) { + attest_protocol_json(value, vault)?; + } else { + attest_schema(value, vault)?; + } + } + } + _ => {} + } + Ok(()) +} + +pub(super) fn attest_tools( + tools: &[Tool], + vault: &CredentialVault, +) -> Result { + let attestation = vault.attest_provider_text("").map_err(anyhow::Error::msg)?; + for tool in tools { + check(attestation, vault.attest_provider_identifier(&tool.name))?; + check(attestation, vault.attest_provider_text(&tool.description))?; + attest_schema(&tool.input_schema, vault)?; + } + anyhow::ensure!( + vault.has_attestation(attestation), + "credential vault changed during provider request preparation" + ); + Ok(attestation) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_short_password_cannot_rewrite_roles_tags_schema_or_continuation_bytes() { + let vault = CredentialVault::new(); + let reference = vault.store("a", crate::agent::CredentialType::Password); + let history = Arc::new(vec![ + Message { + role: Role::Assistant, + content: MessageContent::Blocks(vec![ + ContentBlock::Thinking { + thinking: "a".into(), + signature: Some("dataaaaSignature".into()), + }, + ContentBlock::ToolUse { + id: "call_1".into(), + name: "bash".into(), + input: serde_json::json!({"path":"a"}), + gemini_context: Some(crate::ai::GeminiToolContext { + native_name: "bash".into(), + native_id: Some("call_1".into()), + thought_signature: Some("dataaaaSignature".into()), + }), + }, + ContentBlock::Image { + source: ImageSource::Base64 { + media_type: "image/png".into(), + data: "aaaa".into(), + }, + }, + ]), + }, + Message { + role: Role::User, + content: MessageContent::Blocks(vec![ContentBlock::ToolResult { + tool_use_id: "call_1".into(), + content: "a".into(), + is_error: Some(false), + }]), + }, + Message { + role: Role::System, + content: MessageContent::Text("a".into()), + }, + ]); + let original = serde_json::to_value(history.as_ref()).unwrap(); + let config = RequestConfig { + system: Some("a".into()), + tools: Arc::new(vec![Tool { + name: "bash".into(), + description: "a".into(), + input_schema: serde_json::json!({"type":"object", "additionalProperties":false, + "required":["path"], "properties":{"path":{"type":"array", "items":{"type":"string"}}}}), + schema_enforcement: crate::ai::ToolSchemaEnforcement::Require, + }]), + ..RequestConfig::default() + }; + let expected_schema = config.tools[0].input_schema.clone(); + let safe = ProviderSafeRequest::prepare(&history, config, &vault).unwrap(); + safe.ensure_current(&vault).unwrap(); + let wire = serde_json::to_value(safe.messages.as_ref()).unwrap(); + assert_eq!(wire[0]["role"], "assistant"); + assert_eq!(wire[1]["role"], "user"); + assert_eq!(wire[2]["role"], "system"); + assert_eq!(wire[0]["content"][0]["type"], "thinking"); + assert_eq!(wire[0]["content"][0]["thinking"], reference); + assert_eq!(wire[0]["content"][0]["signature"], "dataaaaSignature"); + assert_eq!(wire[0]["content"][1]["name"], "bash"); + assert_eq!(wire[0]["content"][1]["id"], "call_1"); + assert_eq!(wire[0]["content"][1]["input"]["path"], reference); + assert_eq!( + wire[0]["content"][1]["gemini_context"], + original[0]["content"][1]["gemini_context"] + ); + assert_eq!(wire[0]["content"][2], original[0]["content"][2]); + assert_eq!(wire[1]["content"][0]["tool_use_id"], "call_1"); + assert_eq!(wire[1]["content"][0]["content"], reference); + assert_eq!(wire[2]["content"], reference); + assert_eq!(safe.config.system.as_deref(), Some(reference.as_str())); + assert_eq!(safe.config.tools[0].description, reference); + assert_eq!(safe.config.tools[0].input_schema, expected_schema); + assert_eq!( + safe.config.tools[0].schema_enforcement, + crate::ai::ToolSchemaEnforcement::Require + ); + assert_eq!(serde_json::to_value(history.as_ref()).unwrap(), original); + let _: Vec = + serde_json::from_value(wire).expect("typed history still round trips"); + } + + #[test] + fn protocol_identifiers_still_reject_actual_credentials_and_forged_references() { + let vault = CredentialVault::new(); + vault.store("a", crate::agent::CredentialType::Password); + for identifier in ["a", "prefix_a_suffix", "{{CRED|password|0123456789ab}}"] { + assert!( + vault.attest_provider_identifier(identifier).is_err(), + "{identifier}" + ); + } + assert!( + vault.attest_provider_text("bash").is_err(), + "free content stays strict" + ); + assert!( + vault + .attest_provider_json(&serde_json::json!({"a":"safe"})) + .is_err() + ); + let longer = "vaulted-token-1234567890"; + vault.store(longer, crate::agent::CredentialType::Token); + assert!( + vault + .attest_provider_identifier(&format!("prefix{longer}suffix")) + .is_err() + ); + assert!( + vault + .attest_provider_json(&serde_json::json!({"nested":{"value":longer}})) + .is_err() + ); + } +} diff --git a/packages/runtime-rs/src/agent/native/tests.rs b/packages/runtime-rs/src/agent/native/tests.rs index 294c522f3..064047f1a 100644 --- a/packages/runtime-rs/src/agent/native/tests.rs +++ b/packages/runtime-rs/src/agent/native/tests.rs @@ -7215,6 +7215,21 @@ fn provider_history_and_tool_execution_preserve_references() { ); } +#[test] +fn provider_history_role_survives_a_short_registered_password() { + let vault = CredentialVault::new(); + let reference = vault.store("a", crate::agent::CredentialType::Password); + let history = Arc::new(vec![Message { + role: Role::Assistant, + content: MessageContent::text("a"), + }]); + let safe = ProviderSafeRequest::prepare(&history, RequestConfig::default(), &vault) + .expect("credential substitution must preserve typed provider history"); + assert_eq!(safe.messages[0].role, Role::Assistant); + assert_eq!(safe.messages[0].content.as_text(), Some(reference.as_str())); + safe.ensure_current(&vault).unwrap(); +} + #[test] fn incident_tool_output_remains_opaque_through_tool_execution() { let vault = CredentialVault::new(); diff --git a/vendor/dex-loop/src/budget.rs b/vendor/dex-loop/src/budget.rs index 74a9ceea3..5f85344db 100644 --- a/vendor/dex-loop/src/budget.rs +++ b/vendor/dex-loop/src/budget.rs @@ -96,6 +96,7 @@ mod tests { input_tokens: tokens, output_tokens: 0, cost_micros: cost, + ..Usage::default() }; let short = Duration::from_millis(1); assert_eq!(budget.exhausted(1, usage(10, 1), short), None); diff --git a/vendor/dex-loop/src/compaction.rs b/vendor/dex-loop/src/compaction.rs index bf11c459c..6e66e9b41 100644 --- a/vendor/dex-loop/src/compaction.rs +++ b/vendor/dex-loop/src/compaction.rs @@ -218,6 +218,7 @@ mod tests { calls, reasoning: None, served: None, + timing: None, }, ] } diff --git a/vendor/dex-loop/src/engine.rs b/vendor/dex-loop/src/engine.rs index e43dce1c3..dca66abd5 100644 --- a/vendor/dex-loop/src/engine.rs +++ b/vendor/dex-loop/src/engine.rs @@ -15,7 +15,7 @@ //! those calls as not run; a committed step adopts their results in place //! of running them again. -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::future::Future; use std::pin::{Pin, pin}; use std::sync::{Arc, Mutex, PoisonError}; @@ -146,6 +146,15 @@ struct Reads<'e> { pending: FuturesUnordered>, /// Returned before the step committed. done: HashMap, + completion_order: VecDeque, +} + +impl Reads<'_> { + fn record_done(&mut self, index: usize, result: ToolResult) { + if self.done.insert(index, result).is_none() { + self.completion_order.push_back(index); + } + } } /// The reads' shared handle. A read is only polled while the engine polls @@ -162,6 +171,7 @@ impl<'e> ReadsHandle<'e> { Self(Arc::new(Mutex::new(Reads { pending: FuturesUnordered::new(), done: HashMap::new(), + completion_order: VecDeque::new(), }))) } @@ -192,6 +202,20 @@ impl<'e> ReadsHandle<'e> { std::future::poll_fn(|cx| self.lock().pending.poll_next_unpin(cx)).await } + /// Drain peers completed during a log append before polling remaining reads. + async fn next_completed(&self) -> Option<(usize, ToolResult)> { + std::future::poll_fn(|cx| { + let mut reads = self.lock(); + while let Some(index) = reads.completion_order.pop_front() { + if let Some(result) = reads.done.remove(&index) { + return Poll::Ready(Some((index, result))); + } + } + reads.pending.poll_next_unpin(cx) + }) + .await + } + /// Awaits `work` while polling the running reads, so a read suspended in /// a log write can finish it. Results that arrive meanwhile are kept in /// `done`. @@ -203,7 +227,7 @@ impl<'e> ReadsHandle<'e> { } let mut reads = self.lock(); while let Poll::Ready(Some((index, result))) = reads.pending.poll_next_unpin(cx) { - reads.done.insert(index, result); + reads.record_done(index, result); } Poll::Pending }) @@ -460,6 +484,7 @@ where // The route and model that served this attempt, kept for every // `ModelStepCompleted` below, including a cut-off or cancelled one. let mut served = None; + let mut timing = None; { // The answer-only call offers nothing, not even `tools.search`. let answer_only = self.budget.answer_only(step.saturating_sub(1)); @@ -525,7 +550,7 @@ where calls.push(call); } StreamStep::Prefetched(index, result) => { - prefetch.reads.lock().done.insert(index, result); + prefetch.reads.lock().record_done(index, result); } StreamStep::Chunk(Ok(ModelChunk::Usage(usage))) => { pending_usage.push(Event::Usage(usage)); @@ -536,6 +561,9 @@ where StreamStep::Chunk(Ok(ModelChunk::Served(by))) => { served = Some(by); } + StreamStep::Chunk(Ok(ModelChunk::Timing(t))) => { + timing = Some(t); + } StreamStep::Chunk(Err(error)) => { failure = Some(error.message); break; @@ -600,6 +628,7 @@ where calls: Vec::new(), reasoning: None, served: served.clone(), + timing: timing.take(), }); events.push(Event::Final { text }); self.emit(ctx, events).await?; @@ -625,6 +654,7 @@ where calls: Vec::new(), reasoning: None, served: served.clone(), + timing: timing.take(), }); self.emit(ctx, events).await?; return self.interrupt(ctx).await.map(Some); @@ -654,6 +684,7 @@ where calls: Vec::new(), reasoning, served, + timing, }); if !continues { events.push(Event::Final { text }); @@ -668,6 +699,7 @@ where calls, reasoning, served, + timing, }); // The reads started during the stream are still running: keep // polling them through the commit (see `ReadsHandle`). @@ -919,6 +951,12 @@ where .await?; continue; } + // No park: the preview is the call's result, nothing runs, and + // the model asks the user in its reply. + Verdict::NeedsConfirmation { preview } => { + self.finish(ctx, call, ToolResult::error(preview)).await?; + continue; + } verdict => verdict, }; if let Some(decision) = decision.as_ref() { @@ -939,14 +977,29 @@ where // Policy asked for approval: no human is asked. The call is // granted at once and the receipt goes to the log before the // effect; the pending wave runs first so effects keep order. - if let (None, Verdict::NeedsApproval { approval, summary }) = (decision, verdict) { - if self - .flush(ctx, &calls, &mut wave, cancel, run_started, prefetch) - .await? - { - break; + match (decision, verdict) { + (None, Verdict::NeedsApproval { approval, summary }) => { + if self + .flush(ctx, &calls, &mut wave, cancel, run_started, prefetch) + .await? + { + break; + } + self.auto_approve(ctx, call, approval, summary).await?; } - self.auto_approve(ctx, call, approval, summary).await?; + // The user confirmed this exact call in chat: the receipt + // names them, not the auto approver. + (None, Verdict::Confirmed { approval, summary }) => { + if self + .flush(ctx, &calls, &mut wave, cancel, run_started, prefetch) + .await? + { + break; + } + self.record_grant(ctx, call, approval, summary, call.principal.clone()) + .await?; + } + _ => {} } if spec.executor == ExecutorKind::User { @@ -1136,7 +1189,7 @@ where // gets the full time. A read that overruns is dropped and finished // `Failed`; a read has no effect to wait for, so retrying is safe. let deadline = self.call_deadline(run_started); - let mut running: FuturesUnordered> = FuturesUnordered::new(); + let running = ReadsHandle::new(); for &index in &wave { if let Some(result) = prefetch.reads.take_done(index) { self.finish(ctx, &calls[index], result).await?; @@ -1163,7 +1216,7 @@ where } // On `Fenced` the remaining reads are dropped: a stale owner must not // append, and the new owner runs them again. - while let Some((index, result)) = running.next().await { + while let Some((index, result)) = running.next_completed().await { let still_open = ctx .open_step() .and_then(|step| step.states.get(index)) @@ -1173,7 +1226,9 @@ where if !still_open || !(wave.contains(&index) || prefetch.started.contains(&index)) { continue; } - self.finish(ctx, &calls[index], result).await?; + running + .drive(self.finish(ctx, &calls[index], result)) + .await?; } Ok(()) } @@ -1326,15 +1381,28 @@ where /// `dex_tools::client::declare` for dex-runtime's host); the engine /// does not merge them in itself, so they are never offered twice. fn offered(&self, ctx: &Context) -> Vec { - std::iter::once(self.search.clone()) + // Search and the core tools come first, in catalog order, on every + // step. Exposed tools follow in the order they were exposed, so an + // exposure only appends: the prefix the provider cached last step is + // unchanged and only the new tail is uncached. + let mut offered: Vec = std::iter::once(self.search.clone()) .chain( self.tools .catalog() .iter() - .filter(|spec| spec.core || ctx.exposed_tools().contains(&spec.name)) + .filter(|spec| spec.core) .cloned(), ) - .collect() + .collect(); + for name in ctx.exposed_tools() { + if offered.iter().any(|spec| &spec.name == name) { + continue; + } + if let Some(spec) = self.tools.catalog().iter().find(|spec| &spec.name == name) { + offered.push(spec.clone()); + } + } + offered } /// A tool the model was offered. Calls to anything else are unknown. @@ -1461,6 +1529,26 @@ where call: &ProposedCall, approval: ApprovalId, summary: String, + ) -> Result { + self.record_grant( + ctx, + call, + approval, + summary, + PrincipalId::new(AUTO_APPROVER), + ) + .await + } + + /// Writes the durable grant receipt (`Event::AutoApproved`) for `call` + /// under `principal`. + async fn record_grant( + &self, + ctx: &mut Context, + call: &ProposedCall, + approval: ApprovalId, + summary: String, + principal: PrincipalId, ) -> Result { self.emit( ctx, @@ -1469,7 +1557,7 @@ where approval, args_digest: call.args_digest.clone(), summary, - principal: PrincipalId::new(AUTO_APPROVER), + principal, }], ) .await?; @@ -1510,6 +1598,7 @@ fn search_spec() -> ToolSpec { "type": "object", "properties": {"query": {"type": "string", "description": "What you need to do"}}, "required": ["query"], + "additionalProperties": false, }), read_only: true, core: true, diff --git a/vendor/dex-loop/src/event.rs b/vendor/dex-loop/src/event.rs index 6d8f5256a..f8f4759bc 100644 --- a/vendor/dex-loop/src/event.rs +++ b/vendor/dex-loop/src/event.rs @@ -152,9 +152,18 @@ pub struct ServedBy { /// Model spend reported by one model response. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct Usage { + /// All prompt tokens, cached or not. pub input_tokens: u64, pub output_tokens: u64, pub cost_micros: u64, + /// The part of `input_tokens` served from the prompt cache. Zero on rows + /// written before the field existed, and for providers that do not + /// report it. + #[serde(default)] + pub cache_read_input_tokens: u64, + /// The part of `input_tokens` written to the prompt cache. Same default. + #[serde(default)] + pub cache_creation_input_tokens: u64, } impl Usage { @@ -168,9 +177,34 @@ impl AddAssign for Usage { self.input_tokens = self.input_tokens.saturating_add(other.input_tokens); self.output_tokens = self.output_tokens.saturating_add(other.output_tokens); self.cost_micros = self.cost_micros.saturating_add(other.cost_micros); + self.cache_read_input_tokens = self + .cache_read_input_tokens + .saturating_add(other.cache_read_input_tokens); + self.cache_creation_input_tokens = self + .cache_creation_input_tokens + .saturating_add(other.cache_creation_input_tokens); } } +/// Where one model step's wall clock went. Every span is measured from the +/// moment the step's request began. Nothing here is request or response +/// content, and it never enters the model request. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct StepTiming { + /// Building the request. Zero on a retry. + pub prepare_ms: u64, + /// Minting the gateway bearer. + pub mint_ms: u64, + /// Request start to the gateway's response headers. + pub headers_ms: u64, + /// Request start to the first bytes of the response body. + pub first_event_ms: Option, + /// Request start to the first text delta released to the loop. + pub first_text_ms: Option, + /// Request start to the end of the response. + pub total_ms: u64, +} + /// The default for `ClientToolRequested::deadline_ms` on a row written /// before that field existed: never expires, rather than timing out every /// call still parked from before this field shipped. @@ -463,6 +497,12 @@ pub enum Event { /// existed or from a model port that does not report it. #[serde(default, skip_serializing_if = "Option::is_none")] served: Option, + /// Where the step's time went (`ModelChunk::Timing`); `None` for + /// steps logged before the field existed, cut-off steps, or a model + /// port that does not report it. Debug data: never part of model + /// history. + #[serde(default, skip_serializing_if = "Option::is_none")] + timing: Option, }, /// A model attempt with no `ModelStepCompleted` (a crash mid-stream or a /// model failure). Its streamed text is dropped from model context; @@ -583,10 +623,125 @@ impl Event { } } +impl Event { + /// Storage-adapter metadata for a lossless JSON-string event. JSONB + /// cannot represent NUL codepoints directly, but it can store escaped + /// JSON text inside a string. This key is outside the event vocabulary. + pub const STORED_JSON_V1_KEY: &'static str = "_dex_event_json_v1"; + + /// Intentionally unsupported by ordinary `Event` deserialization: an + /// older reader must refuse an encoded row instead of executing its + /// non-authoritative, sanitized projection. + pub const STORED_JSON_V1_TYPE: &'static str = "_dex_event_json_v1"; + + /// Decode legacy event JSON or a lossless storage-adapter envelope. + /// Durable readers pass the independently stored row kind. A malformed + /// envelope never falls back to the projection, and its original tool + /// argument digests are checked without rewriting any accepted value. + pub fn from_stored_json( + payload: &serde_json::Value, + expected_kind: Option<&str>, + ) -> Result { + let encoded = payload.get(Self::STORED_JSON_V1_KEY); + let event: Self = match encoded { + Some(exact) => { + if payload.get("type").and_then(serde_json::Value::as_str) + != Some(Self::STORED_JSON_V1_TYPE) + { + return Err(serde::de::Error::custom( + "encoded event has no storage type marker", + )); + } + let exact: String = serde_json::from_value(exact.clone())?; + serde_json::from_str(&exact)? + } + None => serde_json::from_value(payload.clone())?, + }; + if let Some(expected_kind) = expected_kind + && serde_json::to_value(&event)? + .get("type") + .and_then(serde_json::Value::as_str) + != Some(expected_kind) + { + return Err(serde::de::Error::custom( + "event type does not match its stored row kind", + )); + } + if let Self::ModelStepCompleted { calls, .. } = &event + && calls + .iter() + .any(|call| call.args_digest != args_digest(&call.args)) + { + return Err(serde::de::Error::custom( + "event tool arguments do not match their digest", + )); + } + Ok(event) + } +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn auto_approval_receipts_decode_losslessly_and_never_become_control_events() { + let event = Event::AutoApproved { + call: CallId::new("t1-1-0"), + approval: ApprovalId::new("ap-1"), + args_digest: args_digest(&serde_json::json!({"key": "w"})), + summary: "Send\0email".into(), + principal: PrincipalId::new(AUTO_APPROVER), + }; + let exact = serde_json::to_string(&event).expect("serialize receipt"); + let envelope = serde_json::json!({ + "type": Event::STORED_JSON_V1_TYPE, + "_dex_event_json_v1": exact, + "summary": "non-authoritative projection", + }); + assert_eq!( + Event::from_stored_json(&envelope, Some("auto_approved")).expect("exact receipt"), + event + ); + assert!(Event::from_stored_json(&envelope, Some("approval_decided")).is_err()); + assert!( + !event.is_control(), + "engine receipts must not advance the host control cursor" + ); + } + + #[test] + fn stored_event_decoder_refuses_corrupt_envelopes_and_call_digests() { + let event = Event::ModelStepCompleted { + step: 1, + text: String::new(), + calls: vec![ProposedCall::new( + CallId::new("turn-1-0"), + ToolName::new("dex.report_feedback"), + serde_json::json!({"diagnosis": "exact\0value"}), + PrincipalId::new("alice"), + )], + reasoning: None, + served: None, + timing: None, + }; + let mut envelope = serde_json::json!({ + "type": Event::STORED_JSON_V1_TYPE, + "_dex_event_json_v1": serde_json::to_string(&event).unwrap(), + "calls": [], + }); + assert_eq!( + Event::from_stored_json(&envelope, Some("model_step_completed")).unwrap(), + event + ); + assert!(Event::from_stored_json(&envelope, Some("tool_started")).is_err()); + envelope[Event::STORED_JSON_V1_KEY] = serde_json::json!("invalid JSON"); + assert!(Event::from_stored_json(&envelope, Some("model_step_completed")).is_err()); + let mut tampered = serde_json::to_value(&event).unwrap(); + tampered["calls"][0]["args"]["diagnosis"] = serde_json::json!("changed"); + assert!(Event::from_stored_json(&tampered, Some("model_step_completed")).is_err()); + } + #[test] fn events_round_trip_through_json() { let events = vec![ @@ -624,6 +779,7 @@ mod tests { input_tokens: 1, output_tokens: 2, cost_micros: 3, + ..Usage::default() }), Event::ModelStepCompleted { step: 1, @@ -636,6 +792,7 @@ mod tests { )], reasoning: None, served: None, + timing: None, }, Event::ModelStepCompleted { step: 2, @@ -650,6 +807,7 @@ mod tests { provider: "vertex-ai".into(), model: "gemini-3.6-flash".into(), }), + timing: None, }, Event::ToolFinished { call: CallId::new("t1-1-0"), @@ -826,4 +984,79 @@ mod tests { assert_ne!(a.args_digest, c.args_digest); assert_eq!(a.args_digest.len(), 64); } + + /// Rows written before `timing` and the cache token counts existed still + /// decode, to `None` and zero; new rows round-trip. + #[test] + fn old_rows_without_timing_or_cache_counts_decode() { + let old_step = serde_json::json!({ + "type": "model_step_completed", "step": 1, "text": "hi", "calls": [] + }); + let Event::ModelStepCompleted { timing, .. } = + serde_json::from_value::(old_step).expect("old step row") + else { + panic!("expected ModelStepCompleted"); + }; + assert_eq!(timing, None); + + let old_usage = serde_json::json!({ + "type": "usage", "input_tokens": 3, "output_tokens": 4, "cost_micros": 5 + }); + assert_eq!( + serde_json::from_value::(old_usage).expect("old usage row"), + Event::Usage(Usage { + input_tokens: 3, + output_tokens: 4, + cost_micros: 5, + ..Usage::default() + }) + ); + } + + #[test] + fn step_timing_and_cache_counts_round_trip_and_none_is_omitted() { + let step = Event::ModelStepCompleted { + step: 2, + text: String::new(), + calls: Vec::new(), + reasoning: None, + served: None, + timing: Some(StepTiming { + prepare_ms: 1, + mint_ms: 2, + headers_ms: 3, + first_event_ms: Some(4), + first_text_ms: None, + total_ms: 5, + }), + }; + let value = serde_json::to_value(&step).expect("encode"); + assert_eq!( + serde_json::from_value::(value).expect("decode"), + step + ); + let untimed = Event::ModelStepCompleted { + step: 2, + text: String::new(), + calls: Vec::new(), + reasoning: None, + served: None, + timing: None, + }; + let text = serde_json::to_string(&untimed).expect("encode"); + assert!(!text.contains("timing"), "{text}"); + + let usage = Event::Usage(Usage { + input_tokens: 10, + output_tokens: 2, + cost_micros: 1, + cache_read_input_tokens: 7, + cache_creation_input_tokens: 1, + }); + let value = serde_json::to_value(&usage).expect("encode"); + assert_eq!( + serde_json::from_value::(value).expect("decode"), + usage + ); + } } diff --git a/vendor/dex-loop/src/lib.rs b/vendor/dex-loop/src/lib.rs index 31a88302c..bf707b0e7 100644 --- a/vendor/dex-loop/src/lib.rs +++ b/vendor/dex-loop/src/lib.rs @@ -35,8 +35,8 @@ pub use engine::{CUT_OFF_NOTICE, DEFAULT_TOOL_CALL_DEADLINE, Engine, Exit, TOOLS pub use event::{ AUTO_APPROVER, ApprovalId, ApprovalMode, ArtifactRef, CallId, ClientToolSpec, Cursor, ErrorCode, Event, HEADLESS_AUTO_APPROVER, MessageId, Outcome, Output, OutputRef, PrincipalId, - ProposedCall, ProviderReasoning, ReceiptId, ServedBy, ThreadId, ToolName, ToolResult, TurnId, - Usage, args_digest, + ProposedCall, ProviderReasoning, ReceiptId, ServedBy, StepTiming, ThreadId, ToolName, + ToolResult, TurnId, Usage, args_digest, }; pub use ports::{ Claim, Effects, ExecutorKind, Fenced, GovernanceClass, Log, Model, ModelChunk, ModelError, diff --git a/vendor/dex-loop/src/ports.rs b/vendor/dex-loop/src/ports.rs index 18026232b..c9b4140b9 100644 --- a/vendor/dex-loop/src/ports.rs +++ b/vendor/dex-loop/src/ports.rs @@ -10,7 +10,7 @@ use tokio_util::sync::CancellationToken; use crate::context::Context; use crate::event::{ ApprovalId, CallId, Cursor, Event, PrincipalId, ProposedCall, ProviderReasoning, ServedBy, - ThreadId, ToolName, ToolResult, Usage, + StepTiming, ThreadId, ToolName, ToolResult, Usage, }; /// The log or the effect ledger refused a write. The engine stops at once and @@ -72,6 +72,9 @@ pub enum ModelChunk { /// once, before the first other chunk; the engine stores it on /// `ModelStepCompleted`. Served(ServedBy), + /// Where the step's time went. Sent at most once, after a clean + /// terminal; the engine stores it on `ModelStepCompleted`. + Timing(StepTiming), } /// The model call failed after the `Model` port's own retries. @@ -169,6 +172,22 @@ pub enum Verdict { approval: ApprovalId, summary: String, }, + /// Nothing runs. The call finishes with `preview` as its result (a + /// structured `needs_confirmation` document the model reads and turns into + /// one plain question for the user) and the turn goes on. No approval is + /// parked: the user's answer arrives as an ordinary message, and the model + /// calls again with a confirmation bound to the same arguments. + NeedsConfirmation { + preview: String, + }, + /// The user confirmed this exact call in chat. Granted at once; the engine + /// writes the receipt under the confirming user's principal + /// (`call.principal`), not `AUTO_APPROVER`. `summary` carries the + /// `confirmed_by_user` decision label. + Confirmed { + approval: ApprovalId, + summary: String, + }, } /// The tool registry, policy, and executors. diff --git a/vendor/dex-loop/tests/client_tool_replay.rs b/vendor/dex-loop/tests/client_tool_replay.rs index 8b8d96870..44c9542af 100644 --- a/vendor/dex-loop/tests/client_tool_replay.rs +++ b/vendor/dex-loop/tests/client_tool_replay.rs @@ -66,6 +66,7 @@ fn log_up_to_model_step(log: &FakeLog, call: &ProposedCall) { calls: vec![call.clone()], reasoning: None, served: None, + timing: None, }); } diff --git a/vendor/dex-loop/tests/prefetch.rs b/vendor/dex-loop/tests/prefetch.rs index d7dbca0d0..611731211 100644 --- a/vendor/dex-loop/tests/prefetch.rs +++ b/vendor/dex-loop/tests/prefetch.rs @@ -68,21 +68,26 @@ async fn reads_start_while_the_model_streams_and_the_step_adopts_their_results() Ok(Exit::Done) ); + // Guaranteed: both reads start in call order during the stream, and both + // finish after the commit. Which finish is recorded first is not: a read + // that returned before the commit is adopted in call order, one that is + // still running is recorded as it returns, and two 10 ms reads can + // return in either order on a loaded host. + let shapes = log.shapes_after(1); assert_eq!( - log.shapes_after(1), - strings(&[ - "step:1", - "started:t1-1-0", - "started:t1-1-1", - "delta:tail", - "completed:tail:[t1-1-0,t1-1-1]", - "finished:t1-1-0:ok", - "finished:t1-1-1:ok", - "step:2", - "delta:done", - "completed:done:[]", - "final:done", - ]) + shapes[..4], + strings(&["step:1", "started:t1-1-0", "started:t1-1-1", "delta:tail",]) + ); + assert_eq!(shapes[4], "completed:tail:[t1-1-0,t1-1-1]"); + let mut finished = shapes[5..7].to_vec(); + finished.sort(); + assert_eq!( + finished, + strings(&["finished:t1-1-0:ok", "finished:t1-1-1:ok"]) + ); + assert_eq!( + shapes[7..], + strings(&["step:2", "delta:done", "completed:done:[]", "final:done"]) ); // Adopted, not run again. let mut runs = tools.run_ids(); @@ -368,9 +373,21 @@ async fn a_new_turn_after_a_failed_stream_starts_its_own_calls_once() { for id in ["t1-1-0", "t2-1-0"] { assert_eq!(started_and_finished(&log, id), (1, 1), "{id}"); } - // The first turn's read ran (a read has no effect), but its result was - // never committed; the second turn's call ran once. - assert_eq!(tools.run_ids(), strings(&["t1-1-0", "t2-1-0"])); + // The first turn's read may or may not have been polled before the stream + // failed and dropped it (a read has no effect, and its `ToolStarted` and + // `ToolFinished` rows are on the log either way, checked above); it never + // ran twice. Its result was never committed. The second turn's call ran + // once. + let runs = tools.run_ids(); + assert!( + runs.iter().filter(|id| *id == "t1-1-0").count() <= 1, + "{runs:?}" + ); + assert_eq!( + runs.iter().filter(|id| *id == "t2-1-0").count(), + 1, + "{runs:?}" + ); assert!( history(&ctx).iter().all(|row| !row.contains("t1-1-0")), "{:?}", @@ -391,7 +408,16 @@ async fn an_interrupt_during_the_stream_ends_cleanly_with_a_read_in_flight() { let cancel = CancellationToken::new(); let host = async { - tokio::time::sleep(Duration::from_millis(100)).await; + // Interrupt once the read has started, however long a loaded host + // takes to get there; the model then hangs, so the stream is + // still open. + tokio::time::timeout(Duration::from_secs(30), async { + while started_and_finished(&log, "t1-1-0").0 == 0 { + tokio::time::sleep(Duration::from_millis(2)).await; + } + }) + .await + .expect("the read started"); log.host_append(Event::Interrupt { principal: alice() }); cancel.cancel(); }; @@ -413,6 +439,14 @@ async fn an_interrupt_during_the_stream_ends_cleanly_with_a_read_in_flight() { ); assert_eq!(started_and_finished(&log, "t1-1-0"), (1, 1)); assert_eq!(history(&ctx), strings(&["user:go", "assistant::[]"])); + // The log an interrupt racing the commit leaves replays to the same turn + // state. (The control watermark differs: the live engine learned of the + // interrupt from the cancel token, not by reading the control rows, and + // the turn is over either way.) + let replayed = log.rehydrate(); + assert_eq!(history(&replayed), history(&ctx)); + assert_eq!(replayed.cursor(), ctx.cursor()); + assert!(!replayed.turn_running()); } #[tokio::test] @@ -427,7 +461,9 @@ async fn a_stream_that_outlives_the_wall_budget_closes_its_reads() { &model, &tools, Budget { - wall: Duration::from_millis(150), + // Long enough that a loaded host still starts the read (the + // first chunk is 5 ms out) before the budget runs out. + wall: Duration::from_millis(600), ..budget() }, ); diff --git a/vendor/dex-loop/tests/read_wave_progress.rs b/vendor/dex-loop/tests/read_wave_progress.rs new file mode 100644 index 000000000..2563d8aba --- /dev/null +++ b/vendor/dex-loop/tests/read_wave_progress.rs @@ -0,0 +1,211 @@ +//! A read awaiting its own log write must keep progressing during peer results. +#[allow(dead_code)] +mod support; + +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use dex_loop::{ + Budget, CancellationToken, Context, Cursor, Engine, Event, Exit, Fenced, Lexicon, Log, + PrincipalId, ProposedCall, ThreadId, ToolName, ToolResult, ToolSpec, Tools, Verdict, +}; +use serde_json::json; +use support::*; +use tokio::sync::{Mutex, Notify}; + +#[derive(Default)] +struct Contention { + row: Mutex<()>, + step_committed: Notify, + read_holds_row: Notify, + finish_attempted: Notify, + runs: AtomicUsize, +} + +struct ContendedLog { + inner: FakeLog, + contention: Arc, +} + +impl Log for ContendedLog { + async fn append(&self, events: &[Event]) -> Result, Fenced> { + if events.iter().any( + |event| matches!(event, Event::ToolFinished { call, .. } if call.as_str() == "t1-1-0"), + ) { + self.contention.finish_attempted.notify_one(); + } + let row = self.contention.row.lock().await; + let result = self.inner.append(events).await; + drop(row); + if result.is_ok() + && events + .iter() + .any(|event| matches!(event, Event::ModelStepCompleted { step: 1, .. })) + { + self.contention.step_committed.notify_one(); + } + result + } + + async fn append_text(&self, text: String) -> Result<(), Fenced> { + self.inner.append_text(text).await + } + + async fn control_since(&self, after: Cursor) -> Result, Fenced> { + self.inner.control_since(after).await + } +} + +struct ContendedReads { + specs: Vec, + contention: Arc, +} + +impl Tools for ContendedReads { + fn catalog(&self) -> &[ToolSpec] { + &self.specs + } + + async fn search(&self, _principal: &PrincipalId, _query: &str) -> Vec { + Vec::new() + } + + async fn policy(&self, _ctx: &Context, _call: &ProposedCall) -> Verdict { + Verdict::Allow + } + + async fn run( + &self, + _thread: &ThreadId, + call: &ProposedCall, + _cancel: &CancellationToken, + ) -> ToolResult { + self.contention.runs.fetch_add(1, Ordering::SeqCst); + if call.args["key"] == "a" { + self.contention.read_holds_row.notified().await; + } else { + self.contention.step_committed.notified().await; + let _row = self.contention.row.lock().await; + self.contention.read_holds_row.notify_one(); + // Model a tool suspended while holding the durable log's row lock. + // Only polling this read again can release it for the peer result. + self.contention.finish_attempted.notified().await; + } + output_for(&call.id) + } +} + +#[tokio::test] +async fn peer_read_progresses_while_engine_records_a_finished_read() { + run_contended_wave(false).await; +} + +#[tokio::test] +async fn prefetched_peer_progresses_after_transfer_into_committed_wave() { + run_contended_wave(true).await; +} + +async fn run_contended_wave(prefetched: bool) { + let log = FakeLog::default(); + log.start_turn("t1", "look"); + if !prefetched { + log.host_append(Event::StepStarted { + step: 1, + control_through: Cursor::START, + }); + let calls = ["a", "b"] + .into_iter() + .enumerate() + .map(|(index, key)| { + ProposedCall::new( + call_id("t1", 1, index), + ToolName::new("search"), + json!({"key": key}), + alice(), + ) + }) + .collect(); + log.host_append(Event::ModelStepCompleted { + step: 1, + text: String::new(), + calls, + reasoning: None, + served: None, + timing: None, + }); + } + let contention = Arc::new(Contention::default()); + let model = if prefetched { + FakeModel::new(vec![ + vec![ + call("search", json!({"key": "a"})), + call("search", json!({"key": "b"})), + ], + vec![text("done")], + ]) + } else { + contention.step_committed.notify_one(); + FakeModel::new(vec![vec![text("done")]]) + }; + let engine = Engine::new( + ContendedLog { + inner: log.clone(), + contention: contention.clone(), + }, + model.clone(), + ContendedReads { + specs: vec![read_tool("search")], + contention: contention.clone(), + }, + FakeEffects::default(), + Lexicon::default(), + Budget { + max_steps: 10, + max_tokens: 1_000_000, + max_cost_micros: 1_000_000, + wall: Duration::from_secs(30), + }, + ); + let mut ctx = log.rehydrate(); + let cancel = CancellationToken::new(); + // A deadlock bound, not a timing benchmark: both dependencies are explicit. + assert_eq!( + tokio::time::timeout(Duration::from_secs(2), engine.run(&mut ctx, &cancel)) + .await + .expect("recording a peer result must keep polling the read holding its log lock"), + Ok(Exit::Done) + ); + assert_eq!(contention.runs.load(Ordering::SeqCst), 2); + let events = log.events(); + let committed = events + .iter() + .position(|event| matches!(event, Event::ModelStepCompleted { step: 1, .. })) + .unwrap(); + for index in 0..2 { + let started = events + .iter() + .position(|event| matches!(event, Event::ToolStarted { call, .. } if call == &call_id("t1", 1, index))) + .unwrap(); + assert_eq!(started < committed, prefetched); + } + for index in 0..2 { + assert_eq!( + log.events() + .iter() + .filter(|event| matches!(event, Event::ToolFinished { call, .. } if call == &call_id("t1", 1, index))) + .count(), + 1 + ); + } + assert_eq!( + view(&model.seen()[usize::from(prefetched)]), + strings(&[ + "user:look", + "assistant::[t1-1-0,t1-1-1]", + "tool:t1-1-0:ok:out/t1-1-0", + "tool:t1-1-1:ok:out/t1-1-1", + ]) + ); + assert_eq!(log.rehydrate(), ctx); +} diff --git a/vendor/dex-loop/tests/scenarios.rs b/vendor/dex-loop/tests/scenarios.rs index e70a45888..094a94bf1 100644 --- a/vendor/dex-loop/tests/scenarios.rs +++ b/vendor/dex-loop/tests/scenarios.rs @@ -373,6 +373,7 @@ async fn a_legacy_parked_call_is_granted_on_rehydrate_and_the_step_continues() { calls: vec![send.clone(), search], reasoning: None, served: None, + timing: None, }, Event::ApprovalRequested { call: send.id.clone(), @@ -1373,6 +1374,99 @@ async fn headless_turn_auto_approves_an_ask_gated_tool_and_records_the_audit_pai assert_eq!(rehydrated, ctx); } +/// A `NeedsConfirmation` verdict parks nothing and runs nothing: its preview +/// is the call's result, the turn goes on, and the model answers. +#[tokio::test] +async fn a_confirmation_verdict_returns_its_preview_without_running_or_parking() { + let log = FakeLog::default(); + let model = FakeModel::new(vec![ + vec![call("send_email", json!({"to": "bob"}))], + vec![text("Send this to bob?")], + ]); + let tools = FakeTools::new(vec![write_tool("send_email")]).verdict( + "send_email", + Verdict::NeedsConfirmation { + preview: "{\"status\":\"needs_confirmation\"}".into(), + }, + ); + let engine = engine(&log, &model, &tools, budget()); + let mut ctx = log.start_turn("t1", "email bob"); + + assert_eq!( + engine.run(&mut ctx, &CancellationToken::new()).await, + Ok(Exit::Done) + ); + assert!(tools.run_ids().is_empty(), "nothing ran"); + assert_eq!( + log.shapes_after(1), + strings(&[ + "step:1", + "completed::[t1-1-0]", + "finished:t1-1-0:err", + "step:2", + "delta:Send this to bob?", + "completed:Send this to bob?:[]", + "final:Send this to bob?", + ]) + ); + let events = log.events(); + assert!( + !events.iter().any(|event| matches!( + event, + Event::ApprovalRequested { .. } + | Event::ApprovalDecided { .. } + | Event::AutoApproved { .. } + )), + "no approval of any kind is written: {events:?}" + ); + assert!( + view(&model.seen()[1]) + .iter() + .any(|line| line.contains("tool:t1-1-0:err:{\"status\":\"needs_confirmation\"}")), + "the model reads the preview" + ); +} + +/// A `Confirmed` verdict runs the call once and the receipt names the user +/// who confirmed, under the call's digest. +#[tokio::test] +async fn a_confirmed_verdict_is_receipted_under_the_confirming_user() { + let log = FakeLog::default(); + let model = FakeModel::new(vec![ + vec![call("send_email", json!({"to": "bob"}))], + vec![text("sent")], + ]); + let tools = FakeTools::new(vec![write_tool("send_email")]).verdict( + "send_email", + Verdict::Confirmed { + approval: ApprovalId::new("ap-1"), + summary: "decision=confirmed_by_user; Send an email".into(), + }, + ); + let engine = engine(&log, &model, &tools, budget()); + let mut ctx = log.start_turn("t1", "yes, send it"); + + assert_eq!( + engine.run(&mut ctx, &CancellationToken::new()).await, + Ok(Exit::Done) + ); + assert_eq!(tools.run_ids(), strings(&["t1-1-0"])); + let call = call_id("t1", 1, 0); + let digest = dex_loop::args_digest(&tools.run_of(&call).args); + let events = log.events(); + assert!( + events.iter().any(|event| matches!( + event, + Event::AutoApproved { call: receipt_call, args_digest, principal, summary, .. } + if *receipt_call == call + && *args_digest == digest + && principal.as_str() == "alice" + && summary.contains("confirmed_by_user") + )), + "{events:?}" + ); +} + #[tokio::test] async fn headless_turn_keeps_a_hard_deny_denied() { let log = FakeLog::default(); @@ -1594,6 +1688,37 @@ async fn tools_search_exposes_schemas_for_the_next_step() { assert_eq!(log.rehydrate(), ctx); } +// An exposure only appends: search and core tools keep their place and the +// exposed tools follow in the order the search returned them, so the prefix +// the provider cached on the previous step is unchanged. +#[tokio::test] +async fn exposed_tools_are_appended_in_exposure_order() { + let log = FakeLog::default(); + let model = FakeModel::new(vec![ + vec![call("tools.search", json!({"query": "crm"}))], + vec![text("done")], + ]); + let tools = FakeTools::new(vec![ + read_tool("search"), + hidden_read_tool("crm.alpha"), + hidden_read_tool("crm.zed"), + ]) + .search_result("crm", &["crm.zed", "crm.alpha"]); + let engine = engine(&log, &model, &tools, budget()); + let mut ctx = log.start_turn("t1", "find acme in the crm"); + + assert_eq!( + engine.run(&mut ctx, &CancellationToken::new()).await, + Ok(Exit::Done) + ); + let offered = model.offered(); + assert_eq!(offered[0], strings(&["tools.search", "search"])); + assert_eq!( + offered[1], + strings(&["tools.search", "search", "crm.zed", "crm.alpha"]) + ); +} + // A question parks the turn; the answer is the call's result. #[tokio::test] async fn question_parks_and_answer_resumes() { diff --git a/vendor/dex-loop/tests/support/mod.rs b/vendor/dex-loop/tests/support/mod.rs index 64d68b421..61ff7d49b 100644 --- a/vendor/dex-loop/tests/support/mod.rs +++ b/vendor/dex-loop/tests/support/mod.rs @@ -258,6 +258,7 @@ pub fn usage( input_tokens, output_tokens, cost_micros, + ..Usage::default() })) } @@ -853,6 +854,7 @@ pub fn crashed_after_start(log: &FakeLog, call: &ProposedCall) { calls: vec![call.clone()], reasoning: None, served: None, + timing: None, }, Event::ToolStarted { call: call.id.clone(),