let (event_tx, event_rx) = mpsc::unbounded_channel(); // 事件通道：V8 线程 → actor，Started/ToolCall/Result 都从这里来let (command_tx, command_rx) = mpsc::unbounded_channel(); // 命令通道：actor → V8 线程，工具结果回填、终止指令走这里let (initial_response_tx, initial_response_rx) = oneshot::channel(); // 首个观察结果专用一次性通道，execute() 的调用方直接 await 它let (runtime_tx, runtime_control_tx, runtime_terminate_handle) = spawn_runtime( // 起 V8 线程，拿回三样东西：命令口、控制口、可强杀的执行句柄    stored_values, // 会话级 store/load 共享值快照——跨 cell 的"全局变量"    runtime_request(request), // JS 源码 + 启用工具清单打包成 ExecuteRequest    event_tx, // 事件出口接上面那条 unbounded channel    PendingRuntimeMode::PauseUntilResumed, // 关键策略：JS 一 await 就冻结线程，等 actor 显式 Resume——不为空转烧 CPU    task_failure_handler.clone(), // panic/任务失败上报钩子)?; // V8 初始化失败直接返回 Err，cell 根本不会创建let handle = CellHandle::new(command_tx, Arc::clone(&cell_state)); // 对外句柄：observe()/terminate() 都通过它和 actor 对话let task = run_cell( // actor 主循环本体，作为 tokio 任务跑在异步运行时上    host, // CellHost：会话回调（invoke_tool/notify/commit_completion）    CellContext { runtime_tx, runtime_control_tx, runtime_terminate_handle, cell_state }, // 对 V8 线程的全部控制面 + 共享状态    event_rx, command_rx, // actor 的两个输入口    Observer { mode: initial_observe_mode, response_tx: initial_response_tx }, // 首个等待者：execute() 的调用方    task_failure_handler, // 失败上报钩子透传);

ensure_v8_initialized()?; // V8 进程级初始化只做一次（平台/沙箱配置），失败直接拒绝创建 celllet (command_tx, command_rx) = std_mpsc::channel(); // 注意是 std 同步通道：V8 线程是裸 OS 线程，不能跑 tokiolet (control_tx, control_rx) = std_mpsc::channel(); // 控制口单独一条：Continue/Resume/Terminate 与数据命令分流let runtime_command_tx = command_tx.clone(); // 再克隆一份送进线程——JS 侧 setTimeout 到期也要走这条命令通道let (isolate_handle_tx, isolate_handle_rx) = std_mpsc::sync_channel(1); // 容量 1 的同步通道：等 V8 isolate 建好才放行let enabled_tools = request.enabled_tools.iter().map(enabled_tool_metadata).collect::<Vec<_>>(); // 工具定义转成运行时元数据（名字/描述/kind）let config = RuntimeConfig { tool_call_id: request.tool_call_id, enabled_tools, source: request.source, stored_values }; // 打包全部启动参数，move 进线程闭包spawn_supervised_runtime_thread(event_tx.clone(), task_failure_handler, move || { run_runtime(config, event_tx, command_rx, control_rx, pending_mode, isolate_handle_tx, runtime_command_tx); }); // catch_unwind 包住整个线程体：panic 也能上报而不是静默死掉let isolate_handle = isolate_handle_rx.recv().map_err(|_| "failed to initialize code mode runtime".to_string())?; // 阻塞等 isolate 就绪——调用方拿到句柄即可随时 terminate_execution() 强杀 CPU 死循环

