From 8978cc256b77986c33a10d56e147f42d95cfaa63 Mon Sep 17 00:00:00 2001 From: Shizuina Date: Mon, 9 Nov 2026 22:44:27 +0800 Subject: [PATCH 01/14] feat(assistant): add scheduled read-only discovery to Rust Bot Add opt-in discovery schedules using fresh read-only worker Sessions and explicit report or no_report settlement through the existing Bot path. Preserve original execution identities across restart and Host handoff. Refs: apache/maka#6716 Generated-by: OpenAI Codex --- packages/plugin-sdk/src/contracts.ts | 87 +++ packages/plugin-sdk/src/generated/runtime.ts | 6 + plugins/agent/api/src/contract/scheduler.rs | 10 +- plugins/agent/native/README.md | 33 + plugins/agent/native/README.zh-CN.md | 26 + plugins/agent/native/src/owner.rs | 34 +- plugins/agent/native/src/owner/discovery.rs | 722 ++++++++++++++++++ plugins/agent/native/src/owner/execution.rs | 2 +- .../agent/native/src/owner/goal/dispatch.rs | 2 +- .../agent/native/src/owner/notifications.rs | 9 +- plugins/agent/native/src/owner/scheduler.rs | 12 +- .../native/src/owner/scheduler/dispatch.rs | 92 ++- .../native/src/owner/scheduler/repository.rs | 6 +- .../agent/native/src/owner/scheduler/tools.rs | 28 +- plugins/agent/native/src/owner/tests.rs | 1 + .../native/src/owner/tests/discovery_tests.rs | 535 +++++++++++++ .../native/src/owner/tests/scheduler_tests.rs | 8 +- plugins/agent/native/src/tools.rs | 9 + 18 files changed, 1577 insertions(+), 45 deletions(-) create mode 100644 plugins/agent/native/src/owner/discovery.rs create mode 100644 plugins/agent/native/src/owner/tests/discovery_tests.rs diff --git a/packages/plugin-sdk/src/contracts.ts b/packages/plugin-sdk/src/contracts.ts index 7ead9aad12f..598c2b447d8 100644 --- a/packages/plugin-sdk/src/contracts.ts +++ b/packages/plugin-sdk/src/contracts.ts @@ -22136,6 +22136,35 @@ export const contracts = { additionalProperties: false, required: ['kind', 'target', 'model'], }, + { + description: + 'Read-only investigation in a fresh worker. The target is the current Bot.', + type: 'object', + properties: { + kind: { + type: 'string', + const: 'discovery', + }, + target: { + $ref: '#/$defs/AgentScheduledTarget', + }, + model: { + $ref: '#/$defs/ModelSelection', + }, + thinkingLevel: { + anyOf: [ + { + $ref: '#/$defs/ThinkingLevel', + }, + { + type: 'null', + }, + ], + }, + }, + additionalProperties: false, + required: ['kind', 'target', 'model'], + }, ], }, AgentScheduledTarget: { @@ -22960,6 +22989,35 @@ export const contracts = { additionalProperties: false, required: ['kind', 'target', 'model'], }, + { + description: + 'Read-only investigation in a fresh worker. The target is the current Bot.', + type: 'object', + properties: { + kind: { + type: 'string', + const: 'discovery', + }, + target: { + $ref: '#/$defs/AgentScheduledTarget', + }, + model: { + $ref: '#/$defs/ModelSelection', + }, + thinkingLevel: { + anyOf: [ + { + $ref: '#/$defs/ThinkingLevel', + }, + { + type: 'null', + }, + ], + }, + }, + additionalProperties: false, + required: ['kind', 'target', 'model'], + }, ], }, AgentScheduledTarget: { @@ -29568,6 +29626,35 @@ export const contracts = { additionalProperties: false, required: ['kind', 'target', 'model'], }, + { + description: + 'Read-only investigation in a fresh worker. The target is the current Bot.', + type: 'object', + properties: { + kind: { + type: 'string', + const: 'discovery', + }, + target: { + $ref: '#/$defs/AgentScheduledTarget', + }, + model: { + $ref: '#/$defs/ModelSelection', + }, + thinkingLevel: { + anyOf: [ + { + $ref: '#/$defs/ThinkingLevel', + }, + { + type: 'null', + }, + ], + }, + }, + additionalProperties: false, + required: ['kind', 'target', 'model'], + }, ], }, AgentScheduledTarget: { diff --git a/packages/plugin-sdk/src/generated/runtime.ts b/packages/plugin-sdk/src/generated/runtime.ts index 86568f501a8..af1df81b826 100644 --- a/packages/plugin-sdk/src/generated/runtime.ts +++ b/packages/plugin-sdk/src/generated/runtime.ts @@ -2293,6 +2293,12 @@ export type AgentScheduledEffect = target: AgentScheduledTarget; model: ModelSelection; thinkingLevel?: ThinkingLevel | null; + } + | { + kind: 'discovery'; + target: AgentScheduledTarget; + model: ModelSelection; + thinkingLevel?: ThinkingLevel | null; }; export type Creator = | { diff --git a/plugins/agent/api/src/contract/scheduler.rs b/plugins/agent/api/src/contract/scheduler.rs index 1422c9f6323..063e72a33ed 100644 --- a/plugins/agent/api/src/contract/scheduler.rs +++ b/plugins/agent/api/src/contract/scheduler.rs @@ -40,12 +40,20 @@ pub enum ScheduledEffect { model: ModelSelection, thinking_level: Option, }, + /// Read-only investigation in a fresh worker. The target is the current Bot. + Discovery { + target: ScheduledTarget, + model: ModelSelection, + thinking_level: Option, + }, } impl maka_scheduler::task::TaskEffect for ScheduledEffect { fn validate(&self) -> Result<(), maka_scheduler::Error> { match self { Self::Reminder => Ok(()), - Self::Session { target, .. } | Self::AgentRun { target, .. } => { + Self::Session { target, .. } + | Self::AgentRun { target, .. } + | Self::Discovery { target, .. } => { if target.session_id.is_empty() || target.boundary_revision > maka_agent_facts::interaction::MAX_SAFE_INTEGER || target.binding.validate().is_err() diff --git a/plugins/agent/native/README.md b/plugins/agent/native/README.md index 022a356cc66..0672686951e 100644 --- a/plugins/agent/native/README.md +++ b/plugins/agent/native/README.md @@ -118,6 +118,39 @@ The default composition enables `recaps`. The Session recap command reads cached The default composition enables `goals`. `/goal` or the composer’s + menu opens its panel; `/goal ` starts immediately. Goals continue through the existing Session queue without an iteration limit. `CreateGoal`, `GetGoal` and `GoalStatus` work in Direct and Code Mode; the current objective, progress and budget accompany model requests, including after compaction. A model-created Goal owns its creating turn. Pause stops current work; Resume continues with the selected model and settings; Cancel targets only that Goal. Transient execution failures remain recoverable. Token budgets are optional scheduling thresholds checked between turns; missing usage is reported as incomplete rather than stopping work, and users can adjust budgets. Reads expose current usage, remaining budget and active elapsed time; completed usage stays frozen after cleanup. Original operation identities and receipts survive restart, and sealed Host handoffs use their reserved successors rather than replaying effects. Archived Sessions and changed execution prerequisites require user action before further execution. +## Read-only discovery + +With `bot` and `scheduledTasks` enabled, the current Bot can propose a `ScheduledTask` +with `effect: "discovery"`. As with other model-created schedules, this is a paused +proposal until a human approves its exact workspace, permissions and model through +`schedule`. Human-created schedules use the `discovery` effect with that Bot as the +target. Existing `once`, `interval` and `cron` timing, `trigger_now`, `pause` and +`resume` controls apply; no discovery work is enabled by default. + +Each fire creates a fresh ordinary worker Session. It inherits the approved workspace +and model but uses read-only permissions without permission requests. Its tools are +the composition's Files read/image/glob/grep operations, Recall when enabled, and +`DiscoveryComplete`; the coordinator's tools, MCP and shell access are not inherited. +The worker must call `DiscoveryComplete` with either +`{"kind":"no_report","note":"What was checked"}` or +`{"kind":"report","text":"Findings and source references"}`. This records the +result and ends the Turn. Ordinary text without that call is an incomplete investigation. + +One investigation, including an outstanding report handoff, occupies the shared slot. +The scheduler waits for actual execution cleanup: admission is not completion. +`no_report` settles silently; a report is admitted to the Bot once, using its existing +notification path, where the Bot decides whether to call `SendUser`. The recorded run +links to the worker and distinguishes report admission from user delivery. Pausing +cancels the current execution, including a handoff successor, and discards unsent +reports; it cannot retract an already admitted Bot message. Restart reconciles the +original fire, Turn and receipt instead of creating replacement work. A sealed Host +handoff uses its reserved successor; an interrupted terminal run is recorded as failed. + +Plans, the active job and completed receipts use the existing scheduler KV namespace; +no separate database or lease is introduced. This slice does not import TS discovery +data, add shared notebook memory or observe external events. The [discovery tests](src/owner/tests/discovery_tests.rs) +exercise the real Host with local model fixtures. + [Source](src/lib.rs) · [Cargo manifest](Cargo.toml) [Performance baseline](checkpoint-baseline.md) diff --git a/plugins/agent/native/README.zh-CN.md b/plugins/agent/native/README.zh-CN.md index c1d657a07ee..55f486d29f2 100644 --- a/plugins/agent/native/README.zh-CN.md +++ b/plugins/agent/native/README.zh-CN.md @@ -116,6 +116,32 @@ Maka 默认组合启用 `automaticTitles`。通过 `automaticTitle: true` 创建 默认组合启用 `goals`。`/goal` 或输入框加号菜单打开目标面板；`/goal 目标内容` 直接开始。目标复用现有会话队列自动续跑，不设迭代上限。`CreateGoal`、`GetGoal` 和 `GoalStatus` 可从 Direct 和 Code Mode 调用；每次模型请求都携带当前目标、进度和预算，压缩后仍可读取。模型创建的目标将当前轮作为第一轮。暂停停止当前工作，继续使用所选模型与设置，取消只作用于此目标。临时执行失败允许恢复。Token 预算为可选的跨轮调度阈值；缺失计量如实显示为用量不完整，不会因此停工，用户可以调整预算。读取提供实时用量、剩余预算和累计执行时间；完成并清理后，用量不再随普通聊天增长。重启保留原始操作身份与回执，Host 交接沿封存的后继执行恢复，避免重放操作。归档会话或执行前提变更时，需要用户处理后继续。 +## 只读主动发现 + +启用 `bot` 和 `scheduledTasks` 后，当前 Bot 可通过 `ScheduledTask` 提议 +`effect: "discovery"`。与其他模型创建的定时任务一样，提议保持暂停，直到用户通过 +`schedule` 批准其精确工作区、权限和模型。用户直接创建时使用 `discovery` 效果， +以当前 Bot 为目标。沿用 `once`、`interval`、`cron` 定时及 `trigger_now`、`pause`、 +`resume` 控制；默认不会启用任何调查。 + +每次触发新建一个普通 worker Session，沿用已批准的工作区和模型，权限收窄为只读， +禁止申请额外权限。工具仅包括当前组合中的 Files 读取、图片读取、glob、grep，已启用的 +Recall 和 `DiscoveryComplete`；不继承协调工具、MCP 或 shell 能力。worker 必须调用 +`DiscoveryComplete`，提交 `{"kind":"no_report","note":"检查了什么"}` 或 +`{"kind":"report","text":"发现及来源"}`。调用持久记录结果并结束本轮；只输出普通文本 +属于未完整提交调查结果。 + +所有调查共用一个执行位，等待报告交接时也占位。调度器等待真实执行清理结束，不把接纳 +当作完成。`no_report` 静默结算；报告通过已有通知路径只向 Bot 接纳一次，由 Bot 判断 +是否调用 `SendUser`。运行记录链接到 worker，并区分“报告已接纳”和“用户已收到”。 +暂停会取消当前执行，包括交接后的继任 invocation，并舍弃尚未投递的报告；已接纳的 +Bot 消息不能撤回。重启按原触发、Turn 和回执恢复，不新建替代调查。已封存的 Host +交接沿用预留的继任者；被中断并终结的执行记录为失败。 + +计划、当前调查及已完成回执复用调度器 KV 命名空间，不增加独立数据库或租约。 +本阶段不导入 TS 主动发现数据、不增加共享记事本或外部事件监听。 +[调查测试](src/owner/tests/discovery_tests.rs) 使用真实 Host 和本地模型夹具验证。 + [源码](src/lib.rs) · [Cargo 配置](Cargo.toml) [性能基线](checkpoint-baseline.md) diff --git a/plugins/agent/native/src/owner.rs b/plugins/agent/native/src/owner.rs index 9a3003dfae7..c86fc4ffe84 100644 --- a/plugins/agent/native/src/owner.rs +++ b/plugins/agent/native/src/owner.rs @@ -31,6 +31,7 @@ mod checkpoints; mod compaction; mod computer_apps; mod context; +mod discovery; mod execution; mod external_setup; mod files; @@ -140,12 +141,13 @@ impl maka_agent_engine::NativeActivationBasis for MessageBasis { fn is_agent_collaboration(mode: &CollaborationMode) -> bool { *mode == CollaborationMode::Agent } -#[derive(Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] enum BackgroundKind { #[default] Schedule, Goal(Uuid), + Discovery, } impl BackgroundKind { fn is_schedule(&self) -> bool { @@ -238,6 +240,7 @@ pub(crate) struct Owner { tool_basis: Arc>, host_tools: Vec, resource_reads: Vec, + discovery_reads: Vec, editing_tools: Vec<(String, maka_agent_facts::execution::EditingTools)>, skills: Option, pub(super) executors: crate::executor::Executors, @@ -503,6 +506,12 @@ impl Owner { .with_always_visible(&[bot::SEND_USER, bot::SLEEP]) .map_err(|error| error.to_string())?; } + if settings.scheduled_tasks { + catalog.tools = catalog + .tools + .with_always_visible(&[discovery::COMPLETE]) + .map_err(|error| error.to_string())?; + } if settings.recall { catalog.tools = catalog .tools @@ -522,6 +531,7 @@ impl Owner { editing: editing_tools, host_tools, resource_reads, + discovery_reads, executor: native_approval_executor, } = match prepared_tools { Ok(tools) => tools, @@ -612,6 +622,7 @@ impl Owner { tool_basis: Arc::new(tool_basis), host_tools: host_tools.into_iter().chain(internal_host_tools).collect(), resource_reads, + discovery_reads, editing_tools, skills, executors, @@ -628,7 +639,15 @@ impl Owner { file_native_hints: std::sync::atomic::AtomicBool::new(true), stop, }); - if let Err(error) = owner.revoke_lost_client_mcp().await { + if let Err(error) = async { + owner + .revoke_lost_client_mcp() + .await + .map_err(|error| error.to_string())?; + owner.restore_discovery().await + } + .await + { let cleanup = owner.close().await; return Err(match cleanup { Ok(()) => error.to_string(), @@ -1208,7 +1227,16 @@ impl Owner { .map(|definition| definition.name.as_str()) .filter(|name| { let speaks = *name == bot::SEND_USER || *name == bot::SLEEP; - (speaks && !bot) + (*name == discovery::COMPLETE + && !session.is_some_and(|id| { + self.scheduler + .investigation + .read() + .unwrap() + .as_ref() + .is_some_and(|active| active.session_id == id) + })) + || (speaks && !bot) || (!(speaks && bot) && allowed_tools.is_some_and(|allowed| { !allowed.iter().any(|allowed| allowed.as_str() == *name) diff --git a/plugins/agent/native/src/owner/discovery.rs b/plugins/agent/native/src/owner/discovery.rs new file mode 100644 index 00000000000..b5462bf7c66 --- /dev/null +++ b/plugins/agent/native/src/owner/discovery.rs @@ -0,0 +1,722 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +//! Discovery is a scheduled, read-only Session. The scheduler retains its fire +//! until execution and any report handoff settle; empty results never wake the Bot. +use super::*; +use maka_agent_facts::{event::InvocationOutcome, tool_call::ToolRejection, tools::*}; +use maka_capabilities::storage::{Data, Mutation, Store}; +use maka_event_log::{ + StoreError, + message_admissions::{PendingExecution, PendingMessageAdmission}, + turns::InvocationState, +}; +use maka_scheduler::{ + plan::Fire, + task::{Outcome, Run}, +}; +use std::sync::Weak; + +pub(super) const COMPLETE: &str = "DiscoveryComplete"; +const ACTIVE: &str = "discovery/active"; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +enum Report { + /// Nothing worth notifying; record what was checked without waking the Bot. + NoReport { note: String }, + /// Actual findings and their source references, not instructions for the Bot. + Report { text: String }, +} + +#[derive(Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +enum Phase { + Investigating { + result: Option, + }, + Delivering { + text: String, + notification: Box, + }, + Stopping, +} +#[derive(Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Job { + fire_id: String, + task_id: String, + bot: String, + admission: PendingMessageAdmission, + phase: Phase, +} +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Completed { + run: Run, + result: Option, +} + +fn completed_key(fire: &str) -> String { + format!( + "discovery/completed/{}", + hex::encode(Sha256::digest(fire.as_bytes())) + ) +} +fn reason(error: impl std::fmt::Display) -> String { + error.to_string() +} +fn mutation( + key: String, + revision: Option, + value: &T, +) -> Result { + Ok(Mutation { + key, + expected_revision: revision, + data: Data::Present(serde_json::to_value(value).map_err(reason)?), + }) +} +pub(super) fn permissions() -> WorkspacePermissions { + WorkspacePermissions { + access: WorkspaceAccess::ReadOnly, + request_permissions: false, + } +} + +impl scheduler::Scheduler { + async fn investigation(&self) -> Result<(Option, Option), String> { + let record = self.storage.read(ACTIVE.into()).await.map_err(reason)?; + let revision = record.as_ref().map(|record| record.revision); + let job: Option = record + .and_then(|record| record.data.value().cloned()) + .map(serde_json::from_value) + .transpose() + .map_err(reason)?; + // Reconcile the visibility cache after an uncertain write acknowledgement. + *self.investigation.write().unwrap() = + job.as_ref().map(|job| job.admission.invocation.clone()); + Ok((revision, job)) + } + async fn save_investigation(&self, revision: Option, job: &Job) -> Result { + let written = self + .storage + .batch(vec![mutation(ACTIVE.into(), revision, job)?]) + .await + .map_err(reason)?; + *self.investigation.write().unwrap() = Some(job.admission.invocation.clone()); + Ok(written[0].revision) + } + pub(super) fn discovery_tool(self: &Arc) -> ToolRegistration { + ToolRegistration { + definition: ToolDefinition { + provider: None, name: COMPLETE.into(), + description: "Finish this read-only discovery investigation. Use no_report with a concise note when nothing merits notifying the user; otherwise report concrete findings and source references. This stores the result and ends the Turn. It does not directly send a user message.".into(), + input_schema: serde_json::to_value(schemars::schema_for!(Report)).expect("discovery schema"), + output_schema: None, freeform: None, + }, + nesting: ToolNesting::DirectOnly, + semantics: ToolSemantics::FinishTurn, + handler: ToolHandler::Prepared(self.clone()), + } + } + pub(super) fn prepare_discovery( + &self, + input: Value, + context: ToolCallContext, + cancellation: CancellationToken, + ) -> PreparationFuture { + let owner = self.link.get().and_then(Weak::upgrade); + Box::pin(async move { + let owner = owner.ok_or(ToolRejection::Unavailable)?; + if cancellation.is_cancelled() { + return Err(ToolRejection::Cancelled); + } + let report: Report = + serde_json::from_value(input).map_err(|e| ToolRejection::InvalidInput { + message: e.to_string(), + })?; + let text = match &report { + Report::NoReport { note } => note, + Report::Report { text } => text, + }; + if text.trim().is_empty() { + return Err(ToolRejection::InvalidInput { + message: "Provide a nonempty result or investigation note".into(), + }); + } + Ok(PreparedEffect::new(move |cancel| { + Box::pin(async move { + if cancel.is_cancelled() { + return Err(ToolError::Failed("Discovery cancelled".into())); + } + // An observer cannot discard an accepted result write. + let captured = owner.clone(); + let job = owner + .workers + .spawn(async move { captured.complete_discovery(context, report).await }); + job.await + .map_err(|e| ToolError::OutcomeUnknown(e.to_string()))? + .map_err(ToolError::Failed)?; + Ok(serde_json::json!({"recorded":true}).into()) + }) + })) + }) + } +} + +impl Owner { + pub(super) async fn restore_discovery(&self) -> Result<(), String> { + self.scheduler.investigation().await?; + Ok(()) + } + + pub(super) fn investigation_basis(&self, approved: &MessageBasis) -> MessageBasis { + let mut basis = approved.clone(); + let mut tools = self.discovery_reads.clone(); + // Recall preserves its existing source authorization. No MCP, shell, + // delegation, approval, or filesystem mutation tool is granted. + for name in ["Recall", "RecallMore", COMPLETE] { + if self.tools.contains(name) { + tools.push(name.into()); + } + } + tools.sort(); + tools.dedup(); + basis.allowed_tools = Some(tools); + basis.background_kind = BackgroundKind::Discovery; + basis + } + + pub(super) async fn validate_discovery_background(&self) -> Result<(), String> { + let (_, job) = self.scheduler.investigation().await?; + let job = job.ok_or("discovery_cancelled")?; + let catalog = self.scheduler.load().await?; + if !self.bot.is(&job.bot) + || matches!(job.phase, Phase::Stopping) + || !catalog.plans.get(&job.task_id).is_some_and(|saved| { + saved.plan.task.status == maka_scheduler::task::Status::Active + && saved + .plan + .pending + .as_ref() + .is_some_and(|fire| fire.id == job.fire_id) + }) + { + return Err("discovery_cancelled".into()); + } + Ok(()) + } + + async fn complete_discovery( + &self, + context: ToolCallContext, + report: Report, + ) -> Result<(), String> { + let _gate = self.scheduler.gate.lock().await; + let (revision, job) = self.scheduler.investigation().await?; + let mut job = job.ok_or("discovery_not_active")?; + let original = &job.admission.invocation; + if context.invocation.session_id != original.session_id + || context.invocation.turn_id != original.turn_id + { + return Err("discovery_wrong_invocation".into()); + } + let store = self.tool_store(&context)?; + let boundary = store + .turn_boundary(&original.session_id, &original.turn_id) + .await + .map_err(reason)? + .ok_or("discovery_execution_unavailable")?; + if boundary.root_invocation() != original + || boundary.invocation != context.invocation + || matches!(boundary.state, InvocationState::Ended { .. }) + { + return Err("discovery_execution_changed".into()); + } + let plan = self.scheduler.load().await?; + if !self.bot.is(&job.bot) + || !plan.plans.get(&job.task_id).is_some_and(|saved| { + saved.plan.task.status == maka_scheduler::task::Status::Active + && saved + .plan + .pending + .as_ref() + .is_some_and(|fire| fire.id == job.fire_id) + }) + { + return Err("discovery_cancelled".into()); + } + let basis: MessageBasis = + serde_json::from_value(job.admission.execution.basis.clone()).map_err(reason)?; + self.validate_background(&basis).await?; + let Phase::Investigating { result } = &mut job.phase else { + return Err("discovery_cancelled".into()); + }; + if let Some(previous) = result { + if *previous != report { + return Err("discovery_result_conflict".into()); + } + return Ok(()); + } + *result = Some(report); + self.scheduler.save_investigation(revision, &job).await?; + self.scheduler.wake.notify_one(); + Ok(()) + } + + pub(super) async fn dispatch_discovery( + self: &Arc, + fire: &Fire, + approved: &MessageBasis, + enabled: bool, + ) -> Result, String> { + let (mut revision, current) = self.scheduler.investigation().await?; + if let Some(record) = self + .scheduler + .storage + .read(completed_key(&fire.id)) + .await + .map_err(reason)? + && let Some(value) = record.data.value() + { + let completed: Completed = serde_json::from_value(value.clone()).map_err(reason)?; + return Ok(Some(completed.run)); + } + let ScheduledEffect::Discovery { target, .. } = &fire.effect else { + unreachable!() + }; + if current.as_ref().is_some_and(|job| job.fire_id != fire.id) { + return Ok(None); + } + let mut job = if let Some(job) = current { + job + } else { + let identity = hex::encode(Sha256::digest(format!("discovery/{}", fire.id).as_bytes())); + let worker = format!("{}{}", self.prefix, &identity[..32]); + let mut basis = approved.clone(); + let content: MessageInput = match basis.language { + Language::ZhCn => format!("执行一次已授权的只读调查：\n{}\n\n仅检查本次任务相关且已授权的材料。不要修改文件、扩大权限或安排后续任务。结束时必须调用 DiscoveryComplete：没有值得通知的发现用 no_report，附简短调查记录；有发现用 report，说明实际依据与来源。普通回复不代表已提交结果。", fire.intent.body()), + Language::ZhTw => format!("執行一次已授權的唯讀調查：\n{}\n\n只檢查本次任務相關且已授權的資料。不要修改檔案、擴大權限或安排後續任務。結束時必須呼叫 DiscoveryComplete：沒有值得通知的發現用 no_report，附簡短調查記錄；有發現用 report，說明實際依據與來源。一般回覆不代表已提交結果。", fire.intent.body()), + _ => format!("Perform one authorized read-only investigation:\n{}\n\nInspect only relevant, authorized material. Do not modify files, expand permissions or schedule more work. Finish with DiscoveryComplete: no_report with a concise investigation note when there is nothing worth notifying; report with concrete findings and source references otherwise. Ordinary text is not a submitted result.", fire.intent.body()), + }.into(); + basis.fingerprint = content.content_digest().map_err(reason)?; + let job = Job { + fire_id: fire.id.clone(), + task_id: fire.task_id.clone(), + bot: target.session_id.clone(), + phase: Phase::Investigating { result: None }, + admission: PendingMessageAdmission { + execution: PendingExecution { + owner: self.prefix.clone(), + basis: serde_json::to_value(basis).map_err(reason)?, + }, + invocation: Invocation { + session_id: worker, + turn_id: format!("t_{identity}"), + run_id: format!("r_{identity}"), + invocation_id: format!("i_{identity}"), + }, + source: RootSourceMessage { + message: DeliveredMessage { + message_id: format!("m_{identity}"), + origin: None, + submitted_content_digest: content.content_digest().map_err(reason)?, + content: content.clone(), + }, + unprepared_content: content, + submitted_placement: Placement::CurrentTurn, + disposition: MessageDisposition::TurnStarted, + submitted_intent: None, + }, + steering_invocation: None, + required_tools: Default::default(), + admitted_at: fire.scheduled_at as u64, + }, + }; + revision = Some(self.scheduler.save_investigation(revision, &job).await?); + job + }; + let basis: MessageBasis = + serde_json::from_value(job.admission.execution.basis.clone()).map_err(reason)?; + if let Phase::Delivering { notification, .. } = &job.phase + && self.discovery_notification_admitted(notification).await? + { + return self + .settle_discovery(fire, revision, &job, Outcome::Ok, "Report admitted to Bot") + .await + .map(Some); + } + let allowed = enabled && self.bot.is(&job.bot); + let authorization = self.validate_background(&basis).await; + let revoked = matches!(&authorization, Err(error) if matches!(error.as_str(), + "scheduler_authorization_changed" | "scheduler_disabled" | "agent_session_unavailable" | "discovery_cancelled")); + if !allowed || revoked { + if !matches!(job.phase, Phase::Stopping) { + job.phase = Phase::Stopping; + revision = Some(self.scheduler.save_investigation(revision, &job).await?); + } + } else { + authorization?; + } + let invocation = &job.admission.invocation; + let worker = &invocation.session_id; + let retired = match self.store.session_retirement(worker).await { + Ok(retirement) => retirement.is_some(), + // The job is persisted before its Session is first created. + Err(StoreError::SessionNotFound) => false, + Err(error) => return Err(reason(error)), + }; + if retired { + return self + .settle_discovery( + fire, + revision, + &job, + Outcome::Blocked, + "Investigation Session retired", + ) + .await + .map(Some); + } + let existing = self + .store + .get_session::(worker) + .await + .map_err(reason)?; + if let Some(existing) = &existing + && basis + .validate_background_target(&resolve_session(existing.clone())?) + .is_err() + && !matches!(job.phase, Phase::Stopping) + { + job.phase = Phase::Stopping; + revision = Some(self.scheduler.save_investigation(revision, &job).await?); + } + if existing.is_none() { + if matches!(job.phase, Phase::Stopping) { + return self + .settle_discovery( + fire, + revision, + &job, + Outcome::Blocked, + "Investigation cancelled before admission", + ) + .await + .map(Some); + } + let source = self.session(&job.bot).await?; + let workspace = source + .workspace_binding + .as_ref() + .ok_or("discovery_workspace_unavailable")? + .reference(); + let model = basis.model.as_ref().ok_or("discovery_model_unavailable")?; + self.store + .create_session( + worker, + &basis.fingerprint, + &StoredSessionConfiguration { + parent_session_id: Some(job.bot.clone()), + backend: Some(BackendPreference::Model { + connection_id: model.binding.connection_id.clone(), + model_id: model.binding.model.clone(), + thinking_level: model.thinking_level, + }), + compaction_prompt: None, + collaboration_mode: CollaborationMode::Agent, + tools: basis.allowed_tools.clone(), + title: fire.title.clone(), + title_is_manual: true, + pinned: false, + labels: vec![], + workspace, + permissions: permissions(), + boundary_revision: 0, + }, + fire.scheduled_at as u64, + ) + .await + .map_err(reason)?; + } + let store = self.store.session(worker).await.map_err(reason)?; + let boundary = store + .turn_boundary(worker, &invocation.turn_id) + .await + .map_err(reason)?; + if boundary + .as_ref() + .is_some_and(|boundary| boundary.root_invocation() != invocation) + { + return Err("discovery_execution_identity_changed".into()); + } + if matches!(job.phase, Phase::Stopping) { + // A sealed handoff keeps the Turn but replaces the active invocation. + let current = boundary + .as_ref() + .map(|boundary| &boundary.invocation) + .unwrap_or(invocation); + let goal = self.goals.gate.lock().await; + self.stop_run_id( + worker, + ¤t.invocation_id, + hex::encode(Sha256::digest( + format!("discovery-stop/{}/{}", fire.id, current.invocation_id).as_bytes(), + )), + &goal, + ) + .await?; + } + let runs = self.runs.lock().await; + let unsettled = runs.busy(worker) || runs.retained_cleanup.contains(worker); + drop(runs); + if unsettled || self.has_parked_invocation(worker).await? { + return Ok(None); + } + let Some(boundary) = boundary else { + if matches!(job.phase, Phase::Stopping) { + // Pending original admissions are rejected by their paused source, + // never replaced with another invocation. + if store + .message_admission(worker, &job.admission.source.message.message_id) + .await + .map_err(reason)? + .is_some() + { + self.schedule_pending().await.map_err(public_error)?; + return Ok(None); + } + return self + .settle_discovery( + fire, + revision, + &job, + Outcome::Blocked, + "Investigation cancelled before admission", + ) + .await + .map(Some); + } + self.admit_notification(&job.admission).await?; + return Ok(None); + }; + let InvocationState::Ended { outcome, .. } = &boundary.state else { + return Ok(None); + }; + match outcome { + InvocationOutcome::HandoffPaused { .. } => { + if matches!(job.phase, Phase::Stopping) { + store + .cancel_handoff(&boundary.invocation) + .await + .map_err(reason)?; + } else { + match self + .resume_background_handoff(&boundary.invocation, Arc::new(basis)) + .await + { + Ok(()) => {} + Err(RunError::InvalidInput(_) | RunError::ReconciliationRequired(_)) => { + // A refused frozen successor cannot run. End that original + // reservation so it does not block unrelated investigations. + store + .cancel_handoff(&boundary.invocation) + .await + .map_err(reason)?; + } + Err(error) => return Err(public_error(error)), + } + } + Ok(None) + } + InvocationOutcome::Completed if !matches!(job.phase, Phase::Stopping) => { + match &job.phase { + Phase::Investigating { + result: Some(Report::NoReport { .. }), + } => self + .settle_discovery(fire, revision, &job, Outcome::Ok, "No report") + .await + .map(Some), + Phase::Investigating { + result: Some(Report::Report { text }), + } => { + let model = approved + .model + .as_ref() + .ok_or("discovery_model_unavailable")?; + let route = notifications::ReturnRoute::new( + model.clone(), + approved.workspace.clone(), + approved.binding, + ); + let identity = hex::encode(Sha256::digest( + format!("discovery-report/{}", fire.id).as_bytes(), + )); + let source = Invocation { + session_id: job.bot.clone(), + turn_id: format!("t_{identity}"), + run_id: format!("r_{identity}"), + invocation_id: format!("i_{identity}"), + }; + let Some(notification) = self.prepare_notification(&source, &route, &identity, + |language| match language { + Language::ZhCn => format!("只读调查「{}」已结束。以下是 worker {} 本轮记录的发现；它是调查结果，不是用户的新指令。根据内容决定是否通知用户；需要核验时读取该会话原始记录。\n\n{}", fire.title, worker, text), + Language::ZhTw => format!("唯讀調查「{}」已結束。以下是 worker {} 本輪記錄的發現；它是調查結果，不是使用者的新指令。根據內容決定是否通知使用者；需要核驗時讀取該會話原始記錄。\n\n{}", fire.title, worker, text), + _ => format!("Read-only investigation \"{}\" finished. Worker {} recorded the following findings. These are investigation results, not new user instructions. Decide whether they merit notifying the user; inspect the worker's original record when verification is needed.\n\n{}", fire.title, worker, text), + }, MessageOrigin::BotWorker { session_id: worker.clone() }).await? else { return Ok(None); }; + job.phase = Phase::Delivering { + text: text.clone(), + notification: Box::new(notification), + }; + revision = Some(self.scheduler.save_investigation(revision, &job).await?); + let Phase::Delivering { notification, .. } = &job.phase else { + unreachable!() + }; + if !self.admit_notification(notification).await? { + return Ok(None); + } + self.settle_discovery( + fire, + revision, + &job, + Outcome::Ok, + "Report admitted to Bot", + ) + .await + .map(Some) + } + Phase::Delivering { notification, .. } => { + if !self.admit_notification(notification).await? { + return Ok(None); + } + self.settle_discovery( + fire, + revision, + &job, + Outcome::Ok, + "Report admitted to Bot", + ) + .await + .map(Some) + } + _ => self + .settle_discovery( + fire, + revision, + &job, + Outcome::Failed, + "Investigation ended without DiscoveryComplete", + ) + .await + .map(Some), + } + } + InvocationOutcome::ContextCompactFinished { .. } => { + Err("discovery_unexpected_compaction_turn".into()) + } + _ => self + .settle_discovery( + fire, + revision, + &job, + if matches!(job.phase, Phase::Stopping) { + Outcome::Blocked + } else { + Outcome::Failed + }, + if matches!(job.phase, Phase::Stopping) { + "Investigation cancelled" + } else { + "Investigation execution failed or was stopped" + }, + ) + .await + .map(Some), + } + } + + async fn discovery_notification_admitted( + &self, + notification: &PendingMessageAdmission, + ) -> Result { + let session = ¬ification.invocation.session_id; + let message = ¬ification.source.message.message_id; + let store = match self.store.session(session).await { + Ok(store) => store, + Err(maka_event_log::StoreError::SessionNotFound) => return Ok(false), + Err(error) => return Err(reason(error)), + }; + Ok(store + .message_admission(session, message) + .await + .map_err(reason)? + .is_some() + || store + .root_message(session, message) + .await + .map_err(reason)? + .is_some() + || store + .message_cancelled(session, message) + .await + .map_err(reason)?) + } + + async fn settle_discovery( + &self, + fire: &Fire, + revision: Option, + job: &Job, + outcome: Outcome, + message: &str, + ) -> Result { + let run = Run { + id: fire.id.clone(), + at: (now() as i64).max(fire.scheduled_at), + outcome, + message: message.into(), + session_id: Some(job.admission.invocation.session_id.clone()), + run_id: Some(job.admission.invocation.run_id.clone()), + }; + let completed = Completed { + run: run.clone(), + result: match &job.phase { + Phase::Investigating { result } => result.clone(), + Phase::Delivering { text, .. } => Some(Report::Report { text: text.clone() }), + Phase::Stopping => None, + }, + }; + self.scheduler + .storage + .batch(vec![ + mutation(completed_key(&fire.id), None, &completed)?, + Mutation { + key: ACTIVE.into(), + expected_revision: revision, + data: Data::Deleted, + }, + ]) + .await + .map_err(reason)?; + *self.scheduler.investigation.write().unwrap() = None; + self.scheduler.wake.notify_one(); + Ok(run) + } +} diff --git a/plugins/agent/native/src/owner/execution.rs b/plugins/agent/native/src/owner/execution.rs index c94a91726b0..d5b7746265b 100644 --- a/plugins/agent/native/src/owner/execution.rs +++ b/plugins/agent/native/src/owner/execution.rs @@ -379,7 +379,7 @@ impl Owner { execution: project(boundary), }) } - pub(super) async fn resume_goal_handoff( + pub(super) async fn resume_background_handoff( self: &Arc, source: &Invocation, basis: Arc, diff --git a/plugins/agent/native/src/owner/goal/dispatch.rs b/plugins/agent/native/src/owner/goal/dispatch.rs index 74c804e0e97..257db21c2f4 100644 --- a/plugins/agent/native/src/owner/goal/dispatch.rs +++ b/plugins/agent/native/src/owner/goal/dispatch.rs @@ -176,7 +176,7 @@ impl Owner { } Status::Active => { match self - .resume_goal_handoff( + .resume_background_handoff( &boundary.invocation, Arc::new(saved.goal.arm.grant.clone()), ) diff --git a/plugins/agent/native/src/owner/notifications.rs b/plugins/agent/native/src/owner/notifications.rs index 25ad2a897db..3244c1784dc 100644 --- a/plugins/agent/native/src/owner/notifications.rs +++ b/plugins/agent/native/src/owner/notifications.rs @@ -304,8 +304,13 @@ impl Owner { return Ok(false); } // Deferred work captures runtime tools when accepted; retries keep that original basis. - let catalog = self.mcp_projects.capture(&basis.workspace).await?; - let catalog = self.mcp_projects.with_client(session, catalog)?; + let catalog = if basis.background_kind == BackgroundKind::Discovery { + // The read-only investigation profile never admits MCP tools. + self.mcp_projects.base.clone() + } else { + let catalog = self.mcp_projects.capture(&basis.workspace).await?; + self.mcp_projects.with_client(session, catalog)? + }; basis.mcp_revision = catalog.revision.clone(); basis.client_mcp = catalog.client_mcp; basis.mcp_catalog = Some(catalog.clone()); diff --git a/plugins/agent/native/src/owner/scheduler.rs b/plugins/agent/native/src/owner/scheduler.rs index 3c24ab0b4f7..aca4007fa7d 100644 --- a/plugins/agent/native/src/owner/scheduler.rs +++ b/plugins/agent/native/src/owner/scheduler.rs @@ -36,16 +36,17 @@ use tokio::sync::Notify; type Plan = TaskPlan; #[derive(Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] -enum Approval { +pub(super) enum Approval { Reminder, Execution { basis: Box }, } pub(crate) struct Scheduler { - storage: Arc, - gate: Mutex<()>, - wake: Notify, + pub(super) storage: Arc, + pub(super) gate: Mutex<()>, + pub(super) wake: Notify, changed: tokio::sync::watch::Sender, - link: OnceLock>, + pub(super) link: OnceLock>, + pub(super) investigation: std::sync::RwLock>, } impl Scheduler { pub fn new(store: Arc, prefix: String) -> Self { @@ -63,6 +64,7 @@ impl Scheduler { wake: Notify::new(), changed: tokio::sync::watch::channel(0).0, link: OnceLock::new(), + investigation: std::sync::RwLock::new(None), } } pub fn bind(&self, owner: &Arc) { diff --git a/plugins/agent/native/src/owner/scheduler/dispatch.rs b/plugins/agent/native/src/owner/scheduler/dispatch.rs index 9f6e61502f3..12030aef4f7 100644 --- a/plugins/agent/native/src/owner/scheduler/dispatch.rs +++ b/plugins/agent/native/src/owner/scheduler/dispatch.rs @@ -27,7 +27,12 @@ impl MessageBasis { if let Some(boundary) = &self.background && (session.archived || session.configuration.workspace != boundary.workspace - || session.configuration.permissions != boundary.permissions + || session.configuration.permissions + != if self.background_kind == BackgroundKind::Discovery { + super::super::discovery::permissions() + } else { + boundary.permissions.clone() + } || session.configuration.binding != boundary.binding || session.configuration.boundary_revision != if session.id == boundary.session_id { @@ -50,11 +55,11 @@ impl Owner { return Ok(()); }; if !match &basis.background_kind { - BackgroundKind::Schedule => self.settings.scheduled_tasks, + BackgroundKind::Schedule | BackgroundKind::Discovery => self.settings.scheduled_tasks, BackgroundKind::Goal(_) => self.settings.goals, } { return Err(match &basis.background_kind { - BackgroundKind::Schedule => "scheduler_disabled", + BackgroundKind::Schedule | BackgroundKind::Discovery => "scheduler_disabled", BackgroundKind::Goal(_) => "goal_disabled", } .into()); @@ -63,6 +68,9 @@ impl Owner { self.check_goal_background(&boundary.session_id, *id) .await? } + if basis.background_kind == BackgroundKind::Discovery { + self.validate_discovery_background().await?; + } let source = self.session(&boundary.session_id).await?; if source.archived || source.configuration.workspace != boundary.workspace @@ -86,12 +94,31 @@ impl Owner { target, model, thinking_level, + } + | ScheduledEffect::Discovery { + target, + model, + thinking_level, } => (target, model, thinking_level), }; + if matches!(plan.task.effect, ScheduledEffect::Discovery { .. }) + && !self.bot.is(&target.session_id) + { + return Err("discovery_requires_bot".into()); + } Ok(Approval::Execution { basis: Box::new( - self.approve_background(target, model.clone(), *thinking, BackgroundKind::Schedule) - .await?, + self.approve_background( + target, + model.clone(), + *thinking, + if matches!(plan.task.effect, ScheduledEffect::Discovery { .. }) { + BackgroundKind::Discovery + } else { + BackgroundKind::Schedule + }, + ) + .await?, ), }) } @@ -142,10 +169,15 @@ impl Owner { .map_err(|_| "agent_preferences_unavailable")?; let language = self.resolve_language(&snapshot); drop(runs); - let selected_catalog = self - .mcp_projects - .capture(&session.configuration.workspace) - .await?; + let selected_catalog = if kind == BackgroundKind::Discovery { + self.mcp_projects.base.clone() + } else { + let catalog = self + .mcp_projects + .capture(&session.configuration.workspace) + .await?; + self.mcp_projects.with_client(id, catalog)? + }; let runs = self.runs.lock().await; if runs.closing || self.stop.is_cancelled() { return Err("agent_closed".into()); @@ -155,8 +187,7 @@ impl Owner { return Err("scheduler_authorization_changed".into()); } - let selected_catalog = self.mcp_projects.with_client(id, selected_catalog)?; - Ok(MessageBasis { + let basis = MessageBasis { mcp_revision: selected_catalog.revision.clone(), client_mcp: selected_catalog.client_mcp, mcp_catalog: Some(selected_catalog.clone()), @@ -173,21 +204,35 @@ impl Owner { background: Some(target.clone()), internal_tools: selected_catalog.definitions.clone(), internal_bindings: selected_catalog.bindings.clone(), - behavior: self.bot_behavior( - &target.session_id, + behavior: if kind == BackgroundKind::Discovery { self.capture_behavior(Some(&session.configuration.workspace), &snapshot, language) + .await? + } else { + self.bot_behavior( + &target.session_id, + self.capture_behavior( + Some(&session.configuration.workspace), + &snapshot, + language, + ) .await?, - &snapshot, - language, - self.admitted_tools(&target.session_id, &session.configuration.tools) - .as_deref(), - ), + &snapshot, + language, + self.admitted_tools(&target.session_id, &session.configuration.tools) + .as_deref(), + ) + }, model: Some(source.accepted), executor: None, workspace: session.configuration.workspace, binding: session.configuration.binding, fingerprint: String::new(), tools: self.tool_basis.clone(), + }; + Ok(if kind == BackgroundKind::Discovery { + self.investigation_basis(&basis) + } else { + basis }) } pub(in crate::owner) fn start_schedules(self: &Arc) { @@ -259,7 +304,10 @@ impl Owner { // A reminder remains durable until the user acknowledges it. // No paint attempt or socket write is described as delivery. if !matches!(fire.effect, ScheduledEffect::Reminder) { - match self.dispatch_scheduled(&fire).await { + match self + .dispatch_scheduled(&fire, plan.task.status == Status::Active) + .await + { Ok(Some(run)) => { plan.settle(run).map_err(reason)?; self.scheduler @@ -288,6 +336,7 @@ impl Owner { async fn dispatch_scheduled( self: &Arc, fire: &maka_scheduler::plan::Fire, + active: bool, ) -> Result, String> { let Some(Approval::Execution { basis }) = &fire.authorization else { return Ok(Some(settlement( @@ -298,10 +347,13 @@ impl Owner { None, ))); }; + if matches!(fire.effect, ScheduledEffect::Discovery { .. }) { + return self.dispatch_discovery(fire, basis, active).await; + } let (source_session, new_session) = match &fire.effect { ScheduledEffect::Session { target, .. } => (&target.session_id, false), ScheduledEffect::AgentRun { target, .. } => (&target.session_id, true), - ScheduledEffect::Reminder => unreachable!(), + ScheduledEffect::Reminder | ScheduledEffect::Discovery { .. } => unreachable!(), }; let identity = hex::encode(Sha256::digest(fire.id.as_bytes())); let session_id = if new_session { diff --git a/plugins/agent/native/src/owner/scheduler/repository.rs b/plugins/agent/native/src/owner/scheduler/repository.rs index 3b869342a80..6bba97aec46 100644 --- a/plugins/agent/native/src/owner/scheduler/repository.rs +++ b/plugins/agent/native/src/owner/scheduler/repository.rs @@ -24,11 +24,11 @@ use maka_plugin_core::operation::RecoveryRef; struct Index { ids: Vec, } -pub(super) struct Saved { +pub(in crate::owner) struct Saved { pub revision: u64, pub plan: Plan, } -pub(super) struct Catalog { +pub(in crate::owner) struct Catalog { pub revision: Option, pub plans: BTreeMap, } @@ -55,7 +55,7 @@ fn value(value: &T) -> Result { .map_err(reason) } impl Scheduler { - pub(super) async fn load(&self) -> Result { + pub(in crate::owner) async fn load(&self) -> Result { let record = self.storage.read("index".into()).await.map_err(reason)?; let revision = record.as_ref().map(|record| record.revision); let index: Index = record diff --git a/plugins/agent/native/src/owner/scheduler/tools.rs b/plugins/agent/native/src/owner/scheduler/tools.rs index 0fb34c69c4b..bc0378750d0 100644 --- a/plugins/agent/native/src/owner/scheduler/tools.rs +++ b/plugins/agent/native/src/owner/scheduler/tools.rs @@ -62,22 +62,28 @@ enum Target { Reminder, Session, AgentRun, + Discovery, } impl Scheduler { pub fn tools(self: &Arc) -> Vec { - vec![ToolRegistration { + let mut tools = vec![ToolRegistration { definition: ToolDefinition { provider: None, name: "ScheduledTask".into(), - description: "Manage this conversation's scheduled tasks. Create saves a paused proposal requiring user approval of its exact execution in Scheduled tasks; model calls cannot grant background access. Session and agent_run use this conversation's workspace and model. Times are Unix milliseconds; timezone is an IANA name. List returns tasks and revisions. Resume requires existing user approval.".into(), + description: "Manage this conversation's scheduled tasks. Create saves a paused proposal requiring user approval of its exact execution in Scheduled tasks; model calls cannot grant background access. Session and agent_run use this conversation's workspace and model. The Bot may propose discovery: a fresh read-only investigation, silent unless it reports a finding. Times are Unix milliseconds; timezone is an IANA name. List returns tasks and revisions. Resume requires existing user approval.".into(), input_schema: serde_json::to_value(schemars::schema_for!(ToolRequest)).expect("task tool schema"), output_schema: None, freeform: None }, nesting: ToolNesting::Nestable, semantics: ToolSemantics::ExclusiveStep, handler: ToolHandler::Prepared(self.clone()), - }] + }]; + tools.push(self.discovery_tool()); + tools } } impl ToolPreparer for Scheduler { fn names(&self) -> Vec { - vec!["ScheduledTask".into()] + vec![ + "ScheduledTask".into(), + super::super::discovery::COMPLETE.into(), + ] } fn prepare( &self, @@ -87,6 +93,9 @@ impl ToolPreparer for Scheduler { cancellation: CancellationToken, ) -> PreparationFuture { let owner = self.link.get().and_then(Weak::upgrade); + if name == super::super::discovery::COMPLETE { + return self.prepare_discovery(input, context, cancellation); + } Box::pin(async move { if name != "ScheduledTask" || owner.is_none() { return Err(ToolRejection::Unavailable); @@ -222,7 +231,16 @@ impl Owner { binding: session.configuration.binding, }; let thinking_level = accepted.thinking_level; - if matches!(effect, Target::Session) { + if matches!(effect, Target::Discovery) { + if !self.bot.is(&context.invocation.session_id) { + return Err("discovery_requires_bot".into()); + } + ScheduledEffect::Discovery { + target, + model, + thinking_level, + } + } else if matches!(effect, Target::Session) { ScheduledEffect::Session { target, model, diff --git a/plugins/agent/native/src/owner/tests.rs b/plugins/agent/native/src/owner/tests.rs index 8dbde888957..b6d66d81f81 100644 --- a/plugins/agent/native/src/owner/tests.rs +++ b/plugins/agent/native/src/owner/tests.rs @@ -91,6 +91,7 @@ mod settings; mod images; +mod discovery_tests; mod imports; mod projects; mod recall_tests; diff --git a/plugins/agent/native/src/owner/tests/discovery_tests.rs b/plugins/agent/native/src/owner/tests/discovery_tests.rs new file mode 100644 index 00000000000..ff8184c8424 --- /dev/null +++ b/plugins/agent/native/src/owner/tests/discovery_tests.rs @@ -0,0 +1,535 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +use super::scheduler_tests::{change, detail, reopen, settled}; +use super::*; +use maka_agent_facts::event::Fact; +use maka_capabilities::storage::{Data, Mutation, Namespace, Store}; + +async fn fixture(listener: &TcpListener, files: bool) -> (Fixture, String) { + let f = Fixture::configured( + &format!("http://{}/v1", listener.local_addr().unwrap()), + None, + None, + files.then(|| service_tools::Setup::files("readText")), + |settings| { + settings.bot = true; + settings.scheduled_tasks = true; + settings.recall = true; + }, + ) + .await; + let bot = f + .mutation(2, "openBot", json!({}), Uuid::new_v4()) + .await + .unwrap()["sessionId"] + .as_str() + .unwrap() + .to_owned(); + (f, bot) +} +async fn create(f: &Fixture, bot: &str, title: &str) -> String { + let boundary = f.owner.session(bot).await.unwrap().configuration; + let created = f.mutation(2, "schedule", json!({"kind":"create","input":{ + "task":{"title":title, "intentBody":format!("Read authorized material for {title}."), + "schedule":{"kind":"interval","everySeconds":3600,"startAt":now()+3_600_000}, + "effect":{"kind":"discovery","target":{"sessionId":bot,"workspace":boundary.workspace, + "permissions":boundary.permissions,"boundaryRevision":boundary.boundary_revision,"binding":boundary.binding}, + "model":f.selection().await,"thinkingLevel":null},"maxFires":null,"expiresAt":null}, + "timezone":"UTC","misfire":"latest"}}), Uuid::new_v4()).await.unwrap(); + assert_eq!(created["kind"], "committed", "{created}"); + created["taskId"].as_str().unwrap().to_owned() +} +fn names(request: &Value) -> Vec<&str> { + request["tools"] + .as_array() + .unwrap() + .iter() + .map(|tool| tool["function"]["name"].as_str().unwrap()) + .collect() +} +async fn finish(socket: &mut TcpStream, report: Value) { + service_tools::named_tool_response(socket, "DiscoveryComplete", report).await; +} +async fn close(f: &Fixture) { + f.app.shutdown().await.unwrap(); + f.config.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn discovery_no_report_is_silent_and_uses_a_read_only_worker_profile() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, true).await; + let id = create(&f, &bot, "Check documents").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + let sent = request(&mut socket).await; + let tools = names(&sent); + assert!( + tools.contains(&"DiscoveryComplete") && tools.contains(&"WorkspaceFile"), + "{tools:?}" + ); + assert!(!tools.iter().any(|name| { + [ + "SendUser", + "sleep", + "ScheduledTask", + "SessionSend", + "SessionCreate", + "Bash", + ] + .contains(name) + })); + assert!(!sent["messages"].to_string().contains("persistent Bot")); + assert_eq!( + detail(&f, &id).await["task"]["runs"], + json!([]), + "admission is not completion" + ); + finish( + &mut socket, + json!({"kind":"no_report","note":"Checked authorized documents; no material change."}), + ) + .await; + let done = settled(&f, &id).await; + assert_eq!(done["task"]["runs"][0]["message"], "No report"); + let worker = done["task"]["runs"][0]["sessionId"].as_str().unwrap(); + let config = f.owner.session(worker).await.unwrap().configuration; + assert_eq!(config.permissions.access, WorkspaceAccess::ReadOnly); + assert!(!config.permissions.request_permissions); + assert_eq!(config.parent_session_id.as_deref(), Some(bot.as_str())); + assert!(f.owner.bot.is(&bot) && !f.owner.bot.is(worker)); + assert_eq!( + f.query("botOutbox", json!({})).await.unwrap()["messages"], + json!([]) + ); + assert!( + tokio::time::timeout(Duration::from_millis(150), listener.accept()) + .await + .is_err() + ); + close(&f).await; + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_reports_the_recorded_result_once_and_reopens_after_lost_scheduler_settlement() { + tokio::time::timeout(Duration::from_secs(75), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (mut f, bot) = fixture(&listener, false).await; + let id = create(&f, &bot, "Source finding").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + let storage = maka_host_runtime::storage::Storage::new( + f.store.clone(), + f.owner.prefix.clone(), + Namespace::new( + "maka.agent.scheduler", + maka_plugin_core::scope::Scope::Profile, + ) + .unwrap(), + ); + let key = format!("task:{id}"); + let frozen = storage + .read(key.clone()) + .await + .unwrap() + .unwrap() + .data + .value() + .unwrap() + .clone(); + finish( + &mut socket, + json!({"kind":"report","text":"Verified discovery-marker from README.md."}), + ) + .await; + let (mut socket, _) = listener.accept().await.unwrap(); + let sent = request(&mut socket).await; + assert!( + sent["messages"] + .to_string() + .contains("Verified discovery-marker from README.md.") + ); + assert!(names(&sent).contains(&"SendUser")); + assert!(!names(&sent).contains(&"DiscoveryComplete")); + service_tools::named_tool_response( + &mut socket, + "SendUser", + json!({"messageId":"discovery-answer","messages":[{"text":"Found discovery-marker."}]}), + ) + .await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + service_tools::named_tool_response(&mut socket, "sleep", json!({})).await; + let done = settled(&f, &id).await; + let run = done["task"]["runs"][0].clone(); + assert_eq!(run["message"], "Report admitted to Bot"); + f.app.shutdown().await.unwrap(); + // Reproduce the durable cut: result and Bot admission succeeded, but + // the scheduler's acknowledgement of its original fire was lost. + let current = storage.read(key.clone()).await.unwrap().unwrap(); + storage + .batch(vec![Mutation { + key, + expected_revision: Some(current.revision), + data: Data::Present(frozen), + }]) + .await + .unwrap(); + reopen(&mut f).await; + let recovered = settled(&f, &id).await; + assert_eq!(recovered["task"]["runs"][0], run); + assert_eq!(recovered["task"]["fireCount"], 1); + let outbox = f.query("botOutbox", json!({})).await.unwrap(); + assert_eq!(outbox["messages"].as_array().unwrap().len(), 1, "{outbox}"); + assert!( + tokio::time::timeout(Duration::from_millis(200), listener.accept()) + .await + .is_err(), + "recovery must not call a model again" + ); + close(&f).await; + storage.finish().await.unwrap(); + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_serializes_investigations_and_starts_a_fresh_session_for_each_fire() { + tokio::time::timeout(Duration::from_secs(60), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + let a = create(&f, &bot, "A").await; + let b = create(&f, &bot, "B").await; + change(&f, "trigger_now", &a, json!({})).await; + let (mut first, _) = listener.accept().await.unwrap(); + request(&mut first).await; + change(&f, "trigger_now", &b, json!({})).await; + assert!( + tokio::time::timeout(Duration::from_millis(200), listener.accept()) + .await + .is_err(), + "second investigation must wait" + ); + finish(&mut first, json!({"kind":"no_report","note":"A checked"})).await; + let (mut second, _) = listener.accept().await.unwrap(); + let sent = request(&mut second).await; + assert!(sent["messages"].to_string().contains("material for B")); + finish(&mut second, json!({"kind":"no_report","note":"B checked"})).await; + let first = settled(&f, &a).await["task"]["runs"][0].clone(); + let second = settled(&f, &b).await["task"]["runs"][0].clone(); + assert_ne!(first["sessionId"], second["sessionId"]); + change(&f, "trigger_now", &a, json!({})).await; + let (mut third, _) = listener.accept().await.unwrap(); + let sent = request(&mut third).await; + assert!( + !sent["messages"].to_string().contains("A checked"), + "new cycle must not copy the previous worker history" + ); + finish( + &mut third, + json!({"kind":"no_report","note":"A checked again"}), + ) + .await; + loop { + let done = detail(&f, &a).await; + if done["task"]["runs"].as_array().unwrap().len() == 2 { + assert_ne!(done["task"]["runs"][0]["sessionId"], first["sessionId"]); + break; + } + tokio::task::yield_now().await; + } + close(&f).await; + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_plain_text_completion_is_failed_not_a_silent_success() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + let id = create(&f, &bot, "Incomplete result").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + response_with_text(&mut socket, "No findings; forgot the completion tool.").await; + let done = settled(&f, &id).await; + assert_eq!(done["task"]["runs"][0]["outcome"], "failed"); + assert_eq!( + done["task"]["runs"][0]["message"], + "Investigation ended without DiscoveryComplete" + ); + assert!( + tokio::time::timeout(Duration::from_millis(150), listener.accept()) + .await + .is_err() + ); + close(&f).await; + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_pause_discards_an_unsent_report_while_the_bot_is_busy() { + tokio::time::timeout(Duration::from_secs(60), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + f.send(&bot).await.unwrap(); + let (mut bot_socket, _) = listener.accept().await.unwrap(); + request(&mut bot_socket).await; + let id = create(&f, &bot, "Cancel pending report").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + finish( + &mut socket, + json!({"kind":"report","text":"Do not deliver this cancelled report."}), + ) + .await; + // Wait for the canonical worker terminal while its report cannot reach the busy Bot. + loop { + let invocation = f + .owner + .scheduler + .investigation + .read() + .unwrap() + .clone() + .unwrap(); + let terminal = f + .store + .session(&invocation.session_id) + .await + .unwrap() + .turn_boundary(&invocation.session_id, &invocation.turn_id) + .await + .unwrap(); + if terminal.is_some_and(|b| { + matches!( + b.state, + maka_event_log::turns::InvocationState::Ended { .. } + ) + }) { + break; + } + tokio::task::yield_now().await; + } + change(&f, "pause", &id, json!({})).await; + let done = settled(&f, &id).await; + assert_eq!(done["task"]["runs"][0]["outcome"], "blocked"); + service_tools::named_tool_response(&mut bot_socket, "sleep", json!({})).await; + assert!( + tokio::time::timeout(Duration::from_millis(200), listener.accept()) + .await + .is_err() + ); + assert_eq!( + f.query("botOutbox", json!({})).await.unwrap()["messages"], + json!([]) + ); + close(&f).await; + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_restart_reconciles_an_interrupted_worker_without_repeating_it() { + tokio::time::timeout(Duration::from_secs(60), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (mut f, bot) = fixture(&listener, false).await; + let id = create(&f, &bot, "Interrupted investigation").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + let original = f + .owner + .scheduler + .investigation + .read() + .unwrap() + .clone() + .unwrap(); + f.app.shutdown().await.unwrap(); + drop(socket); + reopen(&mut f).await; + let done = settled(&f, &id).await; + assert_eq!(done["task"]["runs"][0]["sessionId"], original.session_id); + assert_eq!(done["task"]["runs"][0]["outcome"], "failed"); + let facts = service_tools::facts(&f, &original.session_id).await; + assert_eq!( + facts + .iter() + .filter(|fact| matches!( + fact, + maka_agent_facts::event::Fact::InvocationOpened { .. } + )) + .count(), + 1 + ); + assert!( + tokio::time::timeout(Duration::from_millis(200), listener.accept()) + .await + .is_err() + ); + close(&f).await; + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_one_shot_timer_runs_without_a_manual_trigger() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + let boundary = f.owner.session(&bot).await.unwrap().configuration; + let created = f.mutation(2, "schedule", json!({"kind":"create","input":{ + "task":{"title":"Timed discovery","intentBody":"Check once.", + "schedule":{"kind":"once","runAt":now()+2000}, + "effect":{"kind":"discovery","target":{"sessionId":bot,"workspace":boundary.workspace, + "permissions":boundary.permissions,"boundaryRevision":boundary.boundary_revision,"binding":boundary.binding}, + "model":f.selection().await,"thinkingLevel":null},"maxFires":null,"expiresAt":null}, + "timezone":"UTC","misfire":"latest"}}), Uuid::new_v4()).await.unwrap(); + assert_eq!(created["kind"], "committed", "{created}"); + let (mut socket, _) = listener.accept().await.unwrap(); + let sent = request(&mut socket).await; + assert!(names(&sent).contains(&"DiscoveryComplete")); + finish(&mut socket, json!({"kind":"no_report","note":"Timed check completed."})).await; + let done = settled(&f, created["taskId"].as_str().unwrap()).await; + assert_eq!(done["task"]["status"], "completed"); + assert_eq!(done["task"]["fireCount"], 1); + close(&f).await; + }).await.unwrap(); +} + +#[tokio::test] +async fn discovery_model_proposals_require_the_bot_and_human_approval() { + tokio::time::timeout(Duration::from_secs(60), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + let ordinary = f.create("Ordinary conversation").await; + let proposal = json!({"request":{"mode":"create","title":"Read-only discovery proposal", + "intentBody":"Check authorized documents.","effect":"discovery", + "schedule":{"kind":"interval","everySeconds":3600,"startAt":now()+3_600_000},"timezone":"UTC"}}); + f.send(&ordinary).await.unwrap(); + let (mut socket, _) = listener.accept().await.unwrap(); + let sent = request(&mut socket).await; + assert!(!names(&sent).contains(&"DiscoveryComplete")); + recall_tests::tool(&mut socket, "ScheduledTask", proposal.clone()).await; + let (mut socket, _) = listener.accept().await.unwrap(); + let sent = request(&mut socket).await; + assert!(sent.to_string().contains("discovery_requires_bot")); + response(&mut socket).await; + assert_eq!(f.query("schedules", json!({"kind":"list"})).await.unwrap()["tasks"], json!([])); + + f.send(&bot).await.unwrap(); + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + recall_tests::tool(&mut socket, "ScheduledTask", proposal).await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + let page = f.query("schedules", json!({"kind":"list"})).await.unwrap(); + let task = &page["tasks"][0]; + let id = task["id"].as_str().unwrap(); + assert_eq!(task["status"], "paused"); + assert_eq!(task["approved"], false); + recall_tests::tool(&mut socket, "ScheduledTask", json!({"request":{"mode":"resume","taskId":id,"revision":task["revision"]}})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + assert!(request(&mut socket).await.to_string().contains("scheduler_user_approval_required")); + service_tools::named_tool_response(&mut socket, "sleep", json!({})).await; + assert_eq!(change(&f, "approve", id, json!({})).await["kind"], "committed"); + change(&f, "trigger_now", id, json!({})).await; + let (mut worker, _) = listener.accept().await.unwrap(); + assert!(names(&request(&mut worker).await).contains(&"DiscoveryComplete")); + finish(&mut worker, json!({"kind":"no_report","note":"Checked after approval."})).await; + assert_eq!(settled(&f, id).await["task"]["runs"][0]["outcome"], "ok"); + close(&f).await; + }).await.unwrap(); +} + +#[tokio::test] +async fn discovery_handoff_keeps_the_original_turn_and_pause_stops_its_successor() { + use maka_agent_facts::handoff::HandoffIntent; + tokio::time::timeout(Duration::from_secs(90), async { + for action in ["complete", "pause", "route_changed"] { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (mut f, bot) = fixture(&listener, true).await; + std::fs::write(f.workspace.join("discovery.txt"), "verified handoff-marker").unwrap(); + let id = create(&f, &bot, "Transfer investigation").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut first, _) = listener.accept().await.unwrap(); + request(&mut first).await; + let original = f.owner.scheduler.investigation.read().unwrap().clone().unwrap(); + let control = f.owner.runs.lock().await.controls[&original.session_id].clone(); + let reservation = control.handoff.as_ref().unwrap().reserve(HandoffIntent { + handoff_id:"discovery-transfer".into(), host_epoch:"test-host".into(), + root_run_id:original.run_id.clone(), successor_run_id:"discovery-successor-run".into(), + successor_invocation_id:"discovery-successor-invocation".into(), claim_id:"discovery-transfer-claim".into(), + }).unwrap(); + service_tools::named_tool_response(&mut first, "WorkspaceFile", json!({"path":f.workspace.join("discovery.txt")})).await; + let held = reservation.ready().await.unwrap(); + let owner = f.owner.clone(); + let gate = owner.scheduler.gate.lock().await; + held.commit().unwrap().wait().await.unwrap(); + owner.stop.cancel(); + drop(gate); + if action == "route_changed" { + selection::edit(&f, |changes| { + changes.configuration = Some(json!({"baseUrl":"http://127.0.0.1:1/v1"})); + }).await; + } + reopen(&mut f).await; + if action != "route_changed" { + let (mut successor, _) = listener.accept().await.unwrap(); + let sent = request(&mut successor).await; + assert!(sent["messages"].to_string().contains("verified handoff-marker")); + assert!(names(&sent).contains(&"DiscoveryComplete")); + if action == "pause" { + change(&f, "pause", &id, json!({})).await; + } else { + finish(&mut successor, json!({"kind":"no_report","note":"Investigation continued from its recorded read."})).await; + } + } + let done = settled(&f, &id).await; + let expected = match action { "pause" => "blocked", "route_changed" => "failed", _ => "ok" }; + assert_eq!(done["task"]["runs"][0]["outcome"], expected); + assert_eq!(done["task"]["runs"][0]["sessionId"], original.session_id); + assert_eq!(done["task"]["runs"][0]["runId"], original.run_id); + let facts = service_tools::facts(&f, &original.session_id).await; + // Cancellation closes the reserved successor with canonical facts too. + assert_eq!(facts.iter().filter(|fact| matches!(fact, Fact::InvocationOpened { .. })).count(), 2); + if action == "route_changed" { + assert_eq!(facts.iter().filter(|fact| matches!(fact, Fact::ModelRequested { .. })).count(), 1); + } + assert_eq!(facts.iter().filter(|fact| matches!(fact, Fact::ToolDispatched { name, .. } if name == "WorkspaceFile")).count(), 1); + assert!(tokio::time::timeout(Duration::from_millis(150), listener.accept()).await.is_err()); + close(&f).await; + } + }).await.unwrap(); +} diff --git a/plugins/agent/native/src/owner/tests/scheduler_tests.rs b/plugins/agent/native/src/owner/tests/scheduler_tests.rs index ef80b29f947..db9280f75b1 100644 --- a/plugins/agent/native/src/owner/tests/scheduler_tests.rs +++ b/plugins/agent/native/src/owner/tests/scheduler_tests.rs @@ -28,13 +28,13 @@ async fn fixture(base: &str) -> Fixture { fn task(effect: Value) -> Value { json!({"task":{"title":"Scheduled 中😀", "intentBody":"Remember the original request", "schedule":{"kind":"interval","everySeconds":3600,"startAt":now()+3_600_000},"effect":effect,"maxFires":null,"expiresAt":null},"timezone":"Asia/Singapore","misfire":"latest"}) } -async fn detail(f: &Fixture, id: &str) -> Value { +pub(super) async fn detail(f: &Fixture, id: &str) -> Value { f.query("schedules", json!({"kind":"get","taskId":id})) .await .unwrap()["detail"] .clone() } -async fn change(f: &Fixture, kind: &str, id: &str, extra: Value) -> Value { +pub(super) async fn change(f: &Fixture, kind: &str, id: &str, extra: Value) -> Value { let current = detail(f, id).await; let mut input = json!({"kind":kind,"taskId":id,"revision":current["revision"]}); input @@ -45,7 +45,7 @@ async fn change(f: &Fixture, kind: &str, id: &str, extra: Value) -> Value { .await .unwrap() } -async fn settled(f: &Fixture, id: &str) -> Value { +pub(super) async fn settled(f: &Fixture, id: &str) -> Value { loop { let value = detail(f, id).await; if !value["task"]["runs"].as_array().unwrap().is_empty() { @@ -54,7 +54,7 @@ async fn settled(f: &Fixture, id: &str) -> Value { tokio::time::sleep(Duration::from_millis(10)).await; } } -async fn reopen(f: &mut Fixture) { +pub(super) async fn reopen(f: &mut Fixture) { let deployment = f.app.source().unwrap().deployment; f.app.shutdown().await.unwrap(); let mut native = NativeRegistry::default(); diff --git a/plugins/agent/native/src/tools.rs b/plugins/agent/native/src/tools.rs index 09ecf30145d..a85d4914e76 100644 --- a/plugins/agent/native/src/tools.rs +++ b/plugins/agent/native/src/tools.rs @@ -127,6 +127,8 @@ pub(crate) struct Catalog { pub editing: Vec<(String, EditingTools)>, pub host_tools: Vec, pub resource_reads: Vec, + /// Known read operations, identified by their bound contract rather than aliases. + pub discovery_reads: Vec, pub executor: Arc, } @@ -141,6 +143,7 @@ pub(crate) fn catalog( let mut basis = Vec::new(); let mut host_tools = Vec::new(); let mut resource_reads = Vec::new(); + let mut discovery_reads = Vec::new(); let file_contract = maka_workspace_api::files::contract(); let editing: Vec<_> = selected .iter() @@ -187,6 +190,11 @@ pub(crate) fn catalog( }; let contract = binding.contract(); let resource_read = contract == &file_contract && selection.method == "readText"; + if contract == &file_contract + && matches!(selection.method.as_str(), "readText" | "readImage" | "glob" | "grep") + { + discovery_reads.push(selection.name.clone()); + } let supported = if contract == &maka_agent_api::contract::contract() { matches!(selection.method.as_str(), "models" | "list" | "observe" | "history" | "create" | "fork" | @@ -417,6 +425,7 @@ pub(crate) fn catalog( editing, host_tools, resource_reads, + discovery_reads, executor: Arc::new(ApprovalExecutor { services: native_services, }), From 96d87444c0dde0c1b4062da813df44645c562eeb Mon Sep 17 00:00:00 2001 From: Shizuina Date: Mon, 9 Nov 2026 23:53:00 +0800 Subject: [PATCH 02/14] fix(assistant): enforce discovery cancellation and source boundaries Keep cancellation durable across resume and service disablement, and settle only the original investigation Turn. Recheck report authority at admission and restrict Recall to the enabled built-in implementation. Preserve uncertain result writes and allow approved inactive plans to resume. Generated-by: OpenAI Codex --- plugins/agent/native/README.md | 5 + plugins/agent/native/README.zh-CN.md | 4 + plugins/agent/native/src/owner/discovery.rs | 229 +++++++---- .../agent/native/src/owner/notifications.rs | 13 + plugins/agent/native/src/owner/scheduler.rs | 16 +- .../native/src/owner/scheduler/dispatch.rs | 29 +- .../agent/native/src/owner/scheduler/tools.rs | 6 +- .../native/src/owner/tests/discovery_tests.rs | 367 +++++++++++++++++- .../native/src/owner/tests/service_tools.rs | 53 +++ 9 files changed, 642 insertions(+), 80 deletions(-) diff --git a/plugins/agent/native/README.md b/plugins/agent/native/README.md index 0672686951e..63adbac9f5c 100644 --- a/plugins/agent/native/README.md +++ b/plugins/agent/native/README.md @@ -146,6 +146,11 @@ reports; it cannot retract an already admitted Bot message. Restart reconciles t original fire, Turn and receipt instead of creating replacement work. A sealed Host handoff uses its reserved successor; an interrupted terminal run is recorded as failed. +Disabling Bot or scheduling withdraws outstanding investigation work and unsent reports. +Resuming a plan schedules future work without reviving its cancelled fire. Settlement +tracks the investigation's original Turn, so later user work in its Session cannot hold +the investigation slot. Result-write uncertainty remains an unknown tool outcome. + Plans, the active job and completed receipts use the existing scheduler KV namespace; no separate database or lease is introduced. This slice does not import TS discovery data, add shared notebook memory or observe external events. The [discovery tests](src/owner/tests/discovery_tests.rs) diff --git a/plugins/agent/native/README.zh-CN.md b/plugins/agent/native/README.zh-CN.md index 55f486d29f2..e14567a55f4 100644 --- a/plugins/agent/native/README.zh-CN.md +++ b/plugins/agent/native/README.zh-CN.md @@ -138,6 +138,10 @@ Recall 和 `DiscoveryComplete`；不继承协调工具、MCP 或 shell 能力。 Bot 消息不能撤回。重启按原触发、Turn 和回执恢复，不新建替代调查。已封存的 Host 交接沿用预留的继任者；被中断并终结的执行记录为失败。 +停用 Bot 或调度会撤销尚未结束的调查与未发送报告。恢复计划只安排后续工作，不恢复 +已取消的那次触发。结算按原调查 Turn 跟踪，因此用户之后在同一 Session 中开展的工作 +不会继续占据调查位。结果写入不确定时保留工具结果未知状态。 + 计划、当前调查及已完成回执复用调度器 KV 命名空间，不增加独立数据库或租约。 本阶段不导入 TS 主动发现数据、不增加共享记事本或外部事件监听。 [调查测试](src/owner/tests/discovery_tests.rs) 使用真实 Host 和本地模型夹具验证。 diff --git a/plugins/agent/native/src/owner/discovery.rs b/plugins/agent/native/src/owner/discovery.rs index b5462bf7c66..448aca191c3 100644 --- a/plugins/agent/native/src/owner/discovery.rs +++ b/plugins/agent/native/src/owner/discovery.rs @@ -82,6 +82,14 @@ fn completed_key(fire: &str) -> String { fn reason(error: impl std::fmt::Display) -> String { error.to_string() } +fn result_write_error(error: maka_capabilities::storage::StoreError) -> ToolError { + match error { + maka_capabilities::storage::StoreError::OutcomeUnknown(message) => { + ToolError::OutcomeUnknown(message) + } + error => ToolError::Failed(error.to_string()), + } +} fn mutation( key: String, revision: Option, @@ -114,12 +122,17 @@ impl scheduler::Scheduler { job.as_ref().map(|job| job.admission.invocation.clone()); Ok((revision, job)) } - async fn save_investigation(&self, revision: Option, job: &Job) -> Result { + async fn save_investigation( + &self, + revision: Option, + job: &Job, + ) -> Result { let written = self .storage - .batch(vec![mutation(ACTIVE.into(), revision, job)?]) - .await - .map_err(reason)?; + .batch(vec![mutation(ACTIVE.into(), revision, job).map_err( + maka_capabilities::storage::StoreError::Unavailable, + )?]) + .await?; *self.investigation.write().unwrap() = Some(job.admission.invocation.clone()); Ok(written[0].revision) } @@ -172,8 +185,7 @@ impl scheduler::Scheduler { .workers .spawn(async move { captured.complete_discovery(context, report).await }); job.await - .map_err(|e| ToolError::OutcomeUnknown(e.to_string()))? - .map_err(ToolError::Failed)?; + .map_err(|e| ToolError::OutcomeUnknown(e.to_string()))??; Ok(serde_json::json!({"recorded":true}).into()) }) })) @@ -183,7 +195,17 @@ impl scheduler::Scheduler { impl Owner { pub(super) async fn restore_discovery(&self) -> Result<(), String> { - self.scheduler.investigation().await?; + let (revision, job) = self.scheduler.investigation().await?; + if (!self.settings.bot || !self.settings.scheduled_tasks) + && let Some(mut job) = job + && !matches!(job.phase, Phase::Stopping) + { + job.phase = Phase::Stopping; + self.scheduler + .save_investigation(revision, &job) + .await + .map_err(reason)?; + } Ok(()) } @@ -192,11 +214,14 @@ impl Owner { let mut tools = self.discovery_reads.clone(); // Recall preserves its existing source authorization. No MCP, shell, // delegation, approval, or filesystem mutation tool is granted. - for name in ["Recall", "RecallMore", COMPLETE] { - if self.tools.contains(name) { - tools.push(name.into()); + if self.settings.recall { + for name in ["Recall", "RecallMore"] { + if self.tools.contains(name) { + tools.push(name.into()); + } } } + tools.push(COMPLETE.into()); tools.sort(); tools.dedup(); basis.allowed_tools = Some(tools); @@ -212,11 +237,13 @@ impl Owner { || matches!(job.phase, Phase::Stopping) || !catalog.plans.get(&job.task_id).is_some_and(|saved| { saved.plan.task.status == maka_scheduler::task::Status::Active - && saved - .plan - .pending - .as_ref() - .is_some_and(|fire| fire.id == job.fire_id) + && saved.plan.pending.as_ref().is_some_and(|fire| { + fire.id == job.fire_id + && matches!( + fire.authorization, + Some(scheduler::Approval::Execution { .. }) + ) + }) }) { return Err("discovery_cancelled".into()); @@ -228,55 +255,53 @@ impl Owner { &self, context: ToolCallContext, report: Report, - ) -> Result<(), String> { + ) -> Result<(), ToolError> { let _gate = self.scheduler.gate.lock().await; - let (revision, job) = self.scheduler.investigation().await?; - let mut job = job.ok_or("discovery_not_active")?; - let original = &job.admission.invocation; - if context.invocation.session_id != original.session_id - || context.invocation.turn_id != original.turn_id - { - return Err("discovery_wrong_invocation".into()); - } - let store = self.tool_store(&context)?; - let boundary = store - .turn_boundary(&original.session_id, &original.turn_id) - .await - .map_err(reason)? - .ok_or("discovery_execution_unavailable")?; - if boundary.root_invocation() != original - || boundary.invocation != context.invocation - || matches!(boundary.state, InvocationState::Ended { .. }) - { - return Err("discovery_execution_changed".into()); - } - let plan = self.scheduler.load().await?; - if !self.bot.is(&job.bot) - || !plan.plans.get(&job.task_id).is_some_and(|saved| { - saved.plan.task.status == maka_scheduler::task::Status::Active - && saved - .plan - .pending - .as_ref() - .is_some_and(|fire| fire.id == job.fire_id) - }) - { - return Err("discovery_cancelled".into()); - } - let basis: MessageBasis = - serde_json::from_value(job.admission.execution.basis.clone()).map_err(reason)?; - self.validate_background(&basis).await?; - let Phase::Investigating { result } = &mut job.phase else { - return Err("discovery_cancelled".into()); - }; - if let Some(previous) = result { - if *previous != report { - return Err("discovery_result_conflict".into()); + let update = async { + let (revision, job) = self.scheduler.investigation().await?; + let mut job = job.ok_or("discovery_not_active")?; + let original = &job.admission.invocation; + if context.invocation.session_id != original.session_id + || context.invocation.turn_id != original.turn_id + { + return Err("discovery_wrong_invocation".into()); } - return Ok(()); + let store = self.tool_store(&context)?; + let boundary = store + .turn_boundary(&original.session_id, &original.turn_id) + .await + .map_err(reason)? + .ok_or("discovery_execution_unavailable")?; + if boundary.root_invocation() != original + || boundary.invocation != context.invocation + || matches!(boundary.state, InvocationState::Ended { .. }) + { + return Err("discovery_execution_changed".into()); + } + let basis: MessageBasis = + serde_json::from_value(job.admission.execution.basis.clone()).map_err(reason)?; + self.validate_background(&basis).await?; + let Phase::Investigating { result } = &mut job.phase else { + return Err("discovery_cancelled".into()); + }; + if let Some(previous) = result { + if *previous != report { + return Err("discovery_result_conflict".into()); + } + return Ok(None); + } + *result = Some(report); + Ok(Some((revision, job))) } - *result = Some(report); - self.scheduler.save_investigation(revision, &job).await?; + .await + .map_err(ToolError::Failed)?; + let Some((revision, job)) = update else { + return Ok(()); + }; + self.scheduler + .save_investigation(revision, &job) + .await + .map_err(result_write_error)?; self.scheduler.wake.notify_one(); Ok(()) } @@ -350,7 +375,12 @@ impl Owner { admitted_at: fire.scheduled_at as u64, }, }; - revision = Some(self.scheduler.save_investigation(revision, &job).await?); + revision = Some( + self.scheduler + .save_investigation(revision, &job) + .await + .map_err(reason)?, + ); job }; let basis: MessageBasis = @@ -363,14 +393,19 @@ impl Owner { .await .map(Some); } - let allowed = enabled && self.bot.is(&job.bot); + let allowed = enabled && self.settings.bot && self.bot.is(&job.bot); let authorization = self.validate_background(&basis).await; let revoked = matches!(&authorization, Err(error) if matches!(error.as_str(), "scheduler_authorization_changed" | "scheduler_disabled" | "agent_session_unavailable" | "discovery_cancelled")); if !allowed || revoked { if !matches!(job.phase, Phase::Stopping) { job.phase = Phase::Stopping; - revision = Some(self.scheduler.save_investigation(revision, &job).await?); + revision = Some( + self.scheduler + .save_investigation(revision, &job) + .await + .map_err(reason)?, + ); } } else { authorization?; @@ -407,7 +442,12 @@ impl Owner { && !matches!(job.phase, Phase::Stopping) { job.phase = Phase::Stopping; - revision = Some(self.scheduler.save_investigation(revision, &job).await?); + revision = Some( + self.scheduler + .save_investigation(revision, &job) + .await + .map_err(reason)?, + ); } if existing.is_none() { if matches!(job.phase, Phase::Stopping) { @@ -485,9 +525,26 @@ impl Owner { .await?; } let runs = self.runs.lock().await; - let unsettled = runs.busy(worker) || runs.retained_cleanup.contains(worker); + let running = runs + .controls + .get(worker) + .is_some_and(|control| control.invocation.turn_id == invocation.turn_id); + let retained = runs.retained_cleanup.contains(worker); drop(runs); - if unsettled || self.has_parked_invocation(worker).await? { + // A later Turn could already own this ordinary Session. It neither + // extends this investigation nor authorizes cancelling that new work. + let retained = retained + && store + .latest_turn_boundary(worker) + .await + .map_err(reason)? + .is_none_or(|latest| latest.invocation.turn_id == invocation.turn_id); + let parked = store + .activation_checkpoint(worker) + .await + .map_err(reason)? + .is_some_and(|parked| parked.checkpoint.invocation.turn_id == invocation.turn_id); + if running || retained || parked { return Ok(None); } let Some(boundary) = boundary else { @@ -585,11 +642,19 @@ impl Owner { text: text.clone(), notification: Box::new(notification), }; - revision = Some(self.scheduler.save_investigation(revision, &job).await?); + revision = Some( + self.scheduler + .save_investigation(revision, &job) + .await + .map_err(reason)?, + ); let Phase::Delivering { notification, .. } = &job.phase else { unreachable!() }; - if !self.admit_notification(notification).await? { + if !self + .admit_notification_with_source(notification, Some(&basis)) + .await? + { return Ok(None); } self.settle_discovery( @@ -603,7 +668,10 @@ impl Owner { .map(Some) } Phase::Delivering { notification, .. } => { - if !self.admit_notification(notification).await? { + if !self + .admit_notification_with_source(notification, Some(&basis)) + .await? + { return Ok(None); } self.settle_discovery( @@ -720,3 +788,24 @@ impl Owner { Ok(run) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn result_write_preserves_commit_uncertainty() { + use maka_capabilities::storage::StoreError as StorageError; + assert!(matches!( + result_write_error(StorageError::OutcomeUnknown("acknowledgement lost".into())), + ToolError::OutcomeUnknown(_) + )); + assert!(matches!( + result_write_error(StorageError::Conflict { + expected: "1".into(), + actual: "2".into() + }), + ToolError::Failed(_) + )); + } +} diff --git a/plugins/agent/native/src/owner/notifications.rs b/plugins/agent/native/src/owner/notifications.rs index 3244c1784dc..f91333a9d0d 100644 --- a/plugins/agent/native/src/owner/notifications.rs +++ b/plugins/agent/native/src/owner/notifications.rs @@ -217,6 +217,14 @@ impl Owner { pub(crate) async fn admit_notification( self: &Arc, admission: &PendingMessageAdmission, + ) -> Result { + self.admit_notification_with_source(admission, None).await + } + + pub(super) async fn admit_notification_with_source( + self: &Arc, + admission: &PendingMessageAdmission, + source: Option<&MessageBasis>, ) -> Result { let session = &admission.invocation.session_id; let message = &admission.source.message.message_id; @@ -288,6 +296,11 @@ impl Owner { if original.is_some() { return Ok(true); } + // Recheck after acquiring admission and reserving the recipient. + // Its metadata can no longer change during this new admission. + if let Some(source) = source { + self.validate_background(source).await?; + } let current = match self.session(session).await { Ok(session) => session, Err(error) if error == "agent_session_unavailable" => return Ok(false), diff --git a/plugins/agent/native/src/owner/scheduler.rs b/plugins/agent/native/src/owner/scheduler.rs index aca4007fa7d..8d3105c4848 100644 --- a/plugins/agent/native/src/owner/scheduler.rs +++ b/plugins/agent/native/src/owner/scheduler.rs @@ -39,6 +39,20 @@ type Plan = TaskPlan; pub(super) enum Approval { Reminder, Execution { basis: Box }, + CancelledExecution { basis: Box }, +} +fn pause_plan(plan: &Plan, at: i64) -> Plan { + let mut paused = plan.pause(at); + if let Some(fire) = &mut paused.pending + && matches!(fire.effect, ScheduledEffect::Discovery { .. }) + { + // The original fire stays cancelled even if the plan is immediately resumed. + fire.authorization = fire.authorization.take().map(|approval| match approval { + Approval::Execution { basis } => Approval::CancelledExecution { basis }, + other => other, + }); + } + paused } pub(crate) struct Scheduler { pub(super) storage: Arc, @@ -312,7 +326,7 @@ impl Owner { } next } - ScheduleChange::Pause { .. } => plan()?.pause(at), + ScheduleChange::Pause { .. } => pause_plan(plan()?, at), ScheduleChange::Resume { .. } => { let mut next = plan()?.resume(at).map_err(reason)?; next.authorization = Some(self.approve_schedule(&next).await?); diff --git a/plugins/agent/native/src/owner/scheduler/dispatch.rs b/plugins/agent/native/src/owner/scheduler/dispatch.rs index 12030aef4f7..7cfd834dfa5 100644 --- a/plugins/agent/native/src/owner/scheduler/dispatch.rs +++ b/plugins/agent/native/src/owner/scheduler/dispatch.rs @@ -50,12 +50,23 @@ impl Owner { pub(in crate::owner) async fn validate_background( &self, basis: &MessageBasis, + ) -> Result<(), String> { + self.validate_background_source(basis).await?; + if basis.background_kind == BackgroundKind::Discovery { + self.validate_discovery_background().await?; + } + Ok(()) + } + pub(in crate::owner) async fn validate_background_source( + &self, + basis: &MessageBasis, ) -> Result<(), String> { let Some(boundary) = &basis.background else { return Ok(()); }; if !match &basis.background_kind { - BackgroundKind::Schedule | BackgroundKind::Discovery => self.settings.scheduled_tasks, + BackgroundKind::Schedule => self.settings.scheduled_tasks, + BackgroundKind::Discovery => self.settings.scheduled_tasks && self.settings.bot, BackgroundKind::Goal(_) => self.settings.goals, } { return Err(match &basis.background_kind { @@ -68,8 +79,9 @@ impl Owner { self.check_goal_background(&boundary.session_id, *id) .await? } - if basis.background_kind == BackgroundKind::Discovery { - self.validate_discovery_background().await?; + if basis.background_kind == BackgroundKind::Discovery && !self.bot.is(&boundary.session_id) + { + return Err("discovery_cancelled".into()); } let source = self.session(&boundary.session_id).await?; if source.archived @@ -102,7 +114,7 @@ impl Owner { } => (target, model, thinking_level), }; if matches!(plan.task.effect, ScheduledEffect::Discovery { .. }) - && !self.bot.is(&target.session_id) + && (!self.settings.bot || !self.bot.is(&target.session_id)) { return Err("discovery_requires_bot".into()); } @@ -338,7 +350,9 @@ impl Owner { fire: &maka_scheduler::plan::Fire, active: bool, ) -> Result, String> { - let Some(Approval::Execution { basis }) = &fire.authorization else { + let Some(Approval::Execution { basis } | Approval::CancelledExecution { basis }) = + &fire.authorization + else { return Ok(Some(settlement( fire, Outcome::Blocked, @@ -348,6 +362,11 @@ impl Owner { ))); }; if matches!(fire.effect, ScheduledEffect::Discovery { .. }) { + let active = active + && !matches!( + fire.authorization, + Some(Approval::CancelledExecution { .. }) + ); return self.dispatch_discovery(fire, basis, active).await; } let (source_session, new_session) = match &fire.effect { diff --git a/plugins/agent/native/src/owner/scheduler/tools.rs b/plugins/agent/native/src/owner/scheduler/tools.rs index bc0378750d0..c69c1840321 100644 --- a/plugins/agent/native/src/owner/scheduler/tools.rs +++ b/plugins/agent/native/src/owner/scheduler/tools.rs @@ -232,7 +232,7 @@ impl Owner { }; let thinking_level = accepted.thinking_level; if matches!(effect, Target::Discovery) { - if !self.bot.is(&context.invocation.session_id) { + if !self.settings.bot || !self.bot.is(&context.invocation.session_id) { return Err("discovery_requires_bot".into()); } ScheduledEffect::Discovery { @@ -297,13 +297,13 @@ impl Owner { return Err("scheduler_task_outside_conversation".into()); } let plan = match input { - Input::Pause { .. } => Some(saved.plan.pause(now() as i64)), + Input::Pause { .. } => Some(pause_plan(&saved.plan, now() as i64)), Input::Resume { .. } => { let Some(approval) = &saved.plan.authorization else { return Err("scheduler_user_approval_required".into()); }; if let Approval::Execution { basis } = approval { - self.validate_background(basis).await?; + self.validate_background_source(basis).await?; } Some(saved.plan.resume(now() as i64).map_err(reason)?) } diff --git a/plugins/agent/native/src/owner/tests/discovery_tests.rs b/plugins/agent/native/src/owner/tests/discovery_tests.rs index ff8184c8424..9f8d7af28dc 100644 --- a/plugins/agent/native/src/owner/tests/discovery_tests.rs +++ b/plugins/agent/native/src/owner/tests/discovery_tests.rs @@ -44,7 +44,7 @@ async fn fixture(listener: &TcpListener, files: bool) -> (Fixture, String) { .to_owned(); (f, bot) } -async fn create(f: &Fixture, bot: &str, title: &str) -> String { +pub(super) async fn create(f: &Fixture, bot: &str, title: &str) -> String { let boundary = f.owner.session(bot).await.unwrap().configuration; let created = f.mutation(2, "schedule", json!({"kind":"create","input":{ "task":{"title":title, "intentBody":format!("Read authorized material for {title}."), @@ -533,3 +533,368 @@ async fn discovery_handoff_keeps_the_original_turn_and_pause_stops_its_successor } }).await.unwrap(); } + +async fn dispatch_once(f: &Fixture, id: &str) -> Option { + let _gate = f.owner.scheduler.gate.lock().await; + let catalog = f.owner.scheduler.load().await.unwrap(); + let plan = &catalog.plans[id].plan; + let fire = plan.pending.as_ref().unwrap(); + let Some( + crate::owner::scheduler::Approval::Execution { basis } + | crate::owner::scheduler::Approval::CancelledExecution { basis }, + ) = fire.authorization.as_ref() + else { + panic!("missing approval") + }; + f.owner + .dispatch_discovery( + fire, + basis, + plan.task.status == maka_scheduler::task::Status::Active, + ) + .await + .unwrap() +} + +#[tokio::test] +async fn discovery_settlement_does_not_wait_for_a_later_worker_turn() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + f.send(&bot).await.unwrap(); + let (mut bot_socket, _) = listener.accept().await.unwrap(); + request(&mut bot_socket).await; + let first = create(&f, &bot, "First investigation").await; + let second = create(&f, &bot, "Second investigation").await; + change(&f, "trigger_now", &first, json!({})).await; + let (mut worker, _) = listener.accept().await.unwrap(); + request(&mut worker).await; + let original = f + .owner + .scheduler + .investigation + .read() + .unwrap() + .clone() + .unwrap(); + assert_eq!( + f.send(&original.session_id).await.unwrap()["kind"], + "queued" + ); + finish( + &mut worker, + json!({"kind":"report","text":"A recorded finding."}), + ) + .await; + let (mut followup, _) = listener.accept().await.unwrap(); + request(&mut followup).await; + assert_ne!( + f.owner.runs.lock().await.controls[&original.session_id] + .invocation + .turn_id, + original.turn_id + ); + let producer = f.owner.producer_gate.lock().await; + change(&f, "pause", &first, json!({})).await; + change(&f, "trigger_now", &second, json!({})).await; + let settled = dispatch_once(&f, &first) + .await + .expect("only the original Turn owns the investigation slot"); + assert_eq!(settled.outcome, maka_scheduler::task::Outcome::Blocked); + drop(producer); + let (mut next, _) = listener.accept().await.unwrap(); + assert!( + request(&mut next).await["messages"] + .to_string() + .contains("Second investigation") + ); + finish( + &mut next, + json!({"kind":"no_report","note":"Second investigation finished."}), + ) + .await; + // The independent follow-up is still running and was not cancelled. + assert!( + f.owner + .runs + .lock() + .await + .controls + .contains_key(&original.session_id) + ); + response(&mut followup).await; + service_tools::named_tool_response(&mut bot_socket, "sleep", json!({})).await; + super::scheduler_tests::settled(&f, &second).await; + close(&f).await; + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_disabled_bot_blocks_an_existing_approved_schedule() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (mut f, bot) = fixture(&listener, false).await; + let id = create(&f, &bot, "Approved before disablement").await; + let deployment = f.app.source().unwrap().deployment; + f.app.shutdown().await.unwrap(); + let (source, packages, _, _) = f + .store + .load_application(deployment) + .await + .unwrap() + .unwrap() + .into_parts(); + let mut input = source.input; + input + .instances + .get_mut("provider") + .unwrap() + .configuration + .values + .insert("bot".into(), json!(false)); + f.store + .commit_application( + deployment, + source.reference.revision, + input, + source.remote_exports, + packages, + ) + .await + .unwrap(); + reopen(&mut f).await; + assert_eq!( + f.mutation(2, "openBot", json!({}), Uuid::new_v4()) + .await + .unwrap()["reason"], + "disabled" + ); + change(&f, "trigger_now", &id, json!({})).await; + let result = tokio::time::timeout(Duration::from_secs(10), settled(&f, &id)) + .await + .expect("disabled discovery must settle without calling a model"); + assert_eq!(result["task"]["runs"][0]["outcome"], "blocked"); + assert!( + tokio::time::timeout(Duration::from_millis(150), listener.accept()) + .await + .is_err() + ); + close(&f).await; + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_model_can_resume_an_approved_plan_without_a_pending_fire() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + f.send(&bot).await.unwrap(); + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + recall_tests::tool(&mut socket, "ScheduledTask", json!({"request":{"mode":"create","title":"Approved proposal", + "intentBody":"Read documents.","effect":"discovery", + "schedule":{"kind":"interval","everySeconds":3600,"startAt":now()+3_600_000},"timezone":"UTC"}})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + let page = f.query("schedules", json!({"kind":"list"})).await.unwrap(); + let id = page["tasks"][0]["id"].as_str().unwrap(); + change(&f, "approve", id, json!({})).await; + change(&f, "pause", id, json!({})).await; + let task = detail(&f, id).await; + assert_eq!(task["approved"], true); + assert_eq!(task["pendingFire"], Value::Null); + recall_tests::tool(&mut socket, "ScheduledTask", json!({"request":{"mode":"resume","taskId":id,"revision":task["revision"]}})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + assert_eq!(detail(&f, id).await["task"]["status"], "active"); + service_tools::named_tool_response(&mut socket, "sleep", json!({})).await; + close(&f).await; + }).await.unwrap(); +} + +#[tokio::test] +async fn discovery_reenable_does_not_deliver_an_unsent_result_from_before_disablement() { + tokio::time::timeout(Duration::from_secs(60), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (mut f, bot) = fixture(&listener, false).await; + f.send(&bot).await.unwrap(); + let (mut bot_socket, _) = listener.accept().await.unwrap(); + request(&mut bot_socket).await; + let id = create(&f, &bot, "Do not revive an old report").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut worker, _) = listener.accept().await.unwrap(); + request(&mut worker).await; + let original = f + .owner + .scheduler + .investigation + .read() + .unwrap() + .clone() + .unwrap(); + finish( + &mut worker, + json!({"kind":"report","text":"Old unsent report."}), + ) + .await; + loop { + if f.session(&original.session_id) + .await + .turn_boundary(&original.session_id, &original.turn_id) + .await + .unwrap() + .is_some_and(|b| { + matches!( + b.state, + maka_event_log::turns::InvocationState::Ended { .. } + ) + }) + { + break; + } + tokio::task::yield_now().await; + } + for enabled in [false, true] { + let deployment = f.app.source().unwrap().deployment; + f.app.shutdown().await.unwrap(); + let (source, packages, _, _) = f + .store + .load_application(deployment) + .await + .unwrap() + .unwrap() + .into_parts(); + let mut input = source.input; + input + .instances + .get_mut("provider") + .unwrap() + .configuration + .values + .insert("scheduledTasks".into(), json!(enabled)); + f.store + .commit_application( + deployment, + source.reference.revision, + input, + source.remote_exports, + packages, + ) + .await + .unwrap(); + reopen(&mut f).await; + } + assert_eq!( + settled(&f, &id).await["task"]["runs"][0]["outcome"], + "blocked" + ); + assert!( + tokio::time::timeout(Duration::from_millis(150), listener.accept()) + .await + .is_err() + ); + assert_eq!( + f.query("botOutbox", json!({})).await.unwrap()["messages"], + json!([]) + ); + close(&f).await; + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn discovery_report_rechecks_permissions_after_waiting_for_bot_admission() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + let id = create(&f, &bot, "Revocable investigation").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut worker, _) = listener.accept().await.unwrap(); + request(&mut worker).await; + let ordered = f.owner.runs.ordered(&bot).await; + finish(&mut worker, json!({"kind":"report","text":"Finding under an old grant."})).await; + loop { + let active = f.owner.scheduler.storage.read("discovery/active".into()).await.unwrap().unwrap(); + if active.data.value().unwrap()["phase"]["kind"] == "delivering" { break; } + tokio::task::yield_now().await; + } + let previous = f.owner.session(&bot).await.unwrap(); + let updated = f.mutation(2, "metadata", json!({ + "sessionId":bot, "expectedRevision":previous.revision, "title":previous.configuration.title,"labels":[], + "permissions":{"settings":{"access":"read_only","requestPermissions":false},"revokeGrants":true} + }), Uuid::new_v4()).await.unwrap(); + assert_eq!(updated["kind"], "committed"); + drop(ordered); + let result = tokio::time::timeout(Duration::from_secs(10), settled(&f, &id)).await.expect("revoked report must settle without waking the Bot"); + assert_eq!(result["task"]["runs"][0]["outcome"], "blocked"); + assert!(tokio::time::timeout(Duration::from_millis(150), listener.accept()).await.is_err()); + close(&f).await; + }).await.unwrap(); +} + +#[tokio::test] +async fn discovery_pause_then_resume_still_cancels_the_original_fire() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (f, bot) = fixture(&listener, false).await; + let id = create(&f, &bot, "Pause original fire").await; + change(&f, "trigger_now", &id, json!({})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + request(&mut socket).await; + let original = f + .owner + .scheduler + .investigation + .read() + .unwrap() + .clone() + .unwrap(); + let producer = f.owner.producer_gate.lock().await; + change(&f, "pause", &id, json!({})).await; + change(&f, "resume", &id, json!({})).await; + drop(producer); + let result = tokio::time::timeout(Duration::from_secs(10), settled(&f, &id)) + .await + .expect("resume must not undo cancellation of the original fire"); + assert_eq!(result["task"]["runs"][0]["outcome"], "blocked"); + assert_eq!(result["task"]["status"], "active"); + change(&f, "trigger_now", &id, json!({})).await; + let (mut next, _) = listener.accept().await.unwrap(); + request(&mut next).await; + assert_ne!( + f.owner + .scheduler + .investigation + .read() + .unwrap() + .as_ref() + .unwrap() + .session_id, + original.session_id + ); + finish( + &mut next, + json!({"kind":"no_report","note":"New fire completed."}), + ) + .await; + loop { + if detail(&f, &id).await["task"]["runs"] + .as_array() + .unwrap() + .len() + == 2 + { + break; + } + tokio::task::yield_now().await; + } + close(&f).await; + }) + .await + .unwrap(); +} diff --git a/plugins/agent/native/src/owner/tests/service_tools.rs b/plugins/agent/native/src/owner/tests/service_tools.rs index 2402e8fe016..b361d9dde49 100644 --- a/plugins/agent/native/src/owner/tests/service_tools.rs +++ b/plugins/agent/native/src/owner/tests/service_tools.rs @@ -166,6 +166,59 @@ pub(super) async fn facts(fixture: &Fixture, session: &str) -> Vec { .collect() } +#[tokio::test] +async fn discovery_excludes_service_aliases_when_builtin_recall_is_disabled() { + tokio::time::timeout(Duration::from_secs(45), async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let counter = Counter::new(Behavior::Returned); + let f = Fixture::configured( + &format!("http://{}/v1", listener.local_addr().unwrap()), + None, + None, + Some(Setup::native(counter.clone(), true).named("Recall")), + |settings| { + settings.bot = true; + settings.scheduled_tasks = true; + settings.recall = false; + }, + ) + .await; + let bot = f + .mutation(2, "openBot", json!({}), Uuid::new_v4()) + .await + .unwrap()["sessionId"] + .as_str() + .unwrap() + .to_owned(); + let id = super::discovery_tests::create(&f, &bot, "Only built-in read tools").await; + super::scheduler_tests::change(&f, "trigger_now", &id, json!({})).await; + let (mut socket, _) = listener.accept().await.unwrap(); + let sent = request(&mut socket).await; + assert!( + !sent["tools"] + .as_array() + .unwrap() + .iter() + .any(|tool| tool["function"]["name"] == "Recall") + ); + named_tool_response( + &mut socket, + "DiscoveryComplete", + json!({"kind":"no_report","note":"No reading source enabled."}), + ) + .await; + assert_eq!( + super::scheduler_tests::settled(&f, &id).await["task"]["runs"][0]["outcome"], + "ok" + ); + assert_eq!(counter.effects.load(Ordering::SeqCst), 0); + f.app.shutdown().await.unwrap(); + f.config.shutdown().await.unwrap(); + }) + .await + .unwrap(); +} + pub(super) async fn stored_facts( fixture: &Fixture, session: &str, From a3d0adcbc0e8ecdbc3b2f05407eaf8f50d5f0c3a Mon Sep 17 00:00:00 2001 From: rick12345 Date: Wed, 11 Nov 2026 00:05:56 +0800 Subject: [PATCH 03/14] test(desktop): follow durable steering semantics Finish each simulated model step before awaiting steering. Verify that Stop targets the current Run while an accepted followup executes once and the next draft stays intact. Keep the existing input, queue, and timeout checks. Generated-by: OpenAI Codex --- .../tests/real_host/desktop_new_session.rs | 32 ++++++++++++++----- 1 file changed, 24 insertions(+), 8 deletions(-) diff --git a/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs b/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs index 657b1d5c43c..8647feb835d 100644 --- a/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs +++ b/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs @@ -1298,6 +1298,8 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { desktop.key("right", false); desktop.artifact(&format!("steering-{theme}-{width_name}-sending")); gate.release(); + // Steering is consumed at the next model step. + original.finish().await; let second = desktop.request(&mut fixture).await; let sent = second.body["messages"].to_string(); assert!( @@ -1310,7 +1312,6 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { "new typing must not join the frozen correction" ); desktop.expect_text("Keep the sidebar compact.").await; - drop(original); desktop.type_text("Summarize the changes afterwards.").await; desktop.right_click("Send message").await; @@ -1396,7 +1397,11 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { .unwrap() ); desktop.type_text("Check keyboard focus too.").await; + gate.arm(); desktop.submit().await; + desktop.held(&gate).await; + gate.release(); + second.finish().await; let third = desktop.request(&mut fixture).await; let sent = third.body["messages"].to_string(); assert!( @@ -1407,7 +1412,6 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { !sent.contains("Summarize the changes afterwards."), "explicit followup waits until the turn ends" ); - drop(second); desktop .type_text("Preserve this draft after stopping.") .await; @@ -1446,27 +1450,39 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { .await; desktop.right_click("Send message").await; desktop.press("Stop response").await; + // Stop names the active Run; an already accepted followup is independent. + let followup = desktop.request(&mut fixture).await; + assert!( + followup.body["messages"] + .to_string() + .contains("Summarize the changes afterwards.") + ); + drop(third); + followup.finish().await; desktop .wait( - "Stop did not end the active turn", + "the queued followup did not settle after Stop", fixture.idle(&desktop.client, &session), ) .await; - drop(third); desktop.press("Focus message composer").await; desktop .expect_text("Preserve this draft after stopping.") .await; assert!( desktop.controls("Retry message").is_empty(), - "a failed message does not force the details panel open" + "stopping a Run must not turn its accepted followup into a failed send" ); - desktop.press("Messages").await; - desktop.control("Retry message").await; + let queue = desktop + .client + .query("agent", "queue", json!({"sessionId":session})) + .await + .unwrap(); + assert_eq!(queue["entries"], json!([])); desktop.press("Focus message composer").await; desktop.key("right", false); desktop.artifact(&format!("steering-{theme}-{width_name}-stopped")); - assert_eq!(fixture.request_count.load(Ordering::SeqCst), 3); + assert_eq!(fixture.request_count.load(Ordering::SeqCst), 4); desktop.close().await; fixture.finish().await; } From 809d5631737cfab97e50006f5d0a024149efa256 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Wed, 11 Nov 2026 00:46:42 +0800 Subject: [PATCH 04/14] test(process): observe Windows fixture before timing drain Acquire the existing child handle before the five-second pipe-drain wait. This separates PowerShell fixture startup from drain timing and preserves cleanup when the wait fails; production process behavior and the drain bound are unchanged. Generated-by: OpenAI Codex --- platform/process/tests/integration/windows_foreground.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/platform/process/tests/integration/windows_foreground.rs b/platform/process/tests/integration/windows_foreground.rs index 1b2009972bc..cbc68b227e0 100644 --- a/platform/process/tests/integration/windows_foreground.rs +++ b/platform/process/tests/integration/windows_foreground.rs @@ -211,13 +211,14 @@ async fn normal_shell_exit_preserves_background_descendants_but_bounds_pipe_drai }), CancellationToken::new(), )); + // Exclude fixture startup from the drain bound and retain cleanup if it times out. + let child = LiveProcess::read(&dir.path().join("child")).await; let result = tokio::time::timeout(Duration::from_secs(5), task) .await .unwrap() .unwrap() .unwrap(); assert!(dir.path().join("child").exists(), "{result}"); - let child = LiveProcess::read(&dir.path().join("child")).await; assert_eq!(result["status"], "completed"); assert_eq!(result["output"]["stdout"], "done"); assert_eq!(result["output"]["stdoutTruncated"], true); From 7232fc47129382e3c5c9d97e4538c9a14ba3b85d Mon Sep 17 00:00:00 2001 From: rick12345 Date: Wed, 11 Nov 2026 00:55:00 +0800 Subject: [PATCH 05/14] fix(agent): wake queued goals after releasing admission gate The queue consumer uses try_lock for Goal admission, so an enqueue wake can be consumed while the producer still owns the gate. Notify after releasing it when the persisted schedule still has ready work; keep the existing authorization and ordering checks. Generated-by: OpenAI Codex --- .../agent/native/src/owner/goal/dispatch.rs | 31 ++++++++++++------- 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/plugins/agent/native/src/owner/goal/dispatch.rs b/plugins/agent/native/src/owner/goal/dispatch.rs index 6742fe3a49b..85fad1b966b 100644 --- a/plugins/agent/native/src/owner/goal/dispatch.rs +++ b/plugins/agent/native/src/owner/goal/dispatch.rs @@ -69,7 +69,7 @@ impl Owner { } async fn tick_goals(self: &Arc) -> Result<(Option, Vec), String> { use maka_goal::goal::Scheduled; - let _gate = self.goals.gate.lock().await; + let gate = self.goals.gate.lock().await; let index = self.goals.repository.schedule().await.map_err(reason)?; let waiting = index .sessions @@ -111,17 +111,24 @@ impl Owner { !index.sessions.is_empty() || !errors.is_empty(), Ordering::Release, ); - Ok(( - index - .sessions - .values() - .filter_map(|entry| match entry { - Scheduled::Due { at } if *at > now() => Some(*at), - _ => None, - }) - .min(), - errors, - )) + let admission_ready = index + .sessions + .values() + .any(|entry| matches!(entry, Scheduled::Ready)); + let deadline = index + .sessions + .values() + .filter_map(|entry| match entry { + Scheduled::Due { at } if *at > now() => Some(*at), + _ => None, + }) + .min(); + // Queue admission uses try_lock on this gate and can miss the enqueue wake. + drop(gate); + if admission_ready { + self.wake_notify.notify_one(); + } + Ok((deadline, errors)) } async fn tick_goal( self: &Arc, From 50293516dc797a6bc9a9595a284f806219416057 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Wed, 11 Nov 2026 01:28:31 +0800 Subject: [PATCH 06/14] refactor(discovery): limit A1 changes to scheduled investigations Withdraw independent engine, Goal, Bot admission, inventory and platform-test repairs. Keep discovery recovery and authorization coverage, and narrow fixture cleanup to report delivery. Generated-by: OpenAI Codex --- .../tests/integration/windows_foreground.rs | 3 +- plugins/agent/engine/src/driver/compaction.rs | 79 +++++++++---------- plugins/agent/native/src/owner.rs | 6 -- plugins/agent/native/src/owner/access.rs | 5 +- plugins/agent/native/src/owner/bot.rs | 1 - .../driver/tests/model_recovery/discovery.rs | 62 ++------------- .../agent/native/src/owner/driver_assembly.rs | 11 +-- .../agent/native/src/owner/goal/dispatch.rs | 31 +++----- plugins/agent/tool-catalog/src/catalog.rs | 3 +- products/maka/cli/DEPENDENCIES.rust.tsv | 1 + .../tests/real_host/desktop_new_session.rs | 32 ++------ 11 files changed, 68 insertions(+), 166 deletions(-) diff --git a/platform/process/tests/integration/windows_foreground.rs b/platform/process/tests/integration/windows_foreground.rs index cbc68b227e0..1b2009972bc 100644 --- a/platform/process/tests/integration/windows_foreground.rs +++ b/platform/process/tests/integration/windows_foreground.rs @@ -211,14 +211,13 @@ async fn normal_shell_exit_preserves_background_descendants_but_bounds_pipe_drai }), CancellationToken::new(), )); - // Exclude fixture startup from the drain bound and retain cleanup if it times out. - let child = LiveProcess::read(&dir.path().join("child")).await; let result = tokio::time::timeout(Duration::from_secs(5), task) .await .unwrap() .unwrap() .unwrap(); assert!(dir.path().join("child").exists(), "{result}"); + let child = LiveProcess::read(&dir.path().join("child")).await; assert_eq!(result["status"], "completed"); assert_eq!(result["output"]["stdout"], "done"); assert_eq!(result["output"]["stdoutTruncated"], true); diff --git a/plugins/agent/engine/src/driver/compaction.rs b/plugins/agent/engine/src/driver/compaction.rs index 2492a9ab61b..bb768862c8f 100644 --- a/plugins/agent/engine/src/driver/compaction.rs +++ b/plugins/agent/engine/src/driver/compaction.rs @@ -21,7 +21,6 @@ use crate::{ Engine, RunError, reconcile::{CompactionLevel, ModelLevel}, }; -use futures_util::future::BoxFuture; use maka_agent_facts::{ accounting::Outcome, context::{ @@ -59,49 +58,43 @@ fn compaction_timing(evidence: &RequestContextEvidence) -> (bool, bool) { } impl Engine { - // Keep the recovery projection future out of the driver's debug poll frame. - pub(super) fn recover_legacy_compaction<'a>( - &'a self, - claim: &'a DriverClaim, - view: &'a RunView, + pub(super) async fn recover_legacy_compaction( + &self, + claim: &DriverClaim, + view: &RunView, at: UnixMillis, - ) -> BoxFuture<'a, Result> { - Box::pin(async move { - let Some((blocker, _)) = view.blockers.iter().find(|(_, blocker)| { - blocker.resolved_by.is_none() - && blocker.decision.is_none() - && blocker.reason.is_legacy_compaction_lookup() - }) else { - return Ok(false); - }; - if view.stopped - || view.has_unsettled_recovery() - || view.recovery_failures().count() != 1 - { - return Ok(false); - } - let Ok(level) = self.model_level(view).await else { - // An invalid or missing summary still fails with the recorded diagnostic. - return Ok(false); - }; - let (CompactionLevel::Candidate { summary, .. } | CompactionLevel::Adopted { summary }) = - level.compaction - else { - return Ok(false); - }; - let evidence = ContentBlob::json(&summary)?; - self.publish( - Some(claim), - view, - vec![Fact::BlockerResolved { - blocker: *blocker, - evidence: evidence.reference().clone(), - }], - vec![evidence], - at, - ) - .await - }) + ) -> Result { + let Some((blocker, _)) = view.blockers.iter().find(|(_, blocker)| { + blocker.resolved_by.is_none() + && blocker.decision.is_none() + && blocker.reason.is_legacy_compaction_lookup() + }) else { + return Ok(false); + }; + if view.stopped || view.has_unsettled_recovery() || view.recovery_failures().count() != 1 { + return Ok(false); + } + let Ok(level) = self.model_level(view).await else { + // An invalid or missing summary still fails with the recorded diagnostic. + return Ok(false); + }; + let (CompactionLevel::Candidate { summary, .. } | CompactionLevel::Adopted { summary }) = + level.compaction + else { + return Ok(false); + }; + let evidence = ContentBlob::json(&summary)?; + self.publish( + Some(claim), + view, + vec![Fact::BlockerResolved { + blocker: *blocker, + evidence: evidence.reference().clone(), + }], + vec![evidence], + at, + ) + .await } pub(super) async fn model_level(&self, view: &RunView) -> Result { diff --git a/plugins/agent/native/src/owner.rs b/plugins/agent/native/src/owner.rs index 84288d74d18..c7690b31a7c 100644 --- a/plugins/agent/native/src/owner.rs +++ b/plugins/agent/native/src/owner.rs @@ -518,12 +518,6 @@ impl Owner { .with_always_visible(&[bot::SEND_USER, bot::SLEEP]) .map_err(|error| error.to_string())?; } - if settings.scheduled_tasks { - catalog.tools = catalog - .tools - .with_always_visible(&[discovery::COMPLETE]) - .map_err(|e| e.to_string())?; - } Ok(catalog) }); let crate::tools::Catalog { diff --git a/plugins/agent/native/src/owner/access.rs b/plugins/agent/native/src/owner/access.rs index f875dbad7fd..8faf36fca80 100644 --- a/plugins/agent/native/src/owner/access.rs +++ b/plugins/agent/native/src/owner/access.rs @@ -353,10 +353,7 @@ impl RunAccess { return Err(ToolError::Failed("Run workspace binding changed".into())); } // Access mode is current configuration; actual effects enforce the Scope's sandbox floor. - current.tools = intersect_tools( - owner.admitted_tools(&self.execution.session_id, ¤t.tools), - self.spec.allowed_tools.clone(), - ); + current.tools = intersect_tools(current.tools, self.spec.allowed_tools.clone()); current.permissions.request_permissions &= self .spec .permission_ceiling diff --git a/plugins/agent/native/src/owner/bot.rs b/plugins/agent/native/src/owner/bot.rs index e05996b31a4..a719f767405 100644 --- a/plugins/agent/native/src/owner/bot.rs +++ b/plugins/agent/native/src/owner/bot.rs @@ -782,7 +782,6 @@ impl Owner { return Some( COORDINATION_TOOLS .iter() - .chain([&SEND_USER, &SLEEP]) .filter(|name| self.tools.contains(name)) .map(|name| (*name).to_owned()) .collect(), diff --git a/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs b/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs index 28bec8754e2..4d953d3140b 100644 --- a/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs +++ b/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs @@ -337,30 +337,10 @@ fn output(body: &Value) -> Value { serde_json::from_str(text).unwrap_or(json!({"error":text})) } async fn quiet(f: &Fixture, listener: &TcpListener, socket: &mut TcpStream) { - tool(socket, "sleep", json!({})).await; - let settled = tokio::time::timeout(Duration::from_secs(45), async { - tokio::select! { - request = listener.accept() => { - let (mut socket, _) = request.unwrap(); - let (_, body) = http_fixture::read_request(&mut socket).await; - let messages = body["messages"].as_array().unwrap(); - panic!("sleep requested another model step: {:?}", &messages[messages.len().saturating_sub(3)..]); - }, - () = f.quiescent(&f.bot) => {} - } - }) - .await; - if settled.is_err() { - let owner = f.owner(); - let run = owner - .durable - .latest_run(f.bot.clone()) - .await - .unwrap() - .unwrap(); - let view = owner.durable.frontier(run).await.unwrap(); - panic!("Bot did not settle after sleep: {view:#?}"); - } + finish_text(socket, "Investigation received.").await; + tokio::time::timeout(Duration::from_secs(45), f.quiescent(&f.bot)) + .await + .expect("Bot Run did not settle before Host teardown"); assert!( tokio::time::timeout(Duration::from_millis(100), listener.accept()) .await @@ -730,32 +710,10 @@ async fn discovery_recovery_delivers_an_authorized_notification_once() { } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn bot_controls_survive_admission_and_live_filtering_but_are_not_worker_tools() { +async fn discovery_completion_is_not_offered_to_an_ordinary_session() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let mut f = Fixture::new(&listener).await; - f.setting("toolDiscovery", json!(true)).await; - f.send("Reply once and sleep").await; - let (mut socket, body) = next(&listener).await; - let names: Vec<_> = body["tools"] - .as_array() - .unwrap() - .iter() - .map(|t| t["function"]["name"].as_str().unwrap()) - .collect(); - assert!( - names.contains(&"SendUser") && names.contains(&"sleep") && names.contains(&"tool_search"), - "{names:?}" - ); - tool( - &mut socket, - "SendUser", - json!({"messageId":"bot-control-check","messages":[{"text":"Reply"}]}), - ) - .await; - let (mut socket, _) = next(&listener).await; - assert_eq!(f.outbox().await.as_array().unwrap().len(), 1); - quiet(&f, &listener, &mut socket).await; - let created=f.query("create",json!({"title":"Ordinary control","automaticTitle":false,"labels":[],"tools":[],"workspace":{"kind":"directory","path":f.rig.directory.path().join("workspace")}})).await; + let created=f.query("create",json!({"title":"Ordinary control","automaticTitle":false,"labels":[],"tools":["DiscoveryComplete"],"workspace":{"kind":"directory","path":f.rig.directory.path().join("workspace")}})).await; let session = created["session"]["id"].as_str().unwrap(); f.query( "send", @@ -769,12 +727,7 @@ async fn bot_controls_survive_admission_and_live_filtering_but_are_not_worker_to .flatten() .map(|t| t["function"]["name"].as_str().unwrap()) .collect(); - assert!( - !names.contains(&"SendUser") - && !names.contains(&"sleep") - && !names.contains(&"DiscoveryComplete"), - "{names:?}" - ); + assert!(!names.contains(&"DiscoveryComplete"), "{names:?}"); finish_text(&mut socket, "Ordinary answer").await; tokio::time::timeout(Duration::from_secs(45), f.quiescent(session)) .await @@ -788,7 +741,6 @@ async fn disabled_bot_blocks_an_approved_discovery_without_calling_a_model() { let mut f = Fixture::new(&listener).await; let task = f.create("Disabled Bot").await; f.setting("bot", json!(false)).await; - assert_eq!(f.query("openBot", json!({})).await["reason"], "disabled"); f.change(&task, "trigger_now").await; assert_eq!( f.settled(&task, 1).await["task"]["runs"][0]["outcome"], diff --git a/plugins/agent/native/src/owner/driver_assembly.rs b/plugins/agent/native/src/owner/driver_assembly.rs index b25cb447998..034d0b3aee3 100644 --- a/plugins/agent/native/src/owner/driver_assembly.rs +++ b/plugins/agent/native/src/owner/driver_assembly.rs @@ -215,16 +215,7 @@ impl Owner { }) .map(|definition| definition.name.as_str()) .collect(); - let selected = live_tools.excluding(&excluded); - let always_visible: Vec<_> = self - .tools - .definitions() - .filter(|tool| self.tools.always_visible(&tool.name) && selected.contains(&tool.name)) - .map(|tool| tool.name.as_str()) - .collect(); - selected - .with_always_visible(&always_visible) - .map_err(|error| RunError::InvalidInput(error.to_string())) + Ok(live_tools.excluding(&excluded)) } async fn unavailable_current( diff --git a/plugins/agent/native/src/owner/goal/dispatch.rs b/plugins/agent/native/src/owner/goal/dispatch.rs index 85fad1b966b..6742fe3a49b 100644 --- a/plugins/agent/native/src/owner/goal/dispatch.rs +++ b/plugins/agent/native/src/owner/goal/dispatch.rs @@ -69,7 +69,7 @@ impl Owner { } async fn tick_goals(self: &Arc) -> Result<(Option, Vec), String> { use maka_goal::goal::Scheduled; - let gate = self.goals.gate.lock().await; + let _gate = self.goals.gate.lock().await; let index = self.goals.repository.schedule().await.map_err(reason)?; let waiting = index .sessions @@ -111,24 +111,17 @@ impl Owner { !index.sessions.is_empty() || !errors.is_empty(), Ordering::Release, ); - let admission_ready = index - .sessions - .values() - .any(|entry| matches!(entry, Scheduled::Ready)); - let deadline = index - .sessions - .values() - .filter_map(|entry| match entry { - Scheduled::Due { at } if *at > now() => Some(*at), - _ => None, - }) - .min(); - // Queue admission uses try_lock on this gate and can miss the enqueue wake. - drop(gate); - if admission_ready { - self.wake_notify.notify_one(); - } - Ok((deadline, errors)) + Ok(( + index + .sessions + .values() + .filter_map(|entry| match entry { + Scheduled::Due { at } if *at > now() => Some(*at), + _ => None, + }) + .min(), + errors, + )) } async fn tick_goal( self: &Arc, diff --git a/plugins/agent/tool-catalog/src/catalog.rs b/plugins/agent/tool-catalog/src/catalog.rs index d9dde7bfee0..63401d39807 100644 --- a/plugins/agent/tool-catalog/src/catalog.rs +++ b/plugins/agent/tool-catalog/src/catalog.rs @@ -352,8 +352,7 @@ impl ToolCatalog { }) } - /// Whether discovery keeps this registered tool visible after authorization. - pub fn always_visible(&self, name: &str) -> bool { + pub(crate) fn always_visible(&self, name: &str) -> bool { self.entries .get(name) .is_some_and(|entry| entry.always_visible) diff --git a/products/maka/cli/DEPENDENCIES.rust.tsv b/products/maka/cli/DEPENDENCIES.rust.tsv index 4a386266323..c6e4907d52c 100644 --- a/products/maka/cli/DEPENDENCIES.rust.tsv +++ b/products/maka/cli/DEPENDENCIES.rust.tsv @@ -826,6 +826,7 @@ ratex-layout@0.1.15 X ratex-lexer@0.1.15 X ratex-parser@0.1.15 X ratex-types@0.1.15 X +raw-window-handle@0.6.2 X X X rayon-core@1.13.0 X X rayon@1.12.0 X X ref-cast-impl@1.0.27 X X diff --git a/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs b/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs index 8647feb835d..657b1d5c43c 100644 --- a/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs +++ b/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs @@ -1298,8 +1298,6 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { desktop.key("right", false); desktop.artifact(&format!("steering-{theme}-{width_name}-sending")); gate.release(); - // Steering is consumed at the next model step. - original.finish().await; let second = desktop.request(&mut fixture).await; let sent = second.body["messages"].to_string(); assert!( @@ -1312,6 +1310,7 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { "new typing must not join the frozen correction" ); desktop.expect_text("Keep the sidebar compact.").await; + drop(original); desktop.type_text("Summarize the changes afterwards.").await; desktop.right_click("Send message").await; @@ -1397,11 +1396,7 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { .unwrap() ); desktop.type_text("Check keyboard focus too.").await; - gate.arm(); desktop.submit().await; - desktop.held(&gate).await; - gate.release(); - second.finish().await; let third = desktop.request(&mut fixture).await; let sent = third.body["messages"].to_string(); assert!( @@ -1412,6 +1407,7 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { !sent.contains("Summarize the changes afterwards."), "explicit followup waits until the turn ends" ); + drop(second); desktop .type_text("Preserve this draft after stopping.") .await; @@ -1450,39 +1446,27 @@ async fn desktop_steering_freezes_each_message_and_queue_is_one_shot() { .await; desktop.right_click("Send message").await; desktop.press("Stop response").await; - // Stop names the active Run; an already accepted followup is independent. - let followup = desktop.request(&mut fixture).await; - assert!( - followup.body["messages"] - .to_string() - .contains("Summarize the changes afterwards.") - ); - drop(third); - followup.finish().await; desktop .wait( - "the queued followup did not settle after Stop", + "Stop did not end the active turn", fixture.idle(&desktop.client, &session), ) .await; + drop(third); desktop.press("Focus message composer").await; desktop .expect_text("Preserve this draft after stopping.") .await; assert!( desktop.controls("Retry message").is_empty(), - "stopping a Run must not turn its accepted followup into a failed send" + "a failed message does not force the details panel open" ); - let queue = desktop - .client - .query("agent", "queue", json!({"sessionId":session})) - .await - .unwrap(); - assert_eq!(queue["entries"], json!([])); + desktop.press("Messages").await; + desktop.control("Retry message").await; desktop.press("Focus message composer").await; desktop.key("right", false); desktop.artifact(&format!("steering-{theme}-{width_name}-stopped")); - assert_eq!(fixture.request_count.load(Ordering::SeqCst), 4); + assert_eq!(fixture.request_count.load(Ordering::SeqCst), 3); desktop.close().await; fixture.finish().await; } From a0c8b6ce2c7dede5b6efa410d0b8e798358e0549 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Thu, 12 Nov 2026 17:10:08 +0800 Subject: [PATCH 07/14] refactor(discovery): reuse active fire validation Read the investigation plan directly in restore and tool admission, and fold ordinary Session tool visibility into the existing approval test. Generated-by: OpenAI Codex --- plugins/agent/native/src/owner/discovery.rs | 36 +++++------ .../driver/tests/model_recovery/discovery.rs | 63 +++++-------------- 2 files changed, 29 insertions(+), 70 deletions(-) diff --git a/plugins/agent/native/src/owner/discovery.rs b/plugins/agent/native/src/owner/discovery.rs index 4b2d863468d..4c4e16c4045 100644 --- a/plugins/agent/native/src/owner/discovery.rs +++ b/plugins/agent/native/src/owner/discovery.rs @@ -148,6 +148,18 @@ impl scheduler::Scheduler { *self.investigation.write().unwrap() = Some(job.worker.clone()); Ok(records[0].revision) } + async fn has_active_fire(&self, job: &Job) -> Result { + Ok(self.change_basis(&job.task_id).await?.is_some_and(|saved| { + saved.plan.task.status == maka_scheduler::task::Status::Active + && saved.plan.pending.as_ref().is_some_and(|fire| { + fire.id == job.fire_id + && matches!( + fire.authorization, + Some(scheduler::Approval::Discovery { .. }) + ) + }) + })) + } pub(super) fn discovery_tool(self: &Arc) -> ToolRegistration { ToolRegistration { definition: ToolDefinition { @@ -212,20 +224,7 @@ impl Owner { return Ok(()); }; let authorized = self.discovery_authorized(&job).await; - let pending = self - .scheduler - .change_basis(&job.task_id) - .await? - .is_some_and(|saved| { - saved.plan.task.status == maka_scheduler::task::Status::Active - && saved.plan.pending.as_ref().is_some_and(|fire| { - fire.id == job.fire_id - && matches!( - fire.authorization, - Some(scheduler::Approval::Discovery { .. }) - ) - }) - }); + let pending = self.scheduler.has_active_fire(&job).await?; if matches!(job.phase, Phase::Stopping) || !pending || authorization_revoked(&authorized) @@ -328,14 +327,7 @@ impl Owner { return Err("discovery_execution_changed".into()); } self.discovery_authorized(job).await?; - let catalog = self.scheduler.load().await?; - if !catalog.plans.get(&job.task_id).is_some_and(|p| { - p.plan.task.status == maka_scheduler::task::Status::Active - && p.plan.pending.as_ref().is_some_and(|f| { - f.id == job.fire_id - && matches!(f.authorization, Some(scheduler::Approval::Discovery { .. })) - }) - }) { + if !self.scheduler.has_active_fire(job).await? { return Err("discovery_cancelled".into()); } Ok(()) diff --git a/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs b/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs index 84a698ec707..624fc6f3556 100644 --- a/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs +++ b/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs @@ -34,9 +34,6 @@ struct Fixture { } impl Fixture { async fn new(listener: &TcpListener) -> Self { - Self::new_bound(listener, false).await - } - async fn new_bound(listener: &TcpListener, project: bool) -> Self { let mut rig = Rig::new().unwrap(); let base = rig.directory.path().to_path_buf(); let mut boot = bootstrap( @@ -60,8 +57,10 @@ impl Fixture { .unwrap() .configuration .values; - std::fs::create_dir_all(base.join("other-workspace")).unwrap(); - config.insert("workspaceRoots".into(),json!([{"label":"source","path":base.join("workspace")},{"label":"destination","path":base.join("other-workspace")} ])); + config.insert( + "workspaceRoots".into(), + json!([{"label":"source","path":base.join("workspace")}]), + ); config.insert("bot".into(), json!(true)); config.insert("scheduledTasks".into(), json!(true)); config.insert("recall".into(), json!(true)); @@ -94,21 +93,7 @@ impl Fixture { let connection=call(client,"settings","create",json!({"expectedCatalogRevision":catalog["revision"],"connection":{"slug":"discovery","name":"Discovery fixture","provider":{"packageId":"maka.providers","entryId":"maka.providers","scope":"profile","name":"ollama"},"configuration":{"baseUrl":boot.base_url},"enabled":true,"enabledModelIds":["llama3.2"],"modelOverrides":{"llama3.2":{"codeMode":false,"apiProtocol":"openai-chat-completions"}}}})).await.unwrap(); assert_eq!(connection["kind"], "committed"); let model = json!({"kind":"model","connectionId":connection["connection"]["connectionId"],"modelId":"llama3.2","thinkingLevel":null}); - let path = base.join("workspace"); - let workspace = if project { - let registered = call( - client, - "agent", - "project", - json!({"kind":"register","path":path,"expected":null}), - ) - .await - .unwrap(); - assert_eq!(registered["kind"], "committed", "{registered}"); - json!({"kind":"project","project":registered["project"]["reference"],"path":path}) - } else { - json!({"kind":"directory","path":path}) - }; + let workspace = json!({"kind":"directory","path":base.join("workspace")}); let opened = call(client, "agent", "openBot", json!({"workspace":workspace})) .await .unwrap(); @@ -716,32 +701,6 @@ async fn discovery_recovery_delivers_an_authorized_notification_once() { f.close().await; } -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn discovery_completion_is_not_offered_to_an_ordinary_session() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let mut f = Fixture::new(&listener).await; - let created=f.query("create",json!({"title":"Ordinary control","automaticTitle":false,"labels":[],"tools":["DiscoveryComplete"],"workspace":{"kind":"directory","path":f.rig.directory.path().join("workspace")}})).await; - let session = created["session"]["id"].as_str().unwrap(); - f.query( - "send", - json!({"sessionId":session,"message":"Hello","backend":f.model,"delivery":"queue"}), - ) - .await; - let (mut socket, body) = next(&listener).await; - let names: Vec<_> = body["tools"] - .as_array() - .into_iter() - .flatten() - .map(|t| t["function"]["name"].as_str().unwrap()) - .collect(); - assert!(!names.contains(&"DiscoveryComplete"), "{names:?}"); - finish_text(&mut socket, "Ordinary answer").await; - tokio::time::timeout(Duration::from_secs(45), f.quiescent(session)) - .await - .expect("ordinary Run did not settle before Host teardown"); - f.close().await; -} - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn disabled_bot_blocks_an_approved_discovery_without_calling_a_model() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -766,10 +725,17 @@ async fn discovery_model_proposals_require_bot_identity_and_human_approval() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let mut f = Fixture::new(&listener).await; let proposal = json!({"request":{"mode":"create","title":"Proposed discovery","intentBody":"Check authorized documents","effect":"discovery","schedule":{"kind":"interval","everySeconds":3600,"startAt":maka_agent_authority::now()+3_600_000},"timezone":"UTC"}}); - let created=f.query("create",json!({"title":"Ordinary proposer","automaticTitle":false,"labels":[],"tools":["ScheduledTask"],"workspace":{"kind":"directory","path":f.rig.directory.path().join("workspace")}})).await; + let created=f.query("create",json!({"title":"Ordinary proposer","automaticTitle":false,"labels":[],"tools":["ScheduledTask","DiscoveryComplete"],"workspace":{"kind":"directory","path":f.rig.directory.path().join("workspace")}})).await; let session = created["session"]["id"].as_str().unwrap(); f.query("send",json!({"sessionId":session,"message":"Propose a discovery","backend":f.model,"delivery":"queue"})).await; - let (mut socket, _) = next(&listener).await; + let (mut socket, body) = next(&listener).await; + assert!( + body["tools"] + .as_array() + .unwrap() + .iter() + .all(|tool| tool["function"]["name"] != "DiscoveryComplete") + ); tool(&mut socket, "ScheduledTask", proposal.clone()).await; let (mut socket, body) = next(&listener).await; assert!( @@ -777,6 +743,7 @@ async fn discovery_model_proposals_require_bot_identity_and_human_approval() { "{body}" ); finish_text(&mut socket, "Cannot propose discovery here").await; + f.quiescent(session).await; assert_eq!( f.query("schedules", json!({"kind":"list"})).await["tasks"], json!([]) From bba89f17369a6982e9b33320fd85e2893277c18d Mon Sep 17 00:00:00 2001 From: rick12345 Date: Thu, 12 Nov 2026 17:13:00 +0800 Subject: [PATCH 08/14] test(discovery): restrict scheduler fixture to exercised methods Generated-by: OpenAI Codex --- plugins/agent/native/src/owner/driver/tests.rs | 4 ---- 1 file changed, 4 deletions(-) diff --git a/plugins/agent/native/src/owner/driver/tests.rs b/plugins/agent/native/src/owner/driver/tests.rs index 04f0321ae5e..9d3331fd067 100644 --- a/plugins/agent/native/src/owner/driver/tests.rs +++ b/plugins/agent/native/src/owner/driver/tests.rs @@ -421,7 +421,6 @@ impl Rig { let methods: &[&str] = if model_test { &[ "create", - "history", "models", "observe", "predictNextMessage", @@ -434,9 +433,6 @@ impl Rig { "botOutbox", "schedule", "schedules", - "metadata", - "project", - "projects", "selectBackend", "stop", ] From 62fac4fb1d21013022c1058e1bfec59df1cc71fc Mon Sep 17 00:00:00 2001 From: rick12345 Date: Thu, 12 Nov 2026 17:19:01 +0800 Subject: [PATCH 09/14] refactor(discovery): keep only the Run in settlement receipts Continue accepting the old receipt shape while avoiding a second durable copy of the worker report. Generated-by: OpenAI Codex --- plugins/agent/native/src/owner/discovery.rs | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/plugins/agent/native/src/owner/discovery.rs b/plugins/agent/native/src/owner/discovery.rs index 4c4e16c4045..23a9a5025bc 100644 --- a/plugins/agent/native/src/owner/discovery.rs +++ b/plugins/agent/native/src/owner/discovery.rs @@ -73,10 +73,9 @@ struct Job { phase: Phase, } #[derive(Serialize, Deserialize)] -#[serde(deny_unknown_fields)] struct Completed { + // Older receipts can also contain a copy of the report. run: Run, - result: Option, } fn completed_key(fire: &str) -> String { format!( @@ -781,16 +780,12 @@ impl Owner { session_id: Some(job.worker.clone()), run_id: job.execution.as_ref().map(|e| e.run_id), }; - let completed = Completed { - run: run.clone(), - result: match &job.phase { - Phase::Investigating { result } => result.clone(), - Phase::Delivering { text, .. } => Some(Report::Report { text: text.clone() }), - Phase::Stopping => None, - }, - }; let writes = vec![ - mutation(completed_key(&fire.id), None, &completed)?, + mutation( + completed_key(&fire.id), + None, + &Completed { run: run.clone() }, + )?, Mutation { key: ACTIVE.into(), expected_revision: revision, From 541d8c8d0f116df594512b8149051a9e429a2980 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Fri, 13 Nov 2026 03:21:32 +0800 Subject: [PATCH 10/14] fix(discovery): keep read tools in the upstream declaration catalog Generated-by: OpenAI Codex --- plugins/agent/native/src/tools.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/plugins/agent/native/src/tools.rs b/plugins/agent/native/src/tools.rs index d0b95f27ff7..f5532d0c93e 100644 --- a/plugins/agent/native/src/tools.rs +++ b/plugins/agent/native/src/tools.rs @@ -285,6 +285,7 @@ pub(crate) fn catalog( editing, host_tools, resource_reads, + discovery_reads, } = catalog_inner(services, selected, additional, None)?; Ok(Catalog { tools: Declarations(tools), @@ -292,6 +293,7 @@ pub(crate) fn catalog( editing, host_tools, resource_reads, + discovery_reads, }) } From df76de58008de384ba0c3cb360e06e9882a76151 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Fri, 13 Nov 2026 03:35:26 +0800 Subject: [PATCH 11/14] fix(discovery): keep scheduled workers independent of subagent presets Generated-by: OpenAI Codex --- plugins/agent/native/src/owner/discovery.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/plugins/agent/native/src/owner/discovery.rs b/plugins/agent/native/src/owner/discovery.rs index 23a9a5025bc..2647155632f 100644 --- a/plugins/agent/native/src/owner/discovery.rs +++ b/plugins/agent/native/src/owner/discovery.rs @@ -518,6 +518,7 @@ impl Owner { thinking_level: job.approval.thinking_level, }), compaction_prompt: None, + applied_preset: None, collaboration_mode: CollaborationMode::Agent, tools: Some(self.discovery_tools(job.recall)), title: fire.title.clone(), From 0e282f71be604ea40c6a6f5f85916682432a9ad2 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Sat, 14 Nov 2026 22:35:45 +0800 Subject: [PATCH 12/14] fix(discovery): retain the approval-required result for unapproved plans Generated-by: OpenAI Codex --- plugins/agent/native/src/owner/scheduler/tools.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/plugins/agent/native/src/owner/scheduler/tools.rs b/plugins/agent/native/src/owner/scheduler/tools.rs index a5741cfa72a..28dc50cca4c 100644 --- a/plugins/agent/native/src/owner/scheduler/tools.rs +++ b/plugins/agent/native/src/owner/scheduler/tools.rs @@ -379,6 +379,9 @@ impl Owner { Input::Pause { .. } => Some(pause_plan(&saved.plan, now() as i64)), Input::Resume { .. } => { let Some(approval) = &saved.plan.authorization else { + if matches!(saved.plan.task.effect, ScheduledEffect::Discovery { .. }) { + return Err("scheduler_user_approval_required".into()); + } return Err("scheduler_execution_unavailable".into()); }; let basis = match approval { From 3df9c146322c9a2a45dba9631895d4c75a749e5c Mon Sep 17 00:00:00 2001 From: rick12345 Date: Sat, 14 Nov 2026 22:59:03 +0800 Subject: [PATCH 13/14] chore(sdk): regenerate the synchronized discovery loop contracts Generated-by: OpenAI Codex --- packages/plugin-sdk/src/contracts.ts | 39 ++++++++++++++++++-- packages/plugin-sdk/src/generated/runtime.ts | 7 ++++ 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/packages/plugin-sdk/src/contracts.ts b/packages/plugin-sdk/src/contracts.ts index 4bfb6b51da8..915f814015b 100644 --- a/packages/plugin-sdk/src/contracts.ts +++ b/packages/plugin-sdk/src/contracts.ts @@ -31575,6 +31575,25 @@ export const contracts: { additionalProperties: false, required: ['kind', 'taskId', 'revision', 'input'], }, + { + type: 'object', + properties: { + kind: { + type: 'string', + const: 'approve', + }, + taskId: { + type: 'string', + }, + revision: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + }, + additionalProperties: false, + required: ['kind', 'taskId', 'revision'], + }, { type: 'object', properties: { @@ -32318,6 +32337,9 @@ export const contracts: { script: { type: 'boolean', }, + approved: { + type: 'boolean', + }, status: { $ref: '#/$defs/Status', }, @@ -32333,7 +32355,7 @@ export const contracts: { }, }, additionalProperties: false, - required: ['id', 'revision', 'title', 'sessionId', 'script', 'status'], + required: ['id', 'revision', 'title', 'sessionId', 'script', 'approved', 'status'], }, Status: { type: 'string', @@ -32356,12 +32378,15 @@ export const contracts: { misfire: { $ref: '#/$defs/Misfire', }, + approved: { + type: 'boolean', + }, pendingFire: { type: ['string', 'null'], }, }, additionalProperties: false, - required: ['revision', 'task', 'timezone', 'misfire'], + required: ['revision', 'task', 'timezone', 'misfire', 'approved'], }, Task: { type: 'object', @@ -40471,6 +40496,9 @@ export const contracts: { script: { type: 'boolean', }, + approved: { + type: 'boolean', + }, status: { $ref: '#/$defs/Status', }, @@ -40486,7 +40514,7 @@ export const contracts: { }, }, additionalProperties: false, - required: ['id', 'revision', 'title', 'sessionId', 'script', 'status'], + required: ['id', 'revision', 'title', 'sessionId', 'script', 'approved', 'status'], }, Status: { type: 'string', @@ -40509,12 +40537,15 @@ export const contracts: { misfire: { $ref: '#/$defs/Misfire', }, + approved: { + type: 'boolean', + }, pendingFire: { type: ['string', 'null'], }, }, additionalProperties: false, - required: ['revision', 'task', 'timezone', 'misfire'], + required: ['revision', 'task', 'timezone', 'misfire', 'approved'], }, Task: { type: 'object', diff --git a/packages/plugin-sdk/src/generated/runtime.ts b/packages/plugin-sdk/src/generated/runtime.ts index 6b584df65b7..226b594f426 100644 --- a/packages/plugin-sdk/src/generated/runtime.ts +++ b/packages/plugin-sdk/src/generated/runtime.ts @@ -2536,6 +2536,11 @@ export type AgentScheduleChange = revision: number; input: AgentScheduledInput; } + | { + kind: 'approve'; + taskId: string; + revision: number; + } | { kind: 'pause'; taskId: string; @@ -7125,6 +7130,7 @@ export interface AgentScheduledSummary { title: string; sessionId: string; script: boolean; + approved: boolean; status: Status; nextFireAt?: number | null; pendingFire?: string | null; @@ -7135,6 +7141,7 @@ export interface AgentScheduledDetail { task: Task; timezone: string; misfire: Misfire; + approved: boolean; pendingFire?: string | null; } export interface Task { From 27f89b932f33a3b213dc78d074a5f07a23c0e2d0 Mon Sep 17 00:00:00 2001 From: rick12345 Date: Sat, 14 Nov 2026 23:19:16 +0800 Subject: [PATCH 14/14] fix(discovery): keep discovery plans outside the automation editor Generated-by: OpenAI Codex --- packages/plugin-sdk/src/contracts.ts | 44 +++++++++++++++++++ packages/plugin-sdk/src/generated/runtime.ts | 6 +++ plugins/agent/api/src/contract/scheduler.rs | 5 +++ .../driver/tests/model_recovery/discovery.rs | 7 ++- plugins/agent/native/src/owner/scheduler.rs | 21 +++++++++ products/maka/workbench/app/Automations.tsx | 4 +- 6 files changed, 84 insertions(+), 3 deletions(-) diff --git a/packages/plugin-sdk/src/contracts.ts b/packages/plugin-sdk/src/contracts.ts index 915f814015b..e58726c83d6 100644 --- a/packages/plugin-sdk/src/contracts.ts +++ b/packages/plugin-sdk/src/contracts.ts @@ -32214,6 +32214,28 @@ export const contracts: { $schema: 'https://json-schema.org/draft/2020-12/schema', title: 'AgentScheduleRead', oneOf: [ + { + type: 'object', + properties: { + kind: { + type: 'string', + const: 'discovery', + }, + sessionId: { + type: 'string', + }, + cursor: { + type: ['string', 'null'], + }, + expectedRevision: { + type: ['integer', 'null'], + format: 'uint64', + minimum: 0, + }, + }, + additionalProperties: false, + required: ['kind', 'sessionId'], + }, { type: 'object', properties: { @@ -40373,6 +40395,28 @@ export const contracts: { $schema: 'https://json-schema.org/draft/2020-12/schema', title: 'AgentScheduleRead', oneOf: [ + { + type: 'object', + properties: { + kind: { + type: 'string', + const: 'discovery', + }, + sessionId: { + type: 'string', + }, + cursor: { + type: ['string', 'null'], + }, + expectedRevision: { + type: ['integer', 'null'], + format: 'uint64', + minimum: 0, + }, + }, + additionalProperties: false, + required: ['kind', 'sessionId'], + }, { type: 'object', properties: { diff --git a/packages/plugin-sdk/src/generated/runtime.ts b/packages/plugin-sdk/src/generated/runtime.ts index 226b594f426..6f92d6a3a26 100644 --- a/packages/plugin-sdk/src/generated/runtime.ts +++ b/packages/plugin-sdk/src/generated/runtime.ts @@ -2430,6 +2430,12 @@ export type AgentRecapFailure = | 'incomplete_output' | 'model_unavailable'; export type AgentScheduleRead = + | { + kind: 'discovery'; + sessionId: string; + cursor?: string | null; + expectedRevision?: number | null; + } | { kind: 'list'; cursor?: string | null; diff --git a/plugins/agent/api/src/contract/scheduler.rs b/plugins/agent/api/src/contract/scheduler.rs index 31f56bdeb2b..a96ad9aa64f 100644 --- a/plugins/agent/api/src/contract/scheduler.rs +++ b/plugins/agent/api/src/contract/scheduler.rs @@ -120,6 +120,11 @@ pub struct ScheduledInput { )] #[schemars(rename = "AgentScheduleRead")] pub enum ScheduleRead { + Discovery { + session_id: String, + cursor: Option, + expected_revision: Option, + }, List { cursor: Option, expected_revision: Option, diff --git a/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs b/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs index 624fc6f3556..9661fbb45f2 100644 --- a/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs +++ b/plugins/agent/native/src/owner/driver/tests/model_recovery/discovery.rs @@ -745,14 +745,17 @@ async fn discovery_model_proposals_require_bot_identity_and_human_approval() { finish_text(&mut socket, "Cannot propose discovery here").await; f.quiescent(session).await; assert_eq!( - f.query("schedules", json!({"kind":"list"})).await["tasks"], + f.query("schedules", json!({"kind":"discovery","sessionId":f.bot})) + .await["tasks"], json!([]) ); f.send("Propose a discovery").await; let (mut socket, _) = next(&listener).await; tool(&mut socket, "ScheduledTask", proposal).await; let (mut socket, _) = next(&listener).await; - let page = f.query("schedules", json!({"kind":"list"})).await; + let page = f + .query("schedules", json!({"kind":"discovery","sessionId":f.bot})) + .await; let task = &page["tasks"][0]; let id = task["id"].as_str().unwrap(); assert_eq!(task["status"], "paused"); diff --git a/plugins/agent/native/src/owner/scheduler.rs b/plugins/agent/native/src/owner/scheduler.rs index 61057663ce9..e771c796e99 100644 --- a/plugins/agent/native/src/owner/scheduler.rs +++ b/plugins/agent/native/src/owner/scheduler.rs @@ -177,10 +177,22 @@ impl Owner { let scheduler = &self.scheduler; let _gate = scheduler.gate.lock().await; let catalog = scheduler.load().await?; + let discovery_session = match &input { + ScheduleRead::Discovery { session_id, .. } => { + self.session(session_id).await?; + Some(session_id.clone()) + } + _ => None, + }; match input { ScheduleRead::List { cursor, expected_revision, + } + | ScheduleRead::Discovery { + cursor, + expected_revision, + .. } => { let revision = catalog.revision.unwrap_or(0); if expected_revision.is_some_and(|expected| expected != revision) { @@ -192,6 +204,15 @@ impl Owner { let mut tasks = Vec::new(); let mut next_cursor = None; for (id, saved) in &catalog.plans { + let matches_view = match &saved.plan.task.effect { + ScheduledEffect::Discovery { target, .. } => { + discovery_session.as_ref() == Some(&target.session_id) + } + _ => discovery_session.is_none(), + }; + if !matches_view { + continue; + } if cursor.as_ref().is_some_and(|after| id <= after) { continue; } diff --git a/products/maka/workbench/app/Automations.tsx b/products/maka/workbench/app/Automations.tsx index a371727dbb4..65f31023985 100644 --- a/products/maka/workbench/app/Automations.tsx +++ b/products/maka/workbench/app/Automations.tsx @@ -87,7 +87,9 @@ function AutomationEditor(props: { ? 'new' : effect?.kind === 'executor_session' ? 'followup' - : (effect?.delivery ?? 'steering'), + : effect?.kind === 'session' + ? (effect.delivery ?? 'steering') + : 'steering', ); const title = signal(task?.title ?? ''); const text = signal(task?.intent.body ?? '');