From c04062558d180c2b70422c352f7ec172a3b4fb51 Mon Sep 17 00:00:00 2001 From: wsp Date: Sun, 20 Sep 2026 14:23:56 +0800 Subject: [PATCH 1/2] fix(dev): Avoid Windows pnpm preparation races Desktop preparation could fail with missing batch-file errors after mobile-web inputs changed. Its workspace install ran alongside pnpm consumers and could rewrite the Windows batch shim they were using. Run mobile-web preparation before parallel resource consumers and invoke pnpm through Node without shell parsing on Windows. Add regression coverage for argument handling, exit failures, and preparation ordering. Validation: 17 script tests, syntax and diff checks, and pnpm run plugin-host:prepare passed. Full desktop startup was not run. --- scripts/dev-startup.test.mjs | 60 ++++++++++++++++++++++++++ scripts/dev.cjs | 63 +++++++++++++++++++--------- scripts/prepare-dsh-profile.test.mjs | 2 +- 3 files changed, 105 insertions(+), 20 deletions(-) diff --git a/scripts/dev-startup.test.mjs b/scripts/dev-startup.test.mjs index 16aa1081e4..45962bd226 100644 --- a/scripts/dev-startup.test.mjs +++ b/scripts/dev-startup.test.mjs @@ -5,6 +5,66 @@ import { join } from 'node:path'; import test from 'node:test'; import { resolveDevServerPorts } from './dev-server-ports.mjs'; import { prepareSherpaDev } from './prepare-sherpa-dev.mjs'; +import { createRequire } from 'node:module'; + +const require = createRequire(import.meta.url); +const { resolveCommandInvocation, runCommandPrefixed } = require('./dev.cjs'); + +test('desktop preparation invokes pnpm without the Windows batch shim', () => { + const invocation = resolveCommandInvocation( + 'pnpm', + ['run', 'plugin-host:prepare'], + { npm_execpath: 'E:\\repo\\node_modules\\pnpm\\bin\\pnpm.cjs' }, + 'win32', + ); + + assert.equal(invocation.cmd, process.execPath); + assert.deepEqual(invocation.args, [ + 'E:\\repo\\node_modules\\pnpm\\bin\\pnpm.cjs', + 'run', + 'plugin-host:prepare', + ]); + assert.equal(invocation.shell, false); +}); + +test('direct launcher supports spaces and shell metacharacters and preserves exit failures', async () => { + const root = mkdtempSync(join(tmpdir(), 'dev launcher & spaces ')); + const script = join(root, 'fake pnpm.cjs'); + const argument = 'literal & value %PATH% "quoted"'; + writeFileSync(script, ` + if (process.argv[2] !== ${JSON.stringify(argument)}) process.exit(9); + process.exit(Number(process.argv[3])); + `); + // Keep fixtures in the configured temporary root for post-test inspection. + const command = process.platform === 'win32' ? 'pnpm' : 'node'; + const args = process.platform === 'win32' ? [argument] : [script, argument]; + const env = { npm_execpath: script }; + const results = await Promise.all([0, 7].map(code => + runCommandPrefixed('fixture', command, [...args, String(code)], root, env), + )); + assert.equal(results[0].ok, true); + assert.equal(results[1].ok, false); + assert.equal(results[1].code, 7); +}); + +test('Windows direct launches without npm_execpath use the installed pnpm entry', () => { + const invocation = resolveCommandInvocation('pnpm', ['--version'], {}, 'win32'); + assert.match(invocation.args[0], /node_modules[\\/]pnpm[\\/]bin[\\/]pnpm\.cjs$/); + assert.equal(invocation.shell, false); + for (const platform of ['linux', 'darwin']) { + assert.deepEqual(resolveCommandInvocation('pnpm', ['--version'], {}, platform), { + cmd: 'pnpm', args: ['--version'], shell: false, + }); + } +}); + +test('workspace installation finishes before parallel dependency consumers start', () => { + const source = readFileSync(new URL('./dev.cjs', import.meta.url), 'utf8'); + const install = source.indexOf("await runCommandPrefixed('mobile-web'"); + const consumers = source.indexOf('const prepTasks = ['); + assert.ok(install > 0 && install < consumers); + assert.equal(source.match(/runCommandPrefixed\('mobile-web'/g).length, 1); +}); test('HTTP and HMR ports share one contract and reject conflicting overrides', () => { assert.deepEqual(resolveDevServerPorts({}), { port: 1422, hmrPort: 1421 }); diff --git a/scripts/dev.cjs b/scripts/dev.cjs index 46387f1c1f..6c18df23c3 100644 --- a/scripts/dev.cjs +++ b/scripts/dev.cjs @@ -144,6 +144,23 @@ function spawnCommand(cmd, args, cwd = ROOT_DIR, envOverrides = {}, shell = fals }); } +function resolveCommandInvocation(cmd, args, env = process.env, platform = process.platform) { + // Avoid the Windows batch shim, which workspace installs may rewrite while + // preparation is running. Keep paths and arguments out of shell parsing. + if (platform === 'win32' && cmd === 'pnpm') { + const entry = env.npm_execpath && /\.(?:c?js|mjs)$/i.test(env.npm_execpath) + ? env.npm_execpath + : path.join(ROOT_DIR, 'node_modules', 'pnpm', 'bin', 'pnpm.cjs'); + return { + cmd: process.execPath, + args: [entry, ...args], + shell: false, + }; + } + + return { cmd: cmd === 'node' ? process.execPath : cmd, args, shell: false }; +} + /** * Run a command asynchronously with piped output, prefixing every line so * parallel preparation steps stay distinguishable. Resolves (never rejects) @@ -151,15 +168,14 @@ function spawnCommand(cmd, args, cwd = ROOT_DIR, envOverrides = {}, shell = fals */ function runCommandPrefixed(prefix, cmd, args, cwd = ROOT_DIR, envOverrides = {}) { return new Promise((resolve) => { - const child = spawn(cmd, args, { + const env = { ...process.env, ...envOverrides }; + const invocation = resolveCommandInvocation(cmd, args, env); + const child = spawn(invocation.cmd, invocation.args, { cwd, - shell: process.platform === 'win32', + shell: invocation.shell, windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'], - env: { - ...process.env, - ...envOverrides, - }, + env, }); const forward = (stream, out) => { @@ -613,20 +629,29 @@ async function main() { const totalSteps = 2; let currentStep = 1; - // Step 1: Run all independent preparation tasks in parallel. - // copy-monaco / generate-version / mobile-web / flashgrep / plugin-host have no - // dependencies on each other; each task's output is line-prefixed so the - // interleaved logs stay attributable. The DeepSeek bridge is not prepared + // mobile-web may install workspace dependencies and rewrite node_modules/.bin. + // Finish it before launching consumers of those shims and dependencies. + // The remaining preparations can run in parallel with prefixed output. + // The DeepSeek bridge is not prepared // here: it is not a compile-time Tauri resource. Official desktop:build // compiles it; local DeepSeek sessions run `pnpm run prepare:dsh-profile`. printStep( currentStep++, totalSteps, desktopMode - ? 'Prepare resources (parallel: monaco, version, mobile-web, flashgrep, plugin-host)' + ? 'Prepare resources (mobile-web first; then parallel: monaco, version, flashgrep, plugin-host)' : 'Prepare resources (parallel: monaco, version)' ); + if (desktopMode) { + const result = await runCommandPrefixed('mobile-web', 'node', ['scripts/mobile-web-build.cjs', '--install']); + if (!result.ok) { + printError('Build mobile-web failed'); + if (result.error?.message) printError(result.error.message); + process.exit(1); + } + } + const prepTasks = [ { name: 'Copy Monaco resources', @@ -675,10 +700,6 @@ async function main() { } })(), }); - prepTasks.push({ - name: 'Build mobile-web', - promise: runCommandPrefixed('mobile-web', 'node', ['scripts/mobile-web-build.cjs', '--install']), - }); } const prepResults = await Promise.all(prepTasks.map((task) => task.promise)); @@ -795,7 +816,11 @@ async function main() { } } -main().catch((error) => { - printError('Startup failed: ' + error.message); - process.exit(1); -}); +if (require.main === module) { + main().catch((error) => { + printError('Startup failed: ' + error.message); + process.exit(1); + }); +} + +module.exports = { resolveCommandInvocation, runCommandPrefixed }; diff --git a/scripts/prepare-dsh-profile.test.mjs b/scripts/prepare-dsh-profile.test.mjs index d602c55516..8d33548401 100644 --- a/scripts/prepare-dsh-profile.test.mjs +++ b/scripts/prepare-dsh-profile.test.mjs @@ -155,7 +155,7 @@ test('desktop:dev does not compile the DeepSeek profile', () => { assert.doesNotMatch(devScript, /\['run', 'prepare:dsh-profile'\]/); assert.match( devScript, - /Prepare resources \(parallel: monaco, version, mobile-web, flashgrep, plugin-host\)/, + /Prepare resources \(mobile-web first; then parallel: monaco, version, flashgrep, plugin-host\)/, ); }); From 01f90bdcab50527bd093f8423f28f36ab4eaf32f Mon Sep 17 00:00:00 2001 From: wsp Date: Sun, 20 Sep 2026 16:54:00 +0800 Subject: [PATCH 2/2] fix(fork): preserve complete tool context Keep live assistant tool calls observable while ensuring fork and interrupted recovery contexts contain complete tool exchanges before provider requests. - Publish semantic assistant tool calls before tool execution - Normalize only the final active round at fork and recovery boundaries - Add informational synthetic results for tools still in progress - Keep synthetic results out of error reporting - Document the final-round performance and legacy-history trade-off - Cover fork normalization and interrupted recovery with focused tests --- src/apps/desktop/src/tray.rs | 1 - .../src/agentic/coordination/coordinator.rs | 15 +- .../src/agentic/execution/execution_engine.rs | 30 ++-- .../src/agentic/execution/round_executor.rs | 6 +- .../fork_agent/context_normalization.rs | 151 ++++++++++++++++++ .../core/src/agentic/fork_agent/mod.rs | 6 + .../core/src/agentic/session/context_store.rs | 15 ++ .../src/agentic/session/session_manager.rs | 75 +++++++++ .../implementations/miniapp_publish_tool.rs | 7 +- .../tools/implementations/skills/builtin.rs | 4 +- .../product-domains/src/canvas/runtime.rs | 4 +- .../relay-service/src/retired_version.rs | 30 +++- .../src/canvas/compiler/mod.rs | 4 +- .../src/plugin_source.rs | 16 +- .../src/remote_connect/relay_failure.rs | 73 +++++++-- 15 files changed, 380 insertions(+), 57 deletions(-) create mode 100644 src/crates/assembly/core/src/agentic/fork_agent/context_normalization.rs diff --git a/src/apps/desktop/src/tray.rs b/src/apps/desktop/src/tray.rs index 4a8926d3de..7d5c8f7df3 100644 --- a/src/apps/desktop/src/tray.rs +++ b/src/apps/desktop/src/tray.rs @@ -368,7 +368,6 @@ fn macos_tray_icon() -> Result, image::ImageError> )) } - pub fn show_main_window(app: &tauri::AppHandle) { if let Some(window) = app.get_webview_window("main") { // Restore before showing: Win+D can leave a visible window minimized. diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index 4519dee935..4f02af31d8 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs @@ -10747,7 +10747,20 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet parent_session_id )) })?; - let context_messages = self.load_session_context_messages(&parent_session).await?; + self.load_session_context_messages(&parent_session).await?; + // The restore path above may acquire the same lock, so take the + // snapshot lock only after restoration has completed. This makes the + // final context read atomic with round-level context publication. + let _mutation_guard = self + .session_manager + .acquire_session_mutation(parent_session_id) + .await?; + let context_messages = self + .session_manager + .get_context_messages(parent_session_id) + .await?; + let context_messages = + crate::agentic::fork_agent::normalize_fork_context_messages(context_messages); ForkAgentContextSnapshot::from_parent_session(&parent_session, context_messages) } diff --git a/src/crates/assembly/core/src/agentic/execution/execution_engine.rs b/src/crates/assembly/core/src/agentic/execution/execution_engine.rs index f583774dbd..ce36ebdaa2 100644 --- a/src/crates/assembly/core/src/agentic/execution/execution_engine.rs +++ b/src/crates/assembly/core/src/agentic/execution/execution_engine.rs @@ -3962,15 +3962,12 @@ impl ExecutionEngine { &round_result.assistant_message, ); + // Publish the assistant message and all tool results as one context + // mutation. A fork must never observe the assistant tool calls + // without their corresponding results. + let mut committed_round_messages = Vec::new(); if !round_result.assistant_message_committed { - // Update the in-memory message caches immediately so subsequent rounds see it. - if let Err(e) = self - .session_manager - .add_message(&context.session_id, round_result.assistant_message.clone()) - .await - { - warn!("Failed to update assistant message in memory: {}", e); - } + committed_round_messages.push(round_result.assistant_message.clone()); } // Add tool result messages to history @@ -3981,15 +3978,14 @@ impl ExecutionEngine { &context.dialog_turn_id, tool_result_msg, ); - - // Update the in-memory message caches immediately so subsequent rounds see it. - if let Err(e) = self - .session_manager - .add_message(&context.session_id, tool_result_msg.clone()) - .await - { - warn!("Failed to update tool result message in memory: {}", e); - } + committed_round_messages.push(tool_result_msg.clone()); + } + if let Err(e) = self + .session_manager + .add_messages(&context.session_id, committed_round_messages) + .await + { + warn!("Failed to update round messages in memory: {}", e); } #[cfg(feature = "agent-runtime")] diff --git a/src/crates/assembly/core/src/agentic/execution/round_executor.rs b/src/crates/assembly/core/src/agentic/execution/round_executor.rs index b516eb1965..45ccf8772c 100644 --- a/src/crates/assembly/core/src/agentic/execution/round_executor.rs +++ b/src/crates/assembly/core/src/agentic/execution/round_executor.rs @@ -1064,8 +1064,10 @@ impl RoundExecutor { .with_memory_citation(parsed_memory_citation) .with_model_response_replay(model_response_replay); - // Commit the complete provider response before any tool can block on IO, - // approval, or cancellation. Tool results remain separate semantic messages. + // Publish the semantic assistant response before tool execution so + // readers can observe the active tool call while it is running. Fork + // snapshots normalize this intentionally incomplete exchange before + // sending it to a provider. let assistant_message_committed = if let Some(session_manager) = session_manager { session_manager .add_message(&context.session_id, assistant_message.clone()) diff --git a/src/crates/assembly/core/src/agentic/fork_agent/context_normalization.rs b/src/crates/assembly/core/src/agentic/fork_agent/context_normalization.rs new file mode 100644 index 0000000000..334188fbec --- /dev/null +++ b/src/crates/assembly/core/src/agentic/fork_agent/context_normalization.rs @@ -0,0 +1,151 @@ +use crate::agentic::core::{Message, MessageContent, ToolResult}; + +/// Ensure a fork never carries an assistant tool call without a matching +/// result. A fork may be requested while the parent is between model output +/// and tool execution; providers commonly reject that history shape outright. +pub(crate) fn normalize_fork_context_messages(mut messages: Vec) -> Vec { + normalize_incomplete_tool_calls( + &mut messages, + "Tool execution was still in progress when forking context; no result was available.", + ); + messages +} + +/// Close assistant tool calls that are still in flight at a provider-history +/// boundary. Live session history intentionally publishes those calls before +/// execution so readers can observe them; snapshots used for a new model +/// request must contain a result for every call. +/// +/// The runtime appends each completed model round as one closed group, and an +/// interruption/fork can only leave the final active round incomplete. We use +/// that invariant to inspect only the final round (`O(k)` for the final round, +/// instead of `O(M)` for the full context). The trade-off is intentional: +/// malformed older history with an incomplete *earlier* round is not repaired +/// by this fast path. Snapshots without round metadata are treated as legacy +/// one-round contexts and use the same logic over the whole snapshot. +pub(crate) fn normalize_incomplete_tool_calls(messages: &mut Vec, result_text: &str) { + let Some(last_round_id) = messages + .iter() + .rev() + .find_map(|message| message.metadata.round_id.as_deref()) + .map(str::to_string) + else { + normalize_round_messages(messages, result_text, None); + return; + }; + + let last_index = messages + .iter() + .rposition(|message| message.metadata.round_id.as_deref() == Some(last_round_id.as_str())) + .unwrap_or(0); + let mut start = last_index; + while start > 0 + && messages[start - 1].metadata.round_id.as_deref() == Some(last_round_id.as_str()) + { + start -= 1; + } + let mut round_messages = messages.split_off(start); + normalize_round_messages(&mut round_messages, result_text, Some(&last_round_id)); + messages.extend(round_messages); +} + +fn normalize_round_messages( + messages: &mut Vec, + result_text: &str, + round_id: Option<&str>, +) { + let result_ids = messages + .iter() + .filter_map(|message| match &message.content { + MessageContent::ToolResult { tool_id, .. } => Some(tool_id.clone()), + _ => None, + }) + .collect::>(); + let mut normalized = Vec::with_capacity(messages.len()); + for message in messages.drain(..) { + let missing = match &message.content { + MessageContent::Mixed { tool_calls, .. } => tool_calls + .iter() + .filter(|call| !result_ids.contains(&call.tool_id)) + .map(|call| ToolResult { + tool_id: call.tool_id.clone(), + tool_name: call.tool_name.clone(), + effective_tool_name: None, + result: serde_json::json!(result_text), + result_for_assistant: Some(result_text.to_string()), + is_error: false, + duration_ms: None, + image_attachments: None, + }) + .collect::>(), + _ => Vec::new(), + }; + normalized.push(message); + normalized.extend(missing.into_iter().map(|result| { + let message = Message::tool_result(result); + match round_id { + Some(round_id) => message.with_round_id(round_id.to_string()), + None => message, + } + })); + } + *messages = normalized; +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::agentic::core::{Message, ToolCall, ToolResult}; + + #[test] + fn fork_context_inserts_in_progress_results_for_unclosed_tool_calls() { + let call = ToolCall { + tool_id: "call-1".to_string(), + tool_name: "Read".to_string(), + arguments: serde_json::json!({"path": "a.txt"}), + ..Default::default() + }; + let messages = normalize_fork_context_messages(vec![Message::assistant_with_tools( + String::new(), + vec![call], + )]); + + assert_eq!(messages.len(), 2); + assert!(matches!( + messages[1].content, + MessageContent::ToolResult { + ref tool_id, + is_error: false, + ref result, + .. + } if tool_id == "call-1" + && result == &serde_json::json!("Tool execution was still in progress when forking context; no result was available.") + )); + } + + #[test] + fn fork_context_does_not_duplicate_existing_tool_results() { + let call = ToolCall { + tool_id: "call-1".to_string(), + tool_name: "Read".to_string(), + arguments: serde_json::json!({}), + ..Default::default() + }; + let result = ToolResult { + tool_id: "call-1".to_string(), + tool_name: "Read".to_string(), + effective_tool_name: None, + result: serde_json::json!("ok"), + result_for_assistant: None, + is_error: false, + duration_ms: None, + image_attachments: None, + }; + let messages = normalize_fork_context_messages(vec![ + Message::assistant_with_tools(String::new(), vec![call]), + Message::tool_result(result), + ]); + + assert_eq!(messages.len(), 2); + } +} diff --git a/src/crates/assembly/core/src/agentic/fork_agent/mod.rs b/src/crates/assembly/core/src/agentic/fork_agent/mod.rs index c74bb24dd7..06693817e7 100644 --- a/src/crates/assembly/core/src/agentic/fork_agent/mod.rs +++ b/src/crates/assembly/core/src/agentic/fork_agent/mod.rs @@ -7,6 +7,12 @@ use crate::agentic::core::{Message, Session, SessionConfig}; use crate::util::errors::{OpenBitFunError, OpenBitFunResult}; +mod context_normalization; + +pub(crate) use context_normalization::{ + normalize_fork_context_messages, normalize_incomplete_tool_calls, +}; + /// Immutable snapshot of a parent session's runtime context at fork time. #[derive(Debug, Clone)] pub struct ForkAgentContextSnapshot { diff --git a/src/crates/assembly/core/src/agentic/session/context_store.rs b/src/crates/assembly/core/src/agentic/session/context_store.rs index 8425b98f5b..3a1c6b9e95 100644 --- a/src/crates/assembly/core/src/agentic/session/context_store.rs +++ b/src/crates/assembly/core/src/agentic/session/context_store.rs @@ -89,6 +89,21 @@ impl SessionContextStore { } } + /// Append a logically complete group of context messages while holding the + /// per-session entry lock. Fork snapshots can therefore observe either + /// the whole group or none of it. + pub fn add_messages(&self, session_id: &str, messages: Vec) { + if messages.is_empty() { + return; + } + if let Some(mut cached_messages) = self.session_contexts.get_mut(session_id) { + cached_messages.extend(messages); + } else { + self.session_contexts + .insert(session_id.to_string(), messages); + } + } + pub fn replace_context(&self, session_id: &str, messages: Vec) { self.session_contexts .insert(session_id.to_string(), messages); diff --git a/src/crates/assembly/core/src/agentic/session/session_manager.rs b/src/crates/assembly/core/src/agentic/session/session_manager.rs index 25d083d527..9fb4d7df91 100644 --- a/src/crates/assembly/core/src/agentic/session/session_manager.rs +++ b/src/crates/assembly/core/src/agentic/session/session_manager.rs @@ -9,6 +9,7 @@ use crate::agentic::core::{ SessionAgentRouteOwner, SessionConfig, SessionKind, SessionModelBindingPolicy, SessionState, SessionSummary, TurnStats, }; +use crate::agentic::fork_agent::normalize_incomplete_tool_calls; use crate::agentic::image_analysis::ImageContextData; use crate::agentic::keyed_lock::{KeyedAsyncLock, KeyedAsyncLockGuard}; use crate::agentic::memories::db::{MemoryDatabase, MEMORY_PHASE2_GLOBAL_JOB_KEY}; @@ -8635,6 +8636,10 @@ impl SessionManager { "Context snapshot is unavailable for interrupted turn: {turn_id}" )) })?; + normalize_incomplete_tool_calls( + &mut messages, + "Tool execution was still in progress while resuming context; no result was available.", + ); messages.retain(|message| { message.internal_reminder_kind() != Some(InternalReminderKind::InterruptedContinue) }); @@ -9285,6 +9290,47 @@ impl SessionManager { Ok(()) } + /// Append a complete model round atomically from the context store's point + /// of view. In particular, an assistant tool-call message and its tool + /// results must not be observable by fork/snapshot readers separately. + pub async fn add_messages( + &self, + session_id: &str, + messages: Vec, + ) -> OpenBitFunResult<()> { + if messages.is_empty() { + return Ok(()); + } + let _mutation_guard = self.acquire_session_mutation(session_id).await?; + for message in &messages { + if let Some(citation) = message.metadata.memory_citation.as_ref() { + if let Err(error) = self + .memory_database + .record_memory_citation( + session_id, + message.metadata.turn_id.as_deref(), + message.metadata.round_id.as_deref(), + &message.id, + citation, + ) + .await + { + warn!( + "Failed to record memory citation: session_id={}, message_id={}, error={}", + session_id, message.id, error + ); + } + } + } + self.context_store.add_messages(session_id, messages); + self.persist_current_turn_context_snapshot_best_effort( + session_id, + "context_messages_added", + ) + .await; + Ok(()) + } + /// Replace the runtime context cache for a session and immediately refresh the current turn /// snapshot. This is primarily used after compression rewrites the model-visible context. pub async fn replace_context_messages(&self, session_id: &str, messages: Vec) { @@ -10509,6 +10555,23 @@ mod tests { ) .await .expect("safe boundary should persist"); + manager + .add_message( + &session.session_id, + Message::assistant_with_tools( + String::new(), + vec![ToolCall { + tool_id: "in-flight-recovery-call".to_string(), + tool_name: "Read".to_string(), + arguments: json!({"path": "still-running.txt"}), + ..ToolCall::default() + }], + ) + .with_turn_id(turn_id.clone()) + .with_round_id("round-1".to_string()), + ) + .await + .expect("in-flight tool call should persist"); let interrupted = manager .mark_dialog_turn_interrupted(&session.session_id, &turn_id) @@ -10555,6 +10618,18 @@ mod tests { .to_string() .contains("previous work was interrupted") })); + assert!(plan.messages.iter().any(|message| { + matches!( + &message.content, + MessageContent::ToolResult { + tool_id, + is_error: false, + result, + .. + } if tool_id == "in-flight-recovery-call" + && result == &json!("Tool execution was still in progress while resuming context; no result was available.") + ) + })); let duplicate = reopen_interrupted_turn_for_test(&manager, &session.session_id, &plan.turn_id, 0) diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/miniapp_publish_tool.rs b/src/crates/assembly/core/src/agentic/tools/implementations/miniapp_publish_tool.rs index fa84fc87dd..ed491a9e01 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/miniapp_publish_tool.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/miniapp_publish_tool.rs @@ -318,10 +318,9 @@ Publishing is an outward-facing action: only call this when the user explicitly .map(str::to_string) .unwrap_or_else(|| suggest_market_slug(&app.name, &app.id)); - let submissions = client - .list_submissions() - .await - .map_err(|e| OpenBitFunError::tool(format!("Could not load submission history: {e}")))?; + let submissions = client.list_submissions().await.map_err(|e| { + OpenBitFunError::tool(format!("Could not load submission history: {e}")) + })?; let release_target = match resolve_release_target(&submissions, &slug) { ReleaseTarget::NewListing if me.is_admin => match client.listing(&slug).await { Ok(listing) => ReleaseTarget::ExistingListing { diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/skills/builtin.rs b/src/crates/assembly/core/src/agentic/tools/implementations/skills/builtin.rs index 8c712c7bd0..106e910f2e 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/skills/builtin.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/skills/builtin.rs @@ -120,7 +120,9 @@ async fn write_installed_manifest(root: &Path) -> OpenBitFunResult<()> { Ok(()) } -async fn acquire_install_lock(user_skills_root: &Path) -> OpenBitFunResult { +async fn acquire_install_lock( + user_skills_root: &Path, +) -> OpenBitFunResult { let lock_path = builtin_skills_install_lock_path(user_skills_root); // Use an OS-backed advisory file lock so parallel test processes and app diff --git a/src/crates/contracts/product-domains/src/canvas/runtime.rs b/src/crates/contracts/product-domains/src/canvas/runtime.rs index bd984fba13..20372ad645 100644 --- a/src/crates/contracts/product-domains/src/canvas/runtime.rs +++ b/src/crates/contracts/product-domains/src/canvas/runtime.rs @@ -1,4 +1,6 @@ -use super::sdk_contract_generated::{OPENBITFUN_CANVAS_RUNTIME_VERSION, OPENBITFUN_CANVAS_SDK_VERSION}; +use super::sdk_contract_generated::{ + OPENBITFUN_CANVAS_RUNTIME_VERSION, OPENBITFUN_CANVAS_SDK_VERSION, +}; use crate::canvas::types::{CanvasCompiledPayload, CanvasDiagnostic, CanvasId, CanvasRevision}; use serde::{Deserialize, Serialize}; diff --git a/src/crates/services/relay-service/src/retired_version.rs b/src/crates/services/relay-service/src/retired_version.rs index 8200880ae4..67b38f8796 100644 --- a/src/crates/services/relay-service/src/retired_version.rs +++ b/src/crates/services/relay-service/src/retired_version.rs @@ -172,19 +172,33 @@ mod tests { // Retirement can only come from explicit operator configuration; a // relay started without it must never answer the retirement catch-all, // whatever prefix the edge proxy announces. - assert!(!is_retired_request("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/api/devices", &headers(Some("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/v/1.0.1")))); + assert!(!is_retired_request( + "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/api/devices", + &headers(Some("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/v/1.0.1")) + )); assert!(!is_retired_request("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/v1/updates", &headers(None))); } #[test] fn a_retired_deployment_answers_every_managed_route() { let policy = RetirementPolicy::retired_deployment(); - for path in ["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/api/devices", "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/api/info", "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/v1/updates", "/v3/sessions/x", "/p"] { + for path in [ + "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/api/devices", + "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/api/info", + "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/v1/updates", + "/v3/sessions/x", + "/p", + ] { assert!(policy.is_retired(path, &headers(None)), "{path}"); } // The health probe and the page that explains the update keep working. assert!(!policy.is_retired("/health", &headers(None))); - for path in ["/", "/index.html", "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/assets/index-abc.js", "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/brand/icon.png"] { + for path in [ + "/", + "/index.html", + "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/assets/index-abc.js", + "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/brand/icon.png", + ] { assert!(!policy.is_retired(path, &headers(None)), "{path}"); } } @@ -203,10 +217,16 @@ mod tests { #[test] fn retirement_switches_read_truthy_values_and_ignore_noise() { for value in ["1", "true", "TRUE", " yes ", "on"] { - assert!(RetirementPolicy::from_values(Some(value), None).retired, "{value}"); + assert!( + RetirementPolicy::from_values(Some(value), None).retired, + "{value}" + ); } for value in ["0", "false", "off", ""] { - assert!(!RetirementPolicy::from_values(Some(value), None).retired, "{value}"); + assert!( + !RetirementPolicy::from_values(Some(value), None).retired, + "{value}" + ); } let malformed = RetirementPolicy::from_values(None, Some(" , ,")); assert!(malformed.prefixes.is_empty()); diff --git a/src/crates/services/services-integrations/src/canvas/compiler/mod.rs b/src/crates/services/services-integrations/src/canvas/compiler/mod.rs index a29796a304..31ea752fd3 100644 --- a/src/crates/services/services-integrations/src/canvas/compiler/mod.rs +++ b/src/crates/services/services-integrations/src/canvas/compiler/mod.rs @@ -22,7 +22,9 @@ use openbitfun_product_domains::canvas::types::{ CanvasCompiledPayload, CanvasDiagnostic, CanvasDiagnosticCategory, CanvasDiagnosticSeverity, CanvasSource, }; -use openbitfun_product_domains::canvas::{OPENBITFUN_CANVAS_RUNTIME_VERSION, OPENBITFUN_CANVAS_SDK_VERSION}; +use openbitfun_product_domains::canvas::{ + OPENBITFUN_CANVAS_RUNTIME_VERSION, OPENBITFUN_CANVAS_SDK_VERSION, +}; pub use html::compile_canvas_html; use html::stable_content_hash; diff --git a/src/crates/services/services-integrations/src/plugin_source.rs b/src/crates/services/services-integrations/src/plugin_source.rs index 107485c1ea..96bbcafb0e 100644 --- a/src/crates/services/services-integrations/src/plugin_source.rs +++ b/src/crates/services/services-integrations/src/plugin_source.rs @@ -1657,10 +1657,9 @@ impl ProductPluginSourceStore { .and_then(|version| u16::try_from(version).ok()) { if schema_version != PLUGIN_TRUST_STORE_SCHEMA_VERSION { - return Err(PluginSourceContractError::UnsupportedTrustStoreSchema( - schema_version, - ) - .into()); + return Err( + PluginSourceContractError::UnsupportedTrustStoreSchema(schema_version).into(), + ); } } let store = serde_json::from_value::(persisted).map_err(|source| { @@ -3540,12 +3539,9 @@ mod tests { .await .expect("create trust state directory"); let original = br#"{"schemaVersion":1,"epoch":7,"records":[]}"#; - tokio::fs::write( - &fixture.trust_path, - original, - ) - .await - .expect("write pre-OpenBitFun trust store"); + tokio::fs::write(&fixture.trust_path, original) + .await + .expect("write pre-OpenBitFun trust store"); let snapshot = fixture.service().refresh(&fixture.workspace).await; let persisted = tokio::fs::read(&fixture.trust_path) diff --git a/src/crates/services/services-integrations/src/remote_connect/relay_failure.rs b/src/crates/services/services-integrations/src/remote_connect/relay_failure.rs index 16c3561b59..a6bd9b6909 100644 --- a/src/crates/services/services-integrations/src/remote_connect/relay_failure.rs +++ b/src/crates/services/services-integrations/src/remote_connect/relay_failure.rs @@ -65,8 +65,15 @@ const OUTDATED_MARKERS: &[&str] = &[ ]; /// Substrings that mean the account session must be re-established. -const AUTH_MARKERS: &[&str] = &["sign in", "unauthorized", "invalid or expired token", - "expired token", "relay auth error", "http 401", "http 403"]; +const AUTH_MARKERS: &[&str] = &[ + "sign in", + "unauthorized", + "invalid or expired token", + "expired token", + "relay auth error", + "http 401", + "http 403", +]; /// Transport failures with no HTTP status, as reported by reqwest and the /// socket stacks. @@ -136,7 +143,11 @@ fn http_status(lower: &str) -> Option { let mut search = lower; while let Some(index) = search.find("http ") { let rest = &search[index + 5..]; - let digits: String = rest.chars().take_while(char::is_ascii_digit).take(3).collect(); + let digits: String = rest + .chars() + .take_while(char::is_ascii_digit) + .take(3) + .collect(); if digits.len() == 3 { if let Ok(status) = digits.parse() { return Some(status); @@ -154,12 +165,18 @@ mod tests { #[test] fn a_retired_version_is_an_update_not_a_network_fault() { - assert_eq!(classify_relay_failure("List devices failed: HTTP 410"), RelayFailureKind::RelayVersionRetired); + assert_eq!( + classify_relay_failure("List devices failed: HTTP 410"), + RelayFailureKind::RelayVersionRetired + ); assert_eq!( classify_relay_failure("relay_version_retired: This Relay version has been retired."), RelayFailureKind::ClientOutdated ); - assert_eq!(RelayFailureKind::RelayVersionRetired.action(), RelayFailureAction::CheckUpdates); + assert_eq!( + RelayFailureKind::RelayVersionRetired.action(), + RelayFailureAction::CheckUpdates + ); assert!(!RelayFailureKind::RelayVersionRetired.is_retryable()); } @@ -178,32 +195,60 @@ mod tests { fn a_temporary_relay_fault_is_retryable() { for status in [408, 425, 429, 500, 502, 503, 504] { let message = format!("List devices failed: HTTP {status}"); - assert_eq!(classify_relay_failure(&message), RelayFailureKind::RelayUnavailable, "{status}"); + assert_eq!( + classify_relay_failure(&message), + RelayFailureKind::RelayUnavailable, + "{status}" + ); assert!(RelayFailureKind::RelayUnavailable.is_retryable()); } } #[test] fn authentication_failures_stay_on_the_sign_in_path() { - assert_eq!(classify_relay_failure("List devices failed: HTTP 401"), RelayFailureKind::Auth); - assert_eq!(classify_relay_failure("Invalid or expired token"), RelayFailureKind::Auth); - assert_eq!(classify_relay_failure("Sign in to continue"), RelayFailureKind::Auth); + assert_eq!( + classify_relay_failure("List devices failed: HTTP 401"), + RelayFailureKind::Auth + ); + assert_eq!( + classify_relay_failure("Invalid or expired token"), + RelayFailureKind::Auth + ); + assert_eq!( + classify_relay_failure("Sign in to continue"), + RelayFailureKind::Auth + ); assert_eq!(RelayFailureKind::Auth.action(), RelayFailureAction::SignIn); } #[test] fn transport_failures_are_network_and_unknown_stays_retryable() { - assert_eq!(classify_relay_failure("error sending request: connection refused"), RelayFailureKind::Network); - assert_eq!(classify_relay_failure("Relay connection failed: tls handshake failed"), RelayFailureKind::Network); - assert_eq!(classify_relay_failure("List devices failed: HTTP 404"), RelayFailureKind::Unknown); - assert_eq!(classify_relay_failure("").action(), RelayFailureAction::Retry); + assert_eq!( + classify_relay_failure("error sending request: connection refused"), + RelayFailureKind::Network + ); + assert_eq!( + classify_relay_failure("Relay connection failed: tls handshake failed"), + RelayFailureKind::Network + ); + assert_eq!( + classify_relay_failure("List devices failed: HTTP 404"), + RelayFailureKind::Unknown + ); + assert_eq!( + classify_relay_failure("").action(), + RelayFailureAction::Retry + ); } #[test] fn the_configured_endpoint_of_another_official_release_counts_as_retired() { assert_eq!(classify_relay_endpoint(DEFAULT_RELAY_URL), None); let older = DEFAULT_RELAY_URL.replace("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/v/1.0.2", "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/v/1.0.1"); - assert_eq!(classify_relay_endpoint(&older), Some(RelayFailureKind::RelayVersionRetired)); + assert_eq!( + classify_relay_endpoint(&older), + Some(RelayFailureKind::RelayVersionRetired) + ); assert_eq!(classify_relay_endpoint("https://example.com/relay"), None); } }