fn next_runtime_command( // V8 线程的"取下一条命令"原语——暂停-恢复协议的核心    event_tx: &mpsc::UnboundedSender<RuntimeEvent>, // 事件出口：要告诉 actor "我卡住了"    command_rx: &std_mpsc::Receiver<RuntimeCommand>, // 数据命令口（工具结果/超时）    control_rx: &std_mpsc::Receiver<RuntimeControlCommand>, // 控制口（Continue/Resume/Terminate）    pending_mode: PendingRuntimeMode, // 卡住时是"冻结等恢复"还是"测试模式直接阻塞") -> Option<RuntimeCommand> { // None = actor 已断开，线程该退出了    loop {        match command_rx.try_recv() { // 非阻塞取命令：有活就干，没有才考虑挂起            Ok(command) => return Some(command), // 拿到工具结果/超时事件 → 回主循环执行            Err(std_mpsc::TryRecvError::Disconnected) => return None, // actor drop 了发送端 = cell 被销毁，线程退出            Err(std_mpsc::TryRecvError::Empty) => {} // 暂无命令：走下面的挂起流程        }        let _ = event_tx.send(RuntimeEvent::Pending); // 先广播"我卡在 await 了"——actor 据此暂停计时/交付 Pending 事件        match pending_mode {            #[cfg(test)] Continue => return command_rx.recv().ok(), // 测试专用：不冻结，直接阻塞等下一条命令            PendingRuntimeMode::PauseUntilResumed => match control_rx.recv().ok()? { // 生产路径：线程在此冻结，零 CPU 消耗                RuntimeControlCommand::Continue => return command_rx.recv().ok(), // actor 回填了工具结果 → 取回命令继续跑                RuntimeControlCommand::Resume => continue, // 观察窗口结束 → 回循环顶部重新 try_recv                RuntimeControlCommand::Terminate => return Some(RuntimeCommand::Terminate), // 终止指令优先级最高，直接上抛            },        }    }}

pub(super) fn install_globals(scope: &mut v8::PinScope<'_, '_>) -> Result<(), String> { // 给 cell 装 API：先删危险全局，再挂白名单函数    let global = scope.get_current_context().global(scope); // 拿到全局对象，接下来在上面增删属性    delete_global(scope, global, "console")?; // 删 console：cell 里不该有浏览器式调试输出    delete_global(scope, global, "Atomics")?; // 删 Atomics：跨线程同步原语对单 cell 无意义，还放大攻击面    delete_global(scope, global, "SharedArrayBuffer")?; // 删共享内存：同上，且是 Spectre 类漏洞的常见入口    delete_global(scope, global, "WebAssembly")?; // 禁 WASM：模型代码只能跑纯 JS，不能加载二进制模块    let tools = build_tools_object(scope)?; // 按 enabled_tools 逐个生成函数挂到 tools.* 下——没启用的工具根本不存在    let all_tools = build_all_tools_value(scope)?; // [{name, description}] 元数据数组，供 JS 代码自省"我能调什么"    let clear_timeout = helper_function(scope, "clearTimeout", clear_timeout_callback)?; // clearTimeout：取消 JS 侧定时器    let set_timeout = helper_function(scope, "setTimeout", set_timeout_callback)?; // setTimeout：到期经命令通道唤醒 V8 线程（不是真异步）    let text = helper_function(scope, "text", text_callback)?; // text()：产出文本内容项（模型可见的输出）    let image = helper_function(scope, "image", image_callback)?; // image()：产出图片内容项    let audio = helper_function(scope, "audio", audio_callback)?; // audio()：产出音频内容项    let generated_image = helper_function(scope, "generatedImage", generated_image_callback)?; // generatedImage()：带 output_hint 的生成图结果    let store = helper_function(scope, "store", store_callback)?; // store(key, value)：写入会话级共享值，跨 cell 存活    let load = helper_function(scope, "load", load_callback)?; // load(key)：读回之前存的值（JSON 往返）    let notify = helper_function(scope, "notify", notify_callback)?; // notify(text)：向会话注入一条消息，如"还在跑，已完成 X%"    let yield_control = helper_function(scope, "yield_control", yield_control_callback)?; // yield_control()：主动让出——立即把当前输出交回模型    let exit = helper_function(scope, "exit", exit_callback)?; // exit()：抛哨兵异常终止 cell，已产出的内容保留    set_global(scope, global, "tools", tools.into())?; // 工具入口：await tools.shell({...}) 这类调用都从这里进    set_global(scope, global, "ALL_TOOLS", all_tools)?; // 元数据数组挂全局    set_global(scope, global, "text", text.into())?; // 其余 helper（image/audio/store/load/notify/yield_control/exit/setTimeout/clearTimeout）同样逐个挂载    Ok(()) // 全部就绪，JS 代码可以开始执行了

pub(super) fn tool_callback(scope: &mut v8::PinScope<'_, '_>, args: v8::FunctionCallbackArguments, mut retval: v8::ReturnValue<v8::Value>) { // JS 调 tools.xxx(...) 时 V8 回调到这里——工具桥的入口    let tool_index = match args.data().to_rust_string_lossy(scope).parse::<usize>() { // 函数模板 data 里存的是工具下标（字符串形式），这里还原成 usize        Ok(tool_index) => tool_index, // 正常路径：拿到下标去查 enabled_tools        Err(_) => { throw_type_error(scope, "invalid tool callback data"); return; } // 数据损坏 → JS 侧抛 TypeError，fail loud    };    let input = if args.length() == 0 { Ok(None) } else { v8_value_to_json(scope, args.get(0)) }; // 参数序列化成 JSON——跨语言边界的统一格式；无参工具传 None    let input = match input { // 序列化结果分派：成功继续，失败在 JS 侧抛错        Ok(input) => input, // JSON 化成功（或 None）        Err(error_text) => { throw_type_error(scope, &error_text); return; } // 参数不可序列化 → TypeError，工具调用不会发生    };    let Some(resolver) = v8::PromiseResolver::new(scope) else { throw_type_error(scope, "failed to create tool promise"); return; }; // 每个工具调用对应一个 Promise：JS 侧 await 的就是它    let promise = resolver.get_promise(scope);    let resolver = v8::Global::new(scope, resolver); // 提升为全局引用——resolver 要跨 scope 存活到结果回填那一刻    let (tool_name, tool_kind) = { // 查表拿工具名与 kind（Function/Freeform）        let Some(state) = scope.get_slot::<RuntimeState>() else { throw_type_error(scope, "runtime state unavailable"); return; }; // 运行时状态不可用属于内部错误，直接抛        let Some(tool) = state.enabled_tools.get(tool_index) else { throw_type_error(scope, "tool callback data is out of range"); return; }; // 越界防护：data 与 enabled_tools 必须一致        (tool.tool_name.clone(), tool.kind) // 克隆出来，后面发事件要用    };    let Some(state) = scope.get_slot_mut::<RuntimeState>() else { throw_type_error(scope, "runtime state unavailable"); return; }; // 取可变状态：要改计数器、登记 pending    let id = format!("tool-{}", state.next_tool_call_id); // 运行时内部 ID：tool-1、tool-2……与模型侧 call_id 区分开    state.next_tool_call_id = state.next_tool_call_id.saturating_add(1); // 自增（饱和加法，防溢出回绕）    let event_tx = state.event_tx.clone();    state.pending_tool_calls.insert(id.clone(), resolver); // 登记 pending：actor 把工具结果送回来时按 id resolve/reject    let _ = event_tx.send(RuntimeEvent::ToolCall { id, name: tool_name, kind: tool_kind, input }); // 事件出线程——actor 收到后才会真正执行这个工具    retval.set(promise.into()); // JS 侧拿到 Promise，await 挂起；V8 线程随后进入 Pending 冻结

pub(crate) fn commit_completion(&self, event: CellEvent, pending_initial_yield_items: Option<Vec<OutputItem>>, commit: impl FnOnce()) -> CompletionCommit { // 提交完成结果：先落账、再改阶段，两步都在同一把锁内    let mut phase = self.phase.lock().unwrap_or_else(std::sync::PoisonError::into_inner); // 阶段锁：commit/deliver/route_observation 三个入口共用一把锁串行化    if !matches!(*phase, CellPhase::Running) || self.cancellation_token.is_cancelled() { return CompletionCommit::Rejected(event); } // 非 Running（已被终止/已交付）或已取消 → 拒绝提交，事件原样退回给 actor    commit(); // 回调里把 store() 的写入合并进会话级 stored_values——先落账再改阶段，保证不丢写    *phase = CellPhase::Completed { pending_initial_yield_items, event }; // 原子地进入 Completed：结果在此"冻结"，等待交付    CompletionCommit::Committed}pub(crate) fn deliver_completion(&self, response_tx: Option<oneshot::Sender<Result<CellEvent, CellError>>>) -> CompletionDelivery { // 交付完成结果：抢锁 → 取事件 → 发送，任何一步失败都有明确回退    let mut phase = self.phase.lock().unwrap_or_else(std::sync::PoisonError::into_inner);    let (pending_initial_yield_items, event) = match std::mem::replace(&mut *phase, CellPhase::Tombstone) { // 先抢成 Tombstone：保证全局只有一个交付者能拿到结果        CellPhase::Completed { pending_initial_yield_items, event } => (pending_initial_yield_items, event), // 正常路径：取出冻结的结果        previous => { *phase = previous; return CompletionDelivery::Rejected(response_tx); } // 不是 Completed（比如已被终止）→ 恢复原状并拒绝交付    };    let Some(response_tx) = response_tx else { *phase = CellPhase::Completed { pending_initial_yield_items, event }; return CompletionDelivery::Buffered; }; // 没有等待者：退回 Completed 缓冲，等下一次 observe() 来取    match response_tx.send(Ok(event)) {        Ok(()) => { self.cancellation_token.cancel(); CompletionDelivery::Delivered } // 交付成功 → 取消 token，cell 生命周期正式结束        Err(Ok(event)) => { *phase = CellPhase::Completed { pending_initial_yield_items, event }; CompletionDelivery::Buffered } // 接收端已 drop：再缓冲一次，不丢结果        Err(Err(error)) => panic!("completion delivery unexpectedly carried an actor error: {error:?}") // 逻辑上不可能：oneshot 里只装过 Ok(event)    }

pub async fn execute(&self, request: ExecuteRequest) -> Result<StartedCell, String> { // 模型侧入口：一次 exec 调用 = 创建一个 cell    let yield_time_ms = request.yield_time_ms.unwrap_or(DEFAULT_EXEC_YIELD_TIME_MS); // 模型没写 pragma 就用默认 10s（DEFAULT_EXEC_YIELD_TIME_MS = 10_000）    let started = self.runtime.execute(runtime_request(request), runtime::ObserveMode::YieldAfter(self.resolve_yield_timeout(yield_time_ms))).await.map_err(|error| error.to_string())?; // 初始观察模式 = YieldAfter：跑满窗口就交付一次中间结果，cell 继续活着    let cell_id = protocol_cell_id(&started.cell_id);    let (response_tx, response_rx) = oneshot::channel(); // 首个事件异步转交：execute() 立即返回 StartedCell，不阻塞调用方    tokio::spawn(async move { let response = started.initial_event().await.map_err(|error| error.to_string()).and_then(|event| runtime_response(&response_cell_id, event)); let _ = response_tx.send(response); }); // 后台任务等首个观察结果并投递    Ok(StartedCell::from_result_receiver(cell_id, response_rx)) // 返回"带接收端"的 cell：模型侧 await initial_response() 拿第一次交付}fn resolve_yield_timeout(&self, yield_time_ms: u64) -> Duration { // 把模型的 yield_time_ms 折算成 actor 用的定时器时长    let yield_time = Duration::from_millis(yield_time_ms);    let timeout = if yield_time >= MIN_YIELD_TIME_FOR_GRACE { // ≥10s（MIN_YIELD_TIME_FOR_GRACE）的长任务额外加宽限        yield_time.saturating_add(YIELD_GRACE_PERIOD) // +1s 宽限：防止"刚好在交付边界上"的工具调用被误判超时    } else { yield_time }; // 短窗口不加——小任务要的是快，不是稳    self.cell_execution_limits.max_yield_time_ms.map(Duration::from_millis).map_or(timeout, |limit| timeout.min(limit)) // 会话级上限兜底：模型不能无限拉长单次观察窗口

let mut lines = input.splitn(2, '\n'); // 只切第一行：pragma 必须独占首行，其余全部是 JS 源码let first_line = lines.next().unwrap_or_default();let rest = lines.next().unwrap_or_default(); // rest = pragma 之后的所有行（真正的代码）let trimmed = first_line.trim_start();let Some(pragma) = trimmed.strip_prefix(CODE_MODE_PRAGMA_PREFIX) else { return Ok(args); }; // 没有 `// @exec:` 前缀 → 整段当源码，零开销直通if rest.trim().is_empty() { return Err("exec pragma must be followed by JavaScript source on subsequent lines".to_string()); } // pragma 后面必须有真代码——防止"只写 pragma 不干活"的空执行let directive = pragma.trim();if directive.is_empty() { return Err("exec pragma must be a JSON object with supported fields `yield_time_ms` and `max_output_tokens`".to_string()); } // 空 pragma 直接报错，不留模糊空间let value: serde_json::Value = serde_json::from_str(directive).map_err(|err| format!("exec pragma must be valid JSON ...: {err}"))?; // 严格 JSON 解析：不是对象/字段非法都会带上下文报错for key in object.keys() { match key.as_str() { "yield_time_ms" | "max_output_tokens" => {}, _ => return Err(format!("exec pragma only supports `yield_time_ms` and `max_output_tokens`; got `{key}`")) } } // 字段白名单：未知字段直接报错——fail loud，不静默忽略args.code = rest.to_string(); // pragma 行被剥掉，V8 里跑的是纯 JSargs.yield_time_ms = pragma.yield_time_ms; // 两个参数原样透传给运行时

Box::pin(async move { // invoke_tool 的异步体：cell actor 调 host.invoke_tool() 时进入这里    let call_id = format!("{PUBLIC_TOOL_NAME}-{}", uuid::Uuid::new_v4()); // 嵌套调用生成独立 call_id：exec-<uuid>，和模型侧的 tool_call_id 区分开    if cancellation_token.is_cancelled() { dispatch_trace.interruption = Some(DispatchInterruption::Cancelled); return Err("code mode nested tool call cancelled".to_string()); } // 入队前已取消 → 直接短路，不打扰 worker    let (response_tx, response_rx) = oneshot::channel(); // 结果回传通道：worker 执行完把 JSON 结果塞回来    self.dispatch_tx.send(DispatchMessage::InvokeTool { invocation, dispatch_trace, cancellation_token: cancellation_token.clone(), response_tx }).await.map_err(|_| "code mode nested tool dispatcher is unavailable".to_string())?; // 投递给 turn worker——只有它能判断"派发是否跑赢取消"    tokio::select! {        response = response_rx => response.map_err(|_| "code mode nested tool dispatcher stopped".to_string())?, // 正常路径：worker 执行完回传结果        _ = cancellation_token.cancelled() => Err("code mode nested tool call cancelled".to_string()), // cell 被终止 → 立即返回错误，V8 侧 Promise 会被 reject    }

exec 调用全链路（模型 → V8 → 宿主工具 → 模型）

▼

▼

▼

▼

▼