You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/10/17 05:49:48 UTC
(maka) branch rust-preview updated: fix(agent): preserve safe model preparation failure causes
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch rust-preview
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/rust-preview by this push:
new a84cddf853 fix(agent): preserve safe model preparation failure causes
a84cddf853 is described below
commit a84cddf853be2a59486e4549178ae75187f6728a
Author: Wang <m4...@gmail.com>
AuthorDate: Sat Oct 17 13:49:39 2026 +0800
fix(agent): preserve safe model preparation failure causes
Record typed provider, configuration and OAuth failures before conversion to public errors, retaining existing identity without raw payloads. Preserve refresh claim, replacement retry, cancellation and replay behavior.
Generated-by: Codex
---
Cargo.lock | 2 +
plugins/agent/native/Cargo.toml | 2 +
plugins/agent/native/README.md | 2 +
plugins/agent/native/README.zh-CN.md | 2 +
plugins/agent/native/src/model.rs | 301 +++++++++++++++------
plugins/agent/native/src/model/diagnostics.rs | 137 ++++++++++
plugins/agent/native/src/model/oauth.rs | 140 ++++++++--
plugins/agent/native/src/model/oauth/settlement.rs | 92 ++++++-
plugins/agent/native/src/owner/tests.rs | 21 ++
plugins/agent/native/src/owner/tests/selection.rs | 97 +++++++
plugins/agent/native/src/owner/tests/title.rs | 14 +-
products/maka/cli/notices.json | 4 +-
12 files changed, 692 insertions(+), 122 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index a8d6c70f47..3f8f5d715a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8217,6 +8217,8 @@ dependencies = [
"tempfile",
"tokio",
"tokio-util",
+ "tracing",
+ "tracing-subscriber",
"uuid",
]
diff --git a/plugins/agent/native/Cargo.toml b/plugins/agent/native/Cargo.toml
index ff75ab2351..ed82b7c1d6 100644
--- a/plugins/agent/native/Cargo.toml
+++ b/plugins/agent/native/Cargo.toml
@@ -69,9 +69,11 @@ serde_json.workspace = true
sha2.workspace = true
tokio.workspace = true
tokio-util.workspace = true
+tracing = { version = "0.1", default-features = false, features = ["std"] }
uuid.workspace = true
[dev-dependencies]
+tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] }
maka-computer-native.workspace = true
maka-external-agent.workspace = true
maka-packages.workspace = true
diff --git a/plugins/agent/native/README.md b/plugins/agent/native/README.md
index f6c541deb1..c11c826d93 100644
--- a/plugins/agent/native/README.md
+++ b/plugins/agent/native/README.md
@@ -47,6 +47,8 @@ The default Maka composition enables `automaticTitles`. Conversations created wi
Automatic-title preparation, model and invalid-output failures are durable session diagnostics. Empty or output-limited responses retain their actual physical usage while recording the logical title failure. Stopping the owner and policy skips do not report title failures; uncertain persistence or cleanup retains its original recovery priority.
+Model preparation and OAuth refresh failures emit Host warnings before their original cause is reduced to a public error. Warnings include a safe error category and existing connection/session identifiers when available; they exclude credentials, configuration, request content and arbitrary provider error text. Re-reading a failed refresh does not log the original failure again.
+
Model context capacity, any independent input ceiling and the compaction threshold are captured with the accepted model selection and retained across queueing and restart. Without an explicit threshold, known input capacity sets the proactive threshold at 85%; output reserve remains a separate engine calculation.
The default composition enables `recaps`. The Session recap command reads cached text; generating a new recap requires explicit confirmation of the selected model and possible charges. Each request reads at most 256 bounded history pages, keeps at most 32 KiB of history text, and limits output to 1,024 tokens and 30 seconds. Incognito mode and session access are checked even for cached reads. Refresh and original operation recovery never regenerate; an unknown request remains pending acr [...]
diff --git a/plugins/agent/native/README.zh-CN.md b/plugins/agent/native/README.zh-CN.md
index 3628f94187..c3e5e818c5 100644
--- a/plugins/agent/native/README.zh-CN.md
+++ b/plugins/agent/native/README.zh-CN.md
@@ -47,6 +47,8 @@ Maka 默认组合启用 `automaticTitles`。通过 `automaticTitle: true` 创建
自动标题的准备、模型及无效输出失败会持久记录到会话诊断。空响应或输出达到上限时，保留实际请求用量，同时记录标题生成失败。服务停止及策略性跳过不报告标题失败；持久化或清理结果不确定时，保留原有恢复优先级。
+模型准备和 OAuth 刷新失败在原始原因被转换为公开错误前写入 Host 警告。警告包含安全错误分类及已有的连接／会话标识（若可用），不包含凭据、配置、请求内容或任意 provider 错误文本。再次读取已失败的刷新结果不会重复记录原始失败。
+
模型上下文总容量、独立输入上限与压缩阈值随模型选择一起保存，排队和重启后继续使用接收时的值。未显式设置压缩阈值时，使用已知输入容量的 85%；输出预留仍由引擎单独计算。
默认组合启用 `recaps`。「会话回顾」命令读取缓存；生成新回顾前需明确确认所选模型及可能产生的费用。每次最多扫描 256 个有界历史页，保留不超过 32 KiB 的历史文本，输出上限为 1,024 token，生成限时 30 秒。读取缓存也检查隐私模式和会话访问范围。刷新和原始操作恢复不会重新生成；结果未知的请求在重启后仍保留为待定。辅助用量归属于原始应用操作，关闭时等待模型请求实际清理完成。
diff --git a/plugins/agent/native/src/model.rs b/plugins/agent/native/src/model.rs
index bf05992a1b..2efd1aef69 100644
--- a/plugins/agent/native/src/model.rs
+++ b/plugins/agent/native/src/model.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+mod diagnostics;
pub(crate) mod oauth;
use futures_util::future::BoxFuture;
@@ -39,6 +40,7 @@ use maka_model::{ProviderAuth, ProviderConfig};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::sync::Arc;
+use tracing::Instrument;
/// Only parameters actually supplied to execution. Catalog labels, tests and
/// other models are observations, not a mutable authority over an accepted run.
@@ -106,47 +108,59 @@ impl Source {
session: String,
mut revision: SourceRevision,
) -> Result<(Self, PreparedModel), RunError> {
- let material = configuration
- .observe_selected_model(selection.clone())
- .await
- .map_err(|_| unavailable())?
- .ok_or_else(unavailable)?;
- let row = &material.connection;
- let definition =
- Arc::new(maka_providers::definition(&row.provider).map_err(|_| unavailable())?);
- let model = resolve(&definition, row, &selection.model_id, thinking_level).await?;
- revision.name = row.provider.name.clone();
- let source = Self {
- configuration,
- credentials,
- accepted: AcceptedModel {
- context: request_context(row, &model.info)?,
- editing_tools: Some(editing_tools(row, &selection.model_id, &model.base_url)),
- tool_mode: ToolMode::for_model(
- &selection.model_id,
- &model.base_url,
- row.model_overrides
- .as_ref()
- .and_then(|models| models.get(&selection.model_id))
- .and_then(|overrides| overrides.code_mode),
- ),
- binding: ModelBinding {
- connection_id: row.connection_id.clone(),
- connection_slug: row.slug.clone(),
- model: selection.model_id,
+ let span = tracing::warn_span!("agent_model_preparation", session_id = %session,
+ connection_id = %selection.connection_id, activation = %revision.activation);
+ async move {
+ let material = configuration
+ .observe_selected_model(selection.clone())
+ .await
+ .map_err(|error| {
+ diagnostics::configuration("selected_model_observation", &error, Some(&selection.connection_id));
+ unavailable()
+ })?
+ .ok_or_else(unavailable)?;
+ let row = &material.connection;
+ let definition =
+ Arc::new(maka_providers::definition(&row.provider).map_err(|_| {
+ tracing::warn!(stage = "provider_definition", cause = "provider_definition_unavailable",
+ provider = %row.provider.name, "Model preparation failed");
+ unavailable()
+ })?);
+ let model = resolve(&definition, row, &selection.model_id, thinking_level).await?;
+ revision.name = row.provider.name.clone();
+ let source = Self {
+ configuration,
+ credentials,
+ accepted: AcceptedModel {
+ context: request_context(row, &model.info)?,
+ editing_tools: Some(editing_tools(row, &selection.model_id, &model.base_url)),
+ tool_mode: ToolMode::for_model(
+ &selection.model_id,
+ &model.base_url,
+ row.model_overrides
+ .as_ref()
+ .and_then(|models| models.get(&selection.model_id))
+ .and_then(|overrides| overrides.code_mode),
+ ),
+ binding: ModelBinding {
+ connection_id: row.connection_id.clone(),
+ connection_slug: row.slug.clone(),
+ model: selection.model_id,
+ },
+ thinking_level,
+ provider: row.provider.clone(),
+ provider_configuration: row.configuration.clone(),
+ basis: ExecutionBasis::new(&model, row),
},
- thinking_level,
- provider: row.provider.clone(),
- provider_configuration: row.configuration.clone(),
- basis: ExecutionBasis::new(&model, row),
- },
- definition,
- session,
- revision,
- };
- let prepared = source.prepare(material, model).await?;
- Ok((source, prepared))
+ definition,
+ session,
+ revision,
+ };
+ let prepared = source.prepare(material, model).await?;
+ Ok((source, prepared))
+ }.instrument(span).await
}
+
pub fn restore(
configuration: Arc<ConfigurationStore>,
credentials: Credentials,
@@ -154,8 +168,15 @@ impl Source {
session: String,
mut revision: SourceRevision,
) -> Result<Self, RunError> {
+ let _span = tracing::warn_span!("agent_model_preparation", session_id = %session,
+ connection_id = %accepted.binding.connection_id, activation = %revision.activation)
+ .entered();
let definition =
- Arc::new(maka_providers::definition(&accepted.provider).map_err(|_| unavailable())?);
+ Arc::new(maka_providers::definition(&accepted.provider).map_err(|_| {
+ tracing::warn!(stage = "restore_provider_definition", cause = "provider_definition_unavailable",
+ provider = %accepted.provider.name, "Model preparation failed");
+ unavailable()
+ })?);
revision.name = accepted.provider.name.clone();
Ok(Self {
configuration,
@@ -173,6 +194,11 @@ impl Source {
) -> Result<PreparedModel, RunError> {
let capabilities = model.info.capabilities.unwrap_or_default();
if capabilities.chat == Some(false) {
+ tracing::warn!(
+ stage = "model_capabilities",
+ cause = "chat_unsupported",
+ "Model preparation failed"
+ );
return Err(unavailable());
}
let row = &material.connection;
@@ -180,7 +206,14 @@ impl Source {
&material.network.proxy,
material.network.password.as_deref(),
)
- .map_err(|_| unavailable())?;
+ .map_err(|error| {
+ diagnostics::network(
+ "network_policy",
+ &error,
+ &self.accepted.binding.connection_id,
+ );
+ unavailable()
+ })?;
let credential = match material.credential {
Some(snapshot) if snapshot.credential().refresh_at.is_some() => Some(
self.credentials
@@ -191,11 +224,16 @@ impl Source {
self.accepted.provider.clone(),
self.definition.clone(),
maka_agent_ports::provider::Context {
- transport: self
- .credentials
- .executor
- .transport(&network)
- .map_err(|_| unavailable())?,
+ transport: self.credentials.executor.transport(&network).map_err(
+ |error| {
+ diagnostics::model(
+ "credential_transport",
+ &error,
+ &self.accepted.binding.connection_id,
+ );
+ unavailable()
+ },
+ )?,
cancellation: tokio_util::sync::CancellationToken::new(),
interaction: None,
},
@@ -212,7 +250,15 @@ impl Source {
.definition
.authorize(connection(row), credential, self.session.clone())
.await
- .map_err(|_| RunError::InvalidInput("agent_authentication_unavailable".into()))?;
+ .map_err(|error| {
+ diagnostics::provider(
+ "provider_authorize",
+ &error,
+ &row.connection_id,
+ &row.provider.name,
+ );
+ RunError::InvalidInput("agent_authentication_unavailable".into())
+ })?;
let auth = match credentials {
maka_agent_ports::model::Credentials::ApiKey(key) => ProviderAuth::ApiKey(key),
maka_agent_ports::model::Credentials::RequestHeaders(headers) => {
@@ -249,6 +295,7 @@ fn request_context(
) -> Result<ModelRequestContext, RunError> {
if matches!((model.context_window, model.input_limit), (Some(capacity), Some(input)) if input > capacity)
{
+ tracing::warn!(stage = "request_context", cause = "input_limit_exceeds_context", connection_id = %row.connection_id, "Model preparation failed");
return Err(unavailable());
}
let input = match (model.context_window, model.input_limit) {
@@ -266,21 +313,54 @@ fn request_context(
.and_then(|profile| profile.compaction_threshold)
.or_else(|| input.map(|limit| ((u128::from(limit) * 85 / 100) as u64).max(1))),
};
- context.validate().map_err(|_| unavailable())?;
+ context.validate().map_err(|_| {
+ tracing::warn!(stage = "request_context", cause = "request_context_invalid", connection_id = %row.connection_id, "Model preparation failed");
+ unavailable()
+ })?;
Ok(context)
}
impl Source {
+ fn preparation_span(&self) -> tracing::Span {
+ tracing::warn_span!("agent_model_preparation", session_id = %self.session,
+ connection_id = %self.accepted.binding.connection_id, activation = %self.revision.activation,
+ provider = %self.accepted.provider.name)
+ }
async fn observe(&self) -> Result<(ConnectionObservation, Model), RunError> {
let material = self
.configuration
.observe_model(self.accepted.binding.clone())
.await
- .map_err(|_| unavailable())?
- .ok_or_else(unavailable)?;
+ .map_err(|error| {
+ diagnostics::configuration(
+ "accepted_model_observation",
+ &error,
+ Some(&self.accepted.binding.connection_id),
+ );
+ unavailable()
+ })?
+ .ok_or_else(|| {
+ tracing::warn!(
+ stage = "accepted_model_observation",
+ cause = "accepted_source_unavailable",
+ "Model preparation failed"
+ );
+ unavailable()
+ })?;
let row = &material.connection;
- if row.provider != self.accepted.provider
- || row.configuration != self.accepted.provider_configuration
- {
+ if row.provider != self.accepted.provider {
+ tracing::warn!(
+ stage = "accepted_model_observation",
+ cause = "accepted_provider_changed",
+ "Model preparation failed"
+ );
+ return Err(unavailable());
+ }
+ if row.configuration != self.accepted.provider_configuration {
+ tracing::warn!(
+ stage = "accepted_model_observation",
+ cause = "accepted_configuration_changed",
+ "Model preparation failed"
+ );
return Err(unavailable());
}
let model = resolve(
@@ -291,37 +371,49 @@ impl Source {
)
.await?;
if ExecutionBasis::new(&model, row) != self.accepted.basis {
+ tracing::warn!(
+ stage = "accepted_model_observation",
+ cause = "accepted_execution_basis_changed",
+ "Model preparation failed"
+ );
return Err(unavailable());
}
Ok((material, model))
}
/// The original accepted route, with the least declared reasoning for a short auxiliary task.
pub(crate) async fn capture_auxiliary(&self) -> Result<PreparedModel, RunError> {
- let (material, model) = self.observe().await?;
- let least = ThinkingLevel::ALL
- .into_iter()
- .find(|level| model.thinking_levels.contains(level));
- let model = match least {
- Some(level) => {
- resolve(
- &self.definition,
- &material.connection,
- &self.accepted.binding.model,
- Some(level),
- )
- .await?
- }
- None => model,
- };
- self.prepare(material, model).await
+ async {
+ let (material, model) = self.observe().await?;
+ let least = ThinkingLevel::ALL
+ .into_iter()
+ .find(|level| model.thinking_levels.contains(level));
+ let model = match least {
+ Some(level) => {
+ resolve(
+ &self.definition,
+ &material.connection,
+ &self.accepted.binding.model,
+ Some(level),
+ )
+ .await?
+ }
+ None => model,
+ };
+ self.prepare(material, model).await
+ }
+ .instrument(self.preparation_span())
+ .await
}
}
impl ModelSource for Source {
fn capture(&self) -> BoxFuture<'_, Result<PreparedModel, RunError>> {
- Box::pin(async move {
- let (material, model) = self.observe().await?;
- self.prepare(material, model).await
- })
+ Box::pin(
+ async move {
+ let (material, model) = self.observe().await?;
+ self.prepare(material, model).await
+ }
+ .instrument(self.preparation_span()),
+ )
}
}
@@ -334,10 +426,21 @@ pub(crate) async fn candidate_editing_tools(
let material = configuration
.observe_selected_model(selection.clone())
.await
- .map_err(|_| unavailable())?
+ .map_err(|error| {
+ diagnostics::configuration(
+ "candidate_model_observation",
+ &error,
+ Some(&selection.connection_id),
+ );
+ unavailable()
+ })?
.ok_or_else(unavailable)?;
let row = &material.connection;
- let definition = maka_providers::definition(&row.provider).map_err(|_| unavailable())?;
+ let definition = maka_providers::definition(&row.provider).map_err(|_| {
+ tracing::warn!(stage = "provider_definition", cause = "provider_definition_unavailable", connection_id = %row.connection_id,
+ provider = %row.provider.name, "Model preparation failed");
+ unavailable()
+ })?;
let model = resolve(&definition, row, &selection.model_id, thinking_level).await?;
Ok(editing_tools(row, &selection.model_id, &model.base_url))
}
@@ -379,7 +482,15 @@ async fn resolve(
thinking_level,
})
.await
- .map_err(|_| unavailable())
+ .map_err(|error| {
+ diagnostics::provider(
+ "provider_prepare",
+ &error,
+ &row.connection_id,
+ &row.provider.name,
+ );
+ unavailable()
+ })
}
fn connection(row: &maka_agent_facts::configuration::ConnectionCatalogEntry) -> Connection {
Connection {
@@ -400,17 +511,29 @@ impl maka_agent_engine::pricing::Pricing for Prices {
provider: &'a str,
model: &'a str,
) -> BoxFuture<'a, Result<maka_agent_facts::pricing::Quote, RunError>> {
- Box::pin(async move {
- self.0
- .quote_model(provider.into(), model.into())
- .await
- .map_err(|error| match error {
- maka_config::ConfigError::CommitUnknown => {
- maka_agent_facts::event::CommitError::OutcomeUnknown(error.to_string())
- .into()
- }
- _ => RunError::Internal("model pricing unavailable".into()),
- })
- })
+ Box::pin(
+ async move {
+ self.0
+ .quote_model(provider.into(), model.into())
+ .await
+ .map_err(|error| {
+ diagnostics::configuration("model_pricing", &error, None);
+ match error {
+ maka_config::ConfigError::CommitUnknown => {
+ maka_agent_facts::event::CommitError::OutcomeUnknown(
+ error.to_string(),
+ )
+ .into()
+ }
+ _ => RunError::Internal("model pricing unavailable".into()),
+ }
+ })
+ }
+ .instrument(tracing::warn_span!(
+ "agent_model_pricing",
+ provider,
+ model_id = model
+ )),
+ )
}
}
diff --git a/plugins/agent/native/src/model/diagnostics.rs b/plugins/agent/native/src/model/diagnostics.rs
new file mode 100644
index 0000000000..faed45e8e6
--- /dev/null
+++ b/plugins/agent/native/src/model/diagnostics.rs
@@ -0,0 +1,137 @@
+/*
+ * 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.
+ */
+
+//! Safe evidence at the boundary where private errors become public rejection codes.
+
+use maka_agent_ports::provider::Error as ProviderError;
+use maka_config::ConfigError;
+use maka_model::ModelError;
+
+pub(super) fn provider(
+ stage: &'static str,
+ error: &ProviderError,
+ connection_id: &str,
+ provider: &str,
+) {
+ let cause = match error {
+ ProviderError::Cancelled => return,
+ ProviderError::Unavailable => "provider_unavailable",
+ ProviderError::AuthenticationRequired => "provider_authentication_required",
+ ProviderError::Rejected(_) => "provider_rejected",
+ ProviderError::Invalid(_) => "provider_invalid",
+ ProviderError::OutcomeUnknown => "provider_outcome_unknown",
+ ProviderError::Transport(_) => "provider_transport",
+ ProviderError::Http(_) => "provider_http",
+ };
+ let status = match error {
+ ProviderError::Http(status) => Some(*status),
+ _ => None,
+ };
+ tracing::warn!(
+ stage,
+ cause,
+ connection_id,
+ provider,
+ status,
+ "Model preparation failed"
+ );
+}
+
+pub(super) fn configuration(stage: &'static str, error: &ConfigError, connection_id: Option<&str>) {
+ let cause = match error {
+ ConfigError::Invalid(_) => "config_invalid",
+ ConfigError::UnsupportedDatabase => "config_unsupported_database",
+ ConfigError::CommitUnknown => "config_commit_unknown",
+ ConfigError::Io(_) => "config_io",
+ ConfigError::Sqlx(_) => "config_sqlx",
+ ConfigError::Migration(_) => "config_migration",
+ ConfigError::Store(_) => "config_store",
+ ConfigError::Json(_) => "config_json",
+ };
+ match error {
+ ConfigError::Io(error) => tracing::warn!(
+ stage, cause, connection_id, error_kind = ?error.kind(), os_code = error.raw_os_error(),
+ "Model preparation failed"
+ ),
+ ConfigError::Sqlx(error) => {
+ // Only a numeric SQLite code, never the database's free-form message.
+ let sql_code = error
+ .as_database_error()
+ .and_then(|error| error.code())
+ .and_then(|code| code.parse::<i32>().ok());
+ tracing::warn!(
+ stage,
+ cause,
+ connection_id,
+ sql_code,
+ "Model preparation failed"
+ );
+ }
+ ConfigError::Json(error) => tracing::warn!(
+ stage, cause, connection_id, category = ?error.classify(),
+ line = error.line(), column = error.column(), "Model preparation failed"
+ ),
+ _ => tracing::warn!(stage, cause, connection_id, "Model preparation failed"),
+ }
+}
+
+pub(super) fn model(stage: &'static str, error: &ModelError, connection_id: &str) {
+ match error {
+ ModelError::Cancelled => {}
+ ModelError::Http { status } => tracing::warn!(
+ stage,
+ cause = "model_http",
+ connection_id,
+ status,
+ "Model preparation failed"
+ ),
+ ModelError::TimedOut => tracing::warn!(
+ stage,
+ cause = "model_timed_out",
+ connection_id,
+ "Model preparation failed"
+ ),
+ ModelError::ContextOverflow { observed_output } => tracing::warn!(
+ stage,
+ cause = "model_context_overflow",
+ connection_id,
+ observed_output,
+ "Model preparation failed"
+ ),
+ ModelError::Provider(error) => {
+ tracing::warn!(stage, cause = "model_provider", connection_id, reason = ?error.reason(), "Model preparation failed")
+ }
+ ModelError::Adapter(_) => tracing::warn!(
+ stage,
+ cause = "model_adapter",
+ connection_id,
+ "Model preparation failed"
+ ),
+ }
+}
+
+pub(super) fn network(stage: &'static str, error: &maka_network::Error, connection_id: &str) {
+ let cause = match error {
+ maka_network::Error::InvalidProxy => "proxy_invalid",
+ maka_network::Error::MissingCredentials => "proxy_credentials_missing",
+ maka_network::Error::InvalidUpgrade => "network_invalid_upgrade",
+ maka_network::Error::Transport(_) => "network_transport",
+ };
+ tracing::warn!(stage, cause, connection_id, "Model preparation failed");
+}
diff --git a/plugins/agent/native/src/model/oauth.rs b/plugins/agent/native/src/model/oauth.rs
index a81d64b4ec..4389d32200 100644
--- a/plugins/agent/native/src/model/oauth.rs
+++ b/plugins/agent/native/src/model/oauth.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use super::diagnostics;
use maka_agent_ports::provider::Connection;
use maka_agent_ports::provider::Context;
use maka_agent_ports::provider::Identity;
@@ -28,6 +29,7 @@ use std::{
};
use tokio::sync::{Mutex as AsyncMutex, watch};
use tokio_util::{sync::CancellationToken, task::TaskTracker};
+use tracing::Instrument;
struct RefreshProvider {
identity: Identity,
@@ -173,7 +175,14 @@ impl Credential {
.snapshot
.current_generation()
.await
- .map_err(failure)?
+ .map_err(|error| {
+ diagnostics::configuration(
+ "credential_generation_observation",
+ &error,
+ Some(&self.snapshot.target().connection_id),
+ );
+ failure(error)
+ })?
.ok_or_else(|| failure("Provider credential was superseded"))?;
if current.basis() != resolved.basis()
|| current.credential() != resolved.credential()
@@ -208,25 +217,39 @@ impl Generation {
let flight = self.flight.clone();
// A cancelled request waiter cannot cancel an accepted exchange.
context.cancellation = shutdown.child_token();
- self.workers.spawn(async move {
- let mut state = state.lock().await;
- let result = if superseded.is_cancelled() {
- Err(failure("Provider credential was superseded"))
- } else {
- state
- .resolve(provider, connection, context, &shutdown, &superseded)
- .await
- };
- sender.send_replace(Some(result));
- *flight.lock().unwrap() = None;
- });
+ let span = tracing::warn_span!("agent_credential_refresh", connection_id = %connection.id,
+ provider = %provider.identity().name);
+ self.workers.spawn(
+ async move {
+ let mut state = state.lock().await;
+ let result = if superseded.is_cancelled() {
+ Err(failure("Provider credential was superseded"))
+ } else {
+ state
+ .resolve(provider, connection, context, &shutdown, &superseded)
+ .await
+ };
+ sender.send_replace(Some(result));
+ *flight.lock().unwrap() = None;
+ }
+ .instrument(span),
+ );
receiver
}
};
let result = receiver
.wait_for(Option::is_some)
.await
- .map_err(|_| failure("Credential owner stopped before settlement"))?
+ .map_err(|_| {
+ if !self.shutdown.is_cancelled() && !self.superseded.is_cancelled() {
+ tracing::warn!(
+ stage = "credential_refresh_wait",
+ cause = "credential_owner_stopped",
+ "Model preparation failed"
+ );
+ }
+ failure("Credential owner stopped before settlement")
+ })?
.as_ref()
.expect("settled result")
.clone();
@@ -302,6 +325,9 @@ mod tests {
let bytes = response.body.next().await.unwrap().unwrap();
assert!(response.body.next().await.unwrap().is_none());
response.body.close().await.unwrap();
+ if response.head.status != 200 {
+ return Err(Error::Http(response.head.status));
+ }
Ok(serde_json::from_slice(&bytes).unwrap())
})
}
@@ -320,7 +346,7 @@ mod tests {
request: oneshot::Sender<Vec<u8>>,
reply: oneshot::Receiver<Envelope>,
}
- struct Remote(Mutex<Option<Exchange>>);
+ struct Remote(Mutex<Option<Exchange>>, u16);
impl Transport for Remote {
fn identity(&self) -> u64 {
self as *const Self as usize as u64
@@ -340,7 +366,7 @@ mod tests {
let replacement = exchange.reply.await.unwrap();
Ok(http::Response {
head: http::Head {
- status: 200,
+ status: self.1,
url: request.url,
headers: vec![],
},
@@ -361,15 +387,27 @@ mod tests {
Context,
oneshot::Receiver<Vec<u8>>,
oneshot::Sender<Envelope>,
+ ) {
+ exchange_with_status(200)
+ }
+ fn exchange_with_status(
+ status: u16,
+ ) -> (
+ Context,
+ oneshot::Receiver<Vec<u8>>,
+ oneshot::Sender<Envelope>,
) {
let (request, received) = oneshot::channel();
let (reply, replacement) = oneshot::channel();
(
Context {
- transport: Arc::new(Remote(Mutex::new(Some(Exchange {
- request,
- reply: replacement,
- })))),
+ transport: Arc::new(Remote(
+ Mutex::new(Some(Exchange {
+ request,
+ reply: replacement,
+ })),
+ status,
+ )),
cancellation: CancellationToken::new(),
interaction: None,
},
@@ -545,11 +583,65 @@ mod tests {
.unwrap();
}
+ #[tokio::test]
+ async fn refresh_http_failure_logs_original_safe_cause_once_without_reusing_claim() {
+ const SECRET: &str = "SECRET_REFRESH_ERROR_BODY_MUST_NOT_BE_LOGGED";
+ tokio::time::timeout(std::time::Duration::from_secs(15), async {
+ let capture = tempfile::NamedTempFile::new().unwrap();
+ let subscriber = tracing_subscriber::fmt()
+ .with_ansi(false)
+ .without_time()
+ .with_writer(capture.reopen().unwrap())
+ .finish();
+ let _subscriber = tracing::subscriber::set_default(subscriber);
+ let fixture = Fixture::new().await;
+ let workers = TaskTracker::new();
+ let authority = Authority::new(workers.clone(), CancellationToken::new());
+ let binding = authority.bind(fixture.snapshot.clone()).unwrap();
+ let (context, admitted, reply) = exchange_with_status(403);
+ let mut resolving = Box::pin(fixture.resolve(&binding, context.clone()));
+ assert!(poll!(resolving.as_mut()).is_pending());
+ assert_eq!(admitted.await.unwrap(), b"old");
+ assert!(reply.send(envelope(SECRET, None)).is_ok());
+ let first = resolving.await.err().unwrap();
+ let replay = fixture.resolve(&binding, context).await.err().unwrap();
+ let current = fixture.snapshot.current_generation().await.unwrap().unwrap();
+ let claimed_again = fixture.snapshot.claim_refresh().await.unwrap();
+ workers.close();
+ workers.wait().await;
+ fixture.store.shutdown().await.unwrap();
+ let log = std::fs::read_to_string(capture.path()).unwrap();
+
+ for error in [first, replay] {
+ assert!(matches!(error, ModelError::Adapter(message) if message == "provider HTTP request failed with status 403"));
+ }
+ assert_eq!(current.basis(), fixture.snapshot.basis());
+ assert!(current.credential() == fixture.snapshot.credential());
+ assert!(!claimed_again, "the failed exchange's original claim must remain spent");
+ let originals = log.lines().filter(|line| line.contains("credential_refresh_exchange")).collect::<Vec<_>>();
+ assert_eq!(originals.len(), 1, "exactly one original typed failure must survive Adapter conversion: {log}");
+ let original = originals[0];
+ assert!(original.contains("WARN"), "{original}");
+ assert!(original.contains("cause=\"provider_http\""), "{original}");
+ assert!(original.contains("status=403"), "{original}");
+ assert!(original.contains(&fixture.snapshot.target().connection_id), "{original}");
+ assert!(!log.contains(SECRET));
+ assert!(!log.contains("provider.invalid"));
+ }).await.unwrap();
+ }
+
#[tokio::test]
async fn failed_and_unknown_sql_commits_retry_only_the_retained_envelope_and_logout_revokes_it()
{
tokio::time::timeout(std::time::Duration::from_secs(15), async {
for unknown in [false, true] {
+ let capture = tempfile::NamedTempFile::new().unwrap();
+ let subscriber = tracing_subscriber::fmt()
+ .with_ansi(false)
+ .without_time()
+ .with_writer(capture.reopen().unwrap())
+ .finish();
+ let _subscriber = tracing::subscriber::set_default(subscriber);
let fixture = Fixture::new().await;
let mut sql = fixture.sql().await;
if unknown {
@@ -582,6 +674,14 @@ mod tests {
use sqlx::Connection;
sql.close().await.unwrap();
fixture.store.shutdown().await.unwrap();
+ let log = std::fs::read_to_string(capture.path()).unwrap();
+ let failures = log.lines().filter(|line| line.contains("credential_refresh_commit")).collect::<Vec<_>>();
+ assert_eq!(failures.len(), 1, "original SQL failure must survive Adapter conversion: {log}");
+ let cause = if unknown { "cause=\"config_commit_unknown\"" } else { "cause=\"config_sqlx\"" };
+ assert!(failures[0].contains(cause), "{log}");
+ assert!(failures[0].contains(&fixture.snapshot.target().connection_id), "{log}");
+ assert!(!log.contains("injected rotation failure"));
+ assert!(!log.contains("rotated"));
}
}).await.unwrap();
}
diff --git a/plugins/agent/native/src/model/oauth/settlement.rs b/plugins/agent/native/src/model/oauth/settlement.rs
index 3242d8d628..ea3879dfe6 100644
--- a/plugins/agent/native/src/model/oauth/settlement.rs
+++ b/plugins/agent/native/src/model/oauth/settlement.rs
@@ -52,7 +52,14 @@ impl State {
.credential
.current_generation()
.await
- .map_err(failure)?
+ .map_err(|error| {
+ diagnostics::configuration(
+ "credential_generation_observation",
+ &error,
+ Some(&self.credential.target().connection_id),
+ );
+ failure(error)
+ })?
.ok_or_else(|| failure("Provider credential was superseded"))?;
if current.basis() != self.credential.basis() {
return Err(failure("Provider credential was superseded"));
@@ -71,8 +78,23 @@ impl State {
}
let refresh = provider
.prepare_refresh(connection, self.credential.credential().clone())
- .map_err(failure)?;
- if !self.credential.claim_refresh().await.map_err(failure)? {
+ .map_err(|error| {
+ diagnostics::provider(
+ "credential_refresh_prepare",
+ &error,
+ &self.credential.target().connection_id,
+ &self.credential.target().provider.name,
+ );
+ failure(error)
+ })?;
+ if !self.credential.claim_refresh().await.map_err(|error| {
+ diagnostics::configuration(
+ "credential_refresh_claim",
+ &error,
+ Some(&self.credential.target().connection_id),
+ );
+ failure(error)
+ })? {
let message = "Provider refresh was already claimed; its outcome is unknown";
self.outcome = Outcome::Failed(message.into());
return Err(failure(message));
@@ -84,8 +106,23 @@ impl State {
Ok(Ok(credential)) => credential,
result => {
let error = match result {
- Ok(Err(error)) => error.to_string(),
- Err(_) => "Provider refresh outcome is unknown".into(),
+ Ok(Err(error)) => {
+ diagnostics::provider(
+ "credential_refresh_exchange",
+ &error,
+ &self.credential.target().connection_id,
+ &self.credential.target().provider.name,
+ );
+ error.to_string()
+ }
+ Err(_) => {
+ tracing::warn!(
+ stage = "credential_refresh_exchange",
+ cause = "credential_refresh_timeout",
+ "Model preparation failed"
+ );
+ "Provider refresh outcome is unknown".into()
+ }
Ok(Ok(_)) => unreachable!(),
};
self.outcome = Outcome::Failed(error.clone());
@@ -105,14 +142,28 @@ impl State {
.credential
.current_generation()
.await
- .map_err(failure)?
+ .map_err(|error| {
+ diagnostics::configuration(
+ "credential_generation_observation",
+ &error,
+ Some(&self.credential.target().connection_id),
+ );
+ failure(error)
+ })?
.ok_or_else(|| failure("Provider credential was superseded"))?;
let next = self
.credential
.basis()
.revision
.checked_add(1)
- .ok_or_else(|| failure("Provider credential revision exhausted"))?;
+ .ok_or_else(|| {
+ tracing::warn!(
+ stage = "credential_refresh_commit",
+ cause = "credential_revision_exhausted",
+ "Model preparation failed"
+ );
+ failure("Provider credential revision exhausted")
+ })?;
// The complete envelope matters: refresh_at is Host scheduling authority.
if current.basis().revision == next && current.credential() == replacement {
self.credential = current;
@@ -125,13 +176,27 @@ impl State {
self.credential
.commit_refresh(replacement.clone(), now()?)
.await
- .map_err(failure)?
+ .map_err(|error| {
+ diagnostics::configuration(
+ "credential_refresh_commit",
+ &error,
+ Some(&self.credential.target().connection_id),
+ );
+ failure(error)
+ })?
.ok_or_else(|| failure("Provider credential was superseded"))?;
let current = self
.credential
.current_generation()
.await
- .map_err(failure)?
+ .map_err(|error| {
+ diagnostics::configuration(
+ "credential_generation_observation",
+ &error,
+ Some(&self.credential.target().connection_id),
+ );
+ failure(error)
+ })?
.ok_or_else(|| failure("Provider credential was superseded"))?;
if current.basis().revision != next || current.credential() != replacement {
return Err(failure("Provider credential was superseded"));
@@ -148,5 +213,12 @@ fn now() -> Result<u64, ModelError> {
.ok()
.and_then(|time| u64::try_from(time.as_millis()).ok())
.filter(|time| *time <= 9_007_199_254_740_991)
- .ok_or_else(|| failure("Invalid system time for credentials"))
+ .ok_or_else(|| {
+ tracing::warn!(
+ stage = "credential_clock",
+ cause = "credential_time_invalid",
+ "Model preparation failed"
+ );
+ failure("Invalid system time for credentials")
+ })
}
diff --git a/plugins/agent/native/src/owner/tests.rs b/plugins/agent/native/src/owner/tests.rs
index c015746fd9..914954581e 100644
--- a/plugins/agent/native/src/owner/tests.rs
+++ b/plugins/agent/native/src/owner/tests.rs
@@ -443,6 +443,27 @@ fn input(settings: &Configuration) -> (Input, Vec<Package>) {
}
(input, vec![provider, consumer, receipts, settings])
}
+// The default dispatcher is thread-local: hold the guard through current-thread
+// fixture shutdown so actual model preparation tasks remain inside the capture.
+fn model_log_capture() -> (std::fs::File, tracing::subscriber::DefaultGuard) {
+ let file = tempfile::tempfile().unwrap();
+ let subscriber = tracing_subscriber::fmt()
+ .with_writer(file.try_clone().unwrap())
+ .with_max_level(tracing::Level::WARN)
+ .with_ansi(false)
+ .without_time()
+ .finish();
+ (file, tracing::subscriber::set_default(subscriber))
+}
+
+fn model_log_text(file: &mut std::fs::File) -> String {
+ use std::io::{Read, Seek};
+ file.rewind().unwrap();
+ let mut text = String::new();
+ file.read_to_string(&mut text).unwrap();
+ text
+}
+
struct Fixture {
services: maka_host_runtime::application::NativeServices,
_dir: tempfile::TempDir,
diff --git a/plugins/agent/native/src/owner/tests/selection.rs b/plugins/agent/native/src/owner/tests/selection.rs
index 96096bbc0c..ca2ea4612f 100644
--- a/plugins/agent/native/src/owner/tests/selection.rs
+++ b/plugins/agent/native/src/owner/tests/selection.rs
@@ -547,9 +547,11 @@ async fn model_search_continues_its_query_across_bounded_scan_windows_and_catalo
#[tokio::test]
async fn queued_model_route_change_settles_original_message_without_retargeting() {
tokio::time::timeout(Duration::from_secs(30), async {
+ let (mut capture, _log_guard) = model_log_capture();
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}/v1", listener.local_addr().unwrap());
let fixture = Fixture::new(&base, None).await;
+ let connection = fixture.selection().await.connection_id;
let session = fixture.create("Queued original model").await;
fixture.send(&session).await.unwrap();
let (mut socket, _) = listener.accept().await.unwrap();
@@ -577,6 +579,15 @@ async fn queued_model_route_change_settles_original_message_without_retargeting(
assert_eq!(entry["detail"]["invocation"],serde_json::to_value(&failed.event.invocation).unwrap());
assert_eq!(entry["detail"]["reason"],"message_preparation_failed");
assert!(entry["detail"]["modelFailure"].is_null(),"a deterministic queued route conflict is not a provider failure");
+ let log = model_log_text(&mut capture);
+ let cause = log.lines().find(|line| line.contains("stage=\"accepted_model_observation\"")
+ && line.contains("cause=\"accepted_configuration_changed\""))
+ .expect("accepted route rejection must retain the original configuration-change cause");
+ assert!(cause.contains("session_id=") && cause.contains(&session));
+ assert!(cause.contains("connection_id=") && cause.contains(&connection));
+ for secret in ["private-fixture-key", base.as_str(), "http://127.0.0.1:1/v1", "Hello"] {
+ assert!(!log.contains(secret), "accepted route warning must not dump either configuration or the queued message");
+ }
}).await.unwrap();
}
@@ -706,6 +717,92 @@ async fn restored_thinking_preserves_explicit_choice_and_refuses_changed_default
fixture.config.shutdown().await.unwrap();
}
+#[tokio::test]
+async fn unsupported_thinking_logs_original_provider_cause_without_admitting_execution() {
+ use maka_agent_facts::execution::ThinkingLevel::Low;
+ tokio::time::timeout(Duration::from_secs(30), async {
+ const MESSAGE: &str = "submitted-message-must-not-enter-model-log";
+ let (mut capture, _log_guard) = model_log_capture();
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let base = format!("http://{}/v1", listener.local_addr().unwrap());
+ let fixture = Fixture::new(&base, None).await;
+ thinking(&fixture, vec![Low], Low).await;
+ let session = fixture.create("Provider preparation failure").await;
+ let selection = fixture.selection().await;
+ let connection = selection.connection_id.clone();
+ let result = fixture
+ .mutation(
+ 2,
+ "send",
+ json!({
+ "sessionId": session,
+ "message": MESSAGE,
+ "model": selection,
+ "thinkingLevel": "high"
+ }),
+ Uuid::new_v4(),
+ )
+ .await;
+ fixture.drained().await;
+ let facts = fixture
+ .store
+ .scoped_prefix(
+ maka_agent_facts::event::LogScope::Session {
+ id: session.clone(),
+ },
+ 256,
+ 1024 * 1024,
+ )
+ .await
+ .unwrap();
+ fixture.app.shutdown().await.unwrap();
+ fixture.config.shutdown().await.unwrap();
+ let dispatched = match listener.into_std().unwrap().accept() {
+ Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => false,
+ Ok(_) => true,
+ Err(error) => panic!("observe actual provider listener: {error}"),
+ };
+ let log = model_log_text(&mut capture);
+ assert_eq!(result.unwrap_err(), "agent_execution_unavailable");
+ assert!(
+ !facts.events.iter().any(|event| matches!(
+ event.event.fact,
+ maka_agent_facts::event::Fact::InvocationOpened { .. }
+ )),
+ "preparation rejection must not admit an invocation"
+ );
+ assert!(
+ !dispatched,
+ "unsupported thinking must fail before provider HTTP dispatch"
+ );
+ let cause = log
+ .lines()
+ .find(|line| {
+ line.contains("stage=\"provider_prepare\"")
+ && line.contains("cause=\"provider_invalid\"")
+ })
+ .expect(
+ "actual typed provider rejection must be logged before its public cause is erased",
+ );
+ assert!(
+ cause.contains("session_id=") && cause.contains(&session),
+ "warning must identify the existing session: {cause}"
+ );
+ assert!(
+ cause.contains("connection_id=") && cause.contains(&connection),
+ "warning must identify the existing connection: {cause}"
+ );
+ for secret in ["private-fixture-key", base.as_str(), MESSAGE] {
+ assert!(
+ !log.contains(secret),
+ "model cause log must omit credentials, URLs and submitted input"
+ );
+ }
+ })
+ .await
+ .unwrap();
+}
+
#[tokio::test]
async fn invalid_thinking_default_remains_unavailable_without_poisoning_other_models() {
use maka_agent_facts::execution::ThinkingLevel::High;
diff --git a/plugins/agent/native/src/owner/tests/title.rs b/plugins/agent/native/src/owner/tests/title.rs
index 7e401ebdc0..ff1423bdf2 100644
--- a/plugins/agent/native/src/owner/tests/title.rs
+++ b/plugins/agent/native/src/owner/tests/title.rs
@@ -568,10 +568,13 @@ async fn verify_visible_title_failure(case: VisibleTitleFailure) {
#[tokio::test]
async fn title_preparation_failure_is_public_without_a_physical_title_attempt() {
tokio::time::timeout(Duration::from_secs(30),async {
+ let (mut capture, _log_guard) = model_log_capture();
let listener=TcpListener::bind("127.0.0.1:0").await.unwrap();
// Automatic dispatch is disabled; this test calls generation once from
// the actual first opening's frozen source after its connection changes.
- let f=Fixture::new(&format!("http://{}/v1",listener.local_addr().unwrap()),None).await;
+ let base = format!("http://{}/v1", listener.local_addr().unwrap());
+ let f=Fixture::new(&base,None).await;
+ let connection = f.selection().await.connection_id;
let session=automatic(&f).await;
let sent=f.send(&session).await.unwrap();
let (mut socket,_)=listener.accept().await.unwrap();
@@ -603,5 +606,14 @@ async fn title_preparation_failure_is_public_without_a_physical_title_attempt()
assert!(entries[0]["sequence"].as_u64().is_some_and(|sequence|sequence>0));
assert!(entries[0]["eventId"].as_str().is_some_and(|id|!id.is_empty()));
assert!(entries[0]["recordedAt"].as_f64().is_some());
+ let log = model_log_text(&mut capture);
+ let cause = log.lines().find(|line| line.contains("stage=\"accepted_model_observation\"")
+ && line.contains("cause=\"accepted_source_unavailable\""))
+ .expect("title preparation must retain its unavailable accepted-source cause");
+ assert!(cause.contains("session_id=") && cause.contains(&session));
+ assert!(cause.contains("connection_id=") && cause.contains(&connection));
+ for secret in ["private-fixture-key", base.as_str(), "Hello"] {
+ assert!(!log.contains(secret), "title preparation warning must omit credentials, URL and original input");
+ }
}).await.unwrap();
}
diff --git a/products/maka/cli/notices.json b/products/maka/cli/notices.json
index 2fb7e184c0..906cf62bb2 100644
--- a/products/maka/cli/notices.json
+++ b/products/maka/cli/notices.json
@@ -1,6 +1,6 @@
{
"inputs": {
- "Cargo.lock": "8758f79e84baa5c681b2d57c14974d3bc8a8d285a641ae1b9b81cb27ea023e09",
+ "Cargo.lock": "5c8de3b9e95f32bfc6c62b9cb79ac109cd2a2283cd0fc9365b607ca10c343a94",
"Cargo.toml": "9a2a29f608172df146deb8fea723605a6c66c3ca7af4871fffc6cc2509f3cefb",
"package-lock.json": "a5af15aca21fabe0e1bdd48784167df4dc81cee17777d54861dcbd641168a4eb",
"package.json": "38baf3beeb5876849ca16e31868d59bfaee769f260c7555b4a8a1847b92e28a1",
@@ -133,7 +133,7 @@
"plugins/agent/insights/Cargo.toml": "e11b391c1304bb27d552322e9f31b9e1858340f81f49ebbb3d484b02c9238def",
"plugins/agent/model/Cargo.toml": "0fbd5b92911a7d3af54f10069d16b1b1b4b2d453b79aeb6ea63e66fd78d6ad85",
"plugins/agent/model/tests/fixtures/provider-errors.mjs": "295a7dec85006792450b8675d21b5628dfba215ef6f51ccd44ad427159a4f4bf",
- "plugins/agent/native/Cargo.toml": "49cff06df21f0dddf7339de700d8e7f88fad3cda24022389854bce316a1e1d39",
+ "plugins/agent/native/Cargo.toml": "f732629932213eeb1dc0ec20db8a125d3e37ddf363651e850d7e1444e9961f5e",
"plugins/agent/navigation/desktop.ts": "7d4633828f8bf64aeac60f078be8328407c6be83b1de8b7926c3796ee32fad15",
"plugins/agent/navigation/package.ts": "eac5fb178bf93bc3d40cb1e9e184707cf3095b026da8c28c5682e636d0242bed",
"plugins/agent/navigation/tui.ts": "fbfefb89c7265f5e0e740ca81d05b300fd3d98eb26474e6ae6114245972ad75d",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org