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 06:07:43 UTC
(maka) branch rust-preview updated: fix(agent): diagnose owner and auxiliary failure custody
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 7dd53afbbb fix(agent): diagnose owner and auxiliary failure custody
7dd53afbbb is described below
commit 7dd53afbbba9703437122d83a3dcfaec581d39e9
Author: Wang <m4...@gmail.com>
AuthorDate: Sat Oct 17 14:07:29 2026 +0800
fix(agent): diagnose owner and auxiliary failure custody
Record safe typed auxiliary persistence causes and distinguish queue quarantine from owner closure with original identities. Preserve shutdown priority, physical custody and unknown settlement recovery.
Generated-by: Codex
---
plugins/agent/native/README.md | 2 +-
plugins/agent/native/README.zh-CN.md | 2 +-
plugins/agent/native/src/diagnostics.rs | 89 +++++++++++++++++++++++
plugins/agent/native/src/lib.rs | 1 +
plugins/agent/native/src/model/diagnostics.rs | 11 +--
plugins/agent/native/src/owner.rs | 19 ++++-
plugins/agent/native/src/owner/auxiliary.rs | 68 +++++++++++++----
plugins/agent/native/src/owner/queue.rs | 40 +++++++++-
plugins/agent/native/src/owner/tests.rs | 69 +++++++++++++++++-
plugins/agent/native/src/owner/tests/selection.rs | 8 +-
plugins/agent/native/src/owner/tests/title.rs | 42 +++++++++--
plugins/agent/native/src/owner/title.rs | 57 ++++++++++-----
12 files changed, 350 insertions(+), 58 deletions(-)
diff --git a/plugins/agent/native/README.md b/plugins/agent/native/README.md
index c11c826d93..6a70c356aa 100644
--- a/plugins/agent/native/README.md
+++ b/plugins/agent/native/README.md
@@ -47,7 +47,7 @@ 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 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. Fatal queue and auxiliary-request errors record safe Host diagnostics with available owner, invocation or requ [...]
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.
diff --git a/plugins/agent/native/README.zh-CN.md b/plugins/agent/native/README.zh-CN.md
index c3e5e818c5..9095966944 100644
--- a/plugins/agent/native/README.zh-CN.md
+++ b/plugins/agent/native/README.zh-CN.md
@@ -47,7 +47,7 @@ Maka 默认组合启用 `automaticTitles`。通过 `automaticTitle: true` 创建
自动标题的准备、模型及无效输出失败会持久记录到会话诊断。空响应或输出达到上限时，保留实际请求用量，同时记录标题生成失败。服务停止及策略性跳过不报告标题失败；持久化或清理结果不确定时，保留原有恢复优先级。
-模型准备和 OAuth 刷新失败在原始原因被转换为公开错误前写入 Host 警告。警告包含安全错误分类及已有的连接／会话标识（若可用），不包含凭据、配置、请求内容或任意 provider 错误文本。再次读取已失败的刷新结果不会重复记录原始失败。
+模型准备和 OAuth 刷新失败在原始原因被转换为公开错误前写入 Host 警告。警告包含安全错误分类及已有的连接／会话标识（若可用），不包含凭据、配置、请求内容或任意 provider 错误文本。再次读取已失败的刷新结果不会重复记录原始失败。队列及辅助请求的致命错误会记录安全 Host 诊断，并关联可用的 owner、原调用或请求身份。诊断区分会话隔离与 owner 关闭；恢复与资源保管仍遵循原有规则。
模型上下文总容量、独立输入上限与压缩阈值随模型选择一起保存，排队和重启后继续使用接收时的值。未显式设置压缩阈值时，使用已知输入容量的 85%；输出预留仍由引擎单独计算。
diff --git a/plugins/agent/native/src/diagnostics.rs b/plugins/agent/native/src/diagnostics.rs
new file mode 100644
index 0000000000..44b99f36bf
--- /dev/null
+++ b/plugins/agent/native/src/diagnostics.rs
@@ -0,0 +1,89 @@
+/*
+ * 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.
+ */
+
+//! Closed diagnostic categories; private error payloads never cross this boundary.
+
+use maka_agent_engine::RunError;
+use maka_agent_facts::event::CommitError;
+use maka_config::ConfigError;
+use maka_event_log::StoreError;
+
+pub(crate) fn config_cause(error: &ConfigError) -> &'static str {
+ 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",
+ }
+}
+
+pub(crate) fn store_cause(error: &StoreError) -> &'static str {
+ match error {
+ StoreError::ConnectionClosed => "store_connection_closed",
+ StoreError::OperationUnknown => "store_operation_unknown",
+ StoreError::CommitUnknown(_) => "store_commit_unknown",
+ StoreError::Sqlx(_) => "store_sqlx",
+ StoreError::Sqlite(_) => "store_sqlite",
+ StoreError::Io(_) => "store_io",
+ StoreError::Json(_) => "store_json",
+ StoreError::Migration(_) => "store_migration",
+ StoreError::Poisoned => "store_poisoned",
+ StoreError::CloseFailed(_) => "store_close_failed",
+ StoreError::StartupCleanupFailed { .. } => "store_startup_cleanup_failed",
+ StoreError::UnsupportedDatabase => "store_unsupported_database",
+ StoreError::Initialization => "store_initialization",
+ _ => "store_error",
+ }
+}
+
+pub(crate) fn store_sql_code(error: &StoreError) -> Option<i32> {
+ match error {
+ StoreError::Sqlx(error) | StoreError::CommitUnknown(error) => error
+ .as_database_error()
+ .and_then(|error| error.code())
+ .and_then(|code| code.parse().ok()),
+ StoreError::Sqlite(error) => error.sqlite_error().map(|error| error.extended_code),
+ _ => None,
+ }
+}
+
+pub(crate) fn run_cause(error: &RunError) -> &'static str {
+ match error {
+ RunError::Store(error) => store_cause(error),
+ RunError::Commit(CommitError::Retired) => "commit_retired",
+ RunError::Commit(CommitError::PermissionRevoked) => "commit_permission_revoked",
+ RunError::Commit(CommitError::Rejected(_)) => "commit_rejected",
+ RunError::Commit(CommitError::OutcomeUnknown(_)) => "commit_outcome_unknown",
+ RunError::Busy => "session_busy",
+ RunError::ContinuationChanged => "continuation_changed",
+ RunError::ReconciliationRequired(_) => "reconciliation_required",
+ RunError::InvalidInput(_) => "invalid_input",
+ RunError::Cancelled => "cancelled",
+ RunError::ModelIncomplete => "model_incomplete",
+ RunError::CleanupUnconfirmed => "cleanup_unconfirmed",
+ RunError::Model(_) => "model_error",
+ RunError::Tool(_) if error.is_cleanup_unconfirmed() => "cleanup_unconfirmed",
+ RunError::Tool(_) => "tool_error",
+ RunError::Internal(_) => "internal_error",
+ }
+}
diff --git a/plugins/agent/native/src/lib.rs b/plugins/agent/native/src/lib.rs
index ed0d8048c5..7e8117f02d 100644
--- a/plugins/agent/native/src/lib.rs
+++ b/plugins/agent/native/src/lib.rs
@@ -21,6 +21,7 @@
//! chooses whether to activate it; installing this crate does not start a product.
/// Public wire types and generated contracts. 公开协议类型与生成契约。
use maka_agent_api::contract;
+mod diagnostics;
pub mod executor;
mod model;
mod owner;
diff --git a/plugins/agent/native/src/model/diagnostics.rs b/plugins/agent/native/src/model/diagnostics.rs
index faed45e8e6..479a3b3935 100644
--- a/plugins/agent/native/src/model/diagnostics.rs
+++ b/plugins/agent/native/src/model/diagnostics.rs
@@ -54,16 +54,7 @@ pub(super) fn provider(
}
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",
- };
+ let cause = crate::diagnostics::config_cause(error);
match error {
ConfigError::Io(error) => tracing::warn!(
stage, cause, connection_id, error_kind = ?error.kind(), os_code = error.raw_os_error(),
diff --git a/plugins/agent/native/src/owner.rs b/plugins/agent/native/src/owner.rs
index 75cd9ec349..fd11d75f76 100644
--- a/plugins/agent/native/src/owner.rs
+++ b/plugins/agent/native/src/owner.rs
@@ -276,7 +276,12 @@ impl Owner {
let configuration = Arc::new(
ConfigurationStore::for_root(root)
.await
- .map_err(|_| "agent_configuration_unavailable")?,
+ .map_err(|error| {
+ tracing::error!(stage = "configuration_store_open", cause = crate::diagnostics::config_cause(&error),
+ activation = %context.activation().epoch(), instance = ?context.activation().instance(),
+ "Agent configuration store failed to open");
+ "agent_configuration_unavailable"
+ })?,
);
let recall = Arc::new(recall::Recall {
store: persistence.store().clone(),
@@ -393,7 +398,12 @@ impl Owner {
configuration
.shutdown()
.await
- .map_err(|_| "agent_configuration_cleanup_unconfirmed")?;
+ .map_err(|error| {
+ tracing::error!(stage = "configuration_store_shutdown", cause = crate::diagnostics::config_cause(&error),
+ activation = %context.activation().epoch(), instance = ?context.activation().instance(),
+ "Agent configuration store cleanup failed");
+ "agent_configuration_cleanup_unconfirmed"
+ })?;
return Err(error);
}
};
@@ -1132,6 +1142,11 @@ impl Owner {
let recap = owner.recap_storage.finish().await;
let auxiliary = owner.auxiliary.resources.finish().await;
let configuration = owner.configuration.shutdown().await;
+ if let Err(error) = &configuration {
+ tracing::error!(stage = "configuration_store_shutdown", cause = crate::diagnostics::config_cause(error),
+ activation = %owner.revision.activation, instance = ?owner.data_source.namespace,
+ "Agent configuration store cleanup failed");
+ }
let runs = owner.runs.lock().await;
if !runs.quarantined.is_empty() {
return Err("agent_cleanup_unconfirmed".into());
diff --git a/plugins/agent/native/src/owner/auxiliary.rs b/plugins/agent/native/src/owner/auxiliary.rs
index 558281b8c1..79a6aaccb8 100644
--- a/plugins/agent/native/src/owner/auxiliary.rs
+++ b/plugins/agent/native/src/owner/auxiliary.rs
@@ -28,6 +28,7 @@ use maka_agent_facts::{
use maka_capabilities::call::Resources;
use maka_event_log::usage::{AuxiliarySource, Outcome};
use maka_model::{ModelRequest, ModelStream, StepBuilder};
+use tracing::Instrument;
pub(super) enum Error {
Model(ModelError),
@@ -84,15 +85,36 @@ impl Auxiliary {
reference,
binding,
session_id,
- } => log
- .begin_auxiliary_operation_model(*reference, binding, session_id, Some(quote))
- .await
- .map_err(|error| match error {
- maka_event_log::StoreError::SessionNotFound
- | maka_event_log::StoreError::SessionRetired
- | maka_event_log::StoreError::SessionBusy => Error::Admission(failed(error)),
- other => Error::Tool(persistence(other)),
- })?,
+ } => {
+ let span = tracing::error_span!("agent_auxiliary_operation", session_id = %session_id, operation_id = %reference.operation);
+ async {
+ log.begin_auxiliary_operation_model(
+ *reference,
+ binding,
+ session_id,
+ Some(quote),
+ )
+ .await
+ .map_err(|error| match error {
+ maka_event_log::StoreError::SessionNotFound
+ | maka_event_log::StoreError::SessionRetired
+ | maka_event_log::StoreError::SessionBusy => {
+ Error::Admission(failed(error))
+ }
+ other => {
+ tracing::error!(
+ stage = "auxiliary_model_admission_failed",
+ cause = crate::diagnostics::store_cause(&other),
+ sql_code = crate::diagnostics::store_sql_code(&other),
+ "Auxiliary model admission failed"
+ );
+ Error::Tool(persistence(other))
+ }
+ })
+ }
+ .instrument(span)
+ .await?
+ }
Admission::Source(source) => log
.begin_auxiliary_model(source, Some(quote))
.await
@@ -102,7 +124,15 @@ impl Auxiliary {
| maka_event_log::StoreError::SessionRetired
| maka_event_log::StoreError::SessionBusy
| maka_event_log::StoreError::PrefixTooLarge => Error::Admission(failed(error)),
- _ => Error::Tool(persistence(error)),
+ _ => {
+ tracing::error!(
+ stage = "auxiliary_model_admission_failed",
+ cause = crate::diagnostics::store_cause(&error),
+ sql_code = crate::diagnostics::store_sql_code(&error),
+ "Auxiliary model admission failed"
+ );
+ Error::Tool(persistence(error))
+ }
})?,
};
let model_id = request.provider.model.clone();
@@ -151,7 +181,11 @@ impl Auxiliary {
};
log.settle_auxiliary_model(id, outcome)
.await
- .map_err(persistence)?;
+ .map_err(|error| {
+ tracing::error!(stage = "auxiliary_model_settlement_failed", cause = crate::diagnostics::store_cause(&error),
+ sql_code = crate::diagnostics::store_sql_code(&error), request_id = %id, "Auxiliary model settlement failed");
+ persistence(error)
+ })?;
result
}
}
@@ -174,7 +208,11 @@ async fn receive(
{
log.observe_auxiliary_model(id, usage.clone())
.await
- .map_err(persistence)?;
+ .map_err(|error| {
+ tracing::error!(stage = "auxiliary_model_observation_failed", cause = crate::diagnostics::store_cause(&error),
+ sql_code = crate::diagnostics::store_sql_code(&error), request_id = %id, "Auxiliary model usage observation failed");
+ persistence(error)
+ })?;
observed_usage = true;
}
bytes = bytes.saturating_add(
@@ -198,7 +236,11 @@ async fn receive(
stream
.cancel_and_wait()
.await
- .map_err(|error| ToolError::CleanupUnconfirmed(error.to_string()))?;
+ .map_err(|error| {
+ tracing::error!(stage = "auxiliary_model_cleanup_unconfirmed", cause = "cleanup_unconfirmed",
+ request_id = %id, "Auxiliary model stream cleanup is unconfirmed");
+ ToolError::CleanupUnconfirmed(error.to_string())
+ })?;
result
}
pub(super) fn persistence(error: impl std::fmt::Display) -> ToolError {
diff --git a/plugins/agent/native/src/owner/queue.rs b/plugins/agent/native/src/owner/queue.rs
index 977c57b99e..bc7ad20f8c 100644
--- a/plugins/agent/native/src/owner/queue.rs
+++ b/plugins/agent/native/src/owner/queue.rs
@@ -57,7 +57,19 @@ impl Owner {
if runs.closing {
return;
}
- if self.schedule_locked(&mut runs).await.is_err() {
+ if let Err(error) = self.schedule_locked(&mut runs).await {
+ if !(self.stop.is_cancelled()
+ && matches!(
+ &error,
+ RunError::Cancelled
+ | RunError::Commit(maka_agent_facts::event::CommitError::Retired)
+ | RunError::Store(StoreError::SessionRetired)
+ ))
+ {
+ tracing::error!(stage = "queue_schedule_failed", cause = crate::diagnostics::run_cause(&error),
+ consequence = "owner_closing", activation = %self.revision.activation, instance = ?self.data_source.namespace,
+ "Agent queue scheduling failed");
+ }
runs.closing = true;
self.stop.cancel();
}
@@ -125,11 +137,21 @@ impl Owner {
Box::pin(async move {
let session = running.invocation().session_id.clone();
loop {
+ let invocation = running.invocation().clone();
let completion = std::panic::AssertUnwindSafe(running.completion())
.catch_unwind()
.await;
let mut runs = self.runs.lock().await;
if !matches!(completion, Ok(ref value) if value.cleanup.is_ok()) {
+ let stage = if completion.is_err() {
+ "queue_completion_panicked"
+ } else {
+ "queue_cleanup_unconfirmed"
+ };
+ tracing::error!(stage, cause = "cleanup_unconfirmed", consequence = "session_quarantined",
+ session_id = %invocation.session_id, turn_id = %invocation.turn_id,
+ run_id = %invocation.run_id, invocation_id = %invocation.invocation_id,
+ activation = %self.revision.activation, "Agent invocation cleanup is unconfirmed");
runs.controls
.get_mut(&session)
.expect("original worker control")
@@ -156,6 +178,22 @@ impl Owner {
return;
}
Err(error) => {
+ if !((self.stop.is_cancelled() || runs.closing)
+ && matches!(
+ &error,
+ RunError::Cancelled
+ | RunError::Commit(
+ maka_agent_facts::event::CommitError::Retired
+ )
+ | RunError::Store(StoreError::SessionRetired)
+ ))
+ {
+ tracing::error!(stage = "queue_advance_failed", cause = crate::diagnostics::run_cause(&error),
+ consequence = "owner_closing", session_id = %invocation.session_id,
+ turn_id = %invocation.turn_id, run_id = %invocation.run_id,
+ invocation_id = %invocation.invocation_id, activation = %self.revision.activation,
+ "Agent queue advancement failed after invocation completion");
+ }
runs.controls
.get_mut(&session)
.expect("original worker control")
diff --git a/plugins/agent/native/src/owner/tests.rs b/plugins/agent/native/src/owner/tests.rs
index 914954581e..17663ba004 100644
--- a/plugins/agent/native/src/owner/tests.rs
+++ b/plugins/agent/native/src/owner/tests.rs
@@ -444,8 +444,8 @@ 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) {
+// fixture shutdown so actual producer tasks remain inside the capture.
+fn log_capture() -> (std::fs::File, tracing::subscriber::DefaultGuard) {
let file = tempfile::tempfile().unwrap();
let subscriber = tracing_subscriber::fmt()
.with_writer(file.try_clone().unwrap())
@@ -456,7 +456,7 @@ fn model_log_capture() -> (std::fs::File, tracing::subscriber::DefaultGuard) {
(file, tracing::subscriber::set_default(subscriber))
}
-fn model_log_text(file: &mut std::fs::File) -> String {
+fn log_text(file: &mut std::fs::File) -> String {
use std::io::{Read, Seek};
file.rewind().unwrap();
let mut text = String::new();
@@ -1253,11 +1253,23 @@ fn controlled_model(panic: bool) -> (ModelExecutor, Fault) {
}
async fn cleanup_case(panic: bool) {
+ let (mut capture, _log_guard) = log_capture();
let (model, fault) = controlled_model(panic);
let fixture = Fixture::new("http://127.0.0.1:1/v1", Some(model)).await;
*fault.store.lock().unwrap() = Some(fixture.store.clone());
let session = fixture.create("Physical lifetime").await;
fixture.send(&session).await.unwrap();
+ let invocation = fixture
+ .owner
+ .runs
+ .lock()
+ .await
+ .controls
+ .get(&session)
+ .unwrap()
+ .invocation
+ .clone();
+ let activation = fixture.owner.revision.activation.clone();
fault.entered.acquire().await.unwrap().forget();
let selection = fixture.selection().await;
let reference = if panic {
@@ -1321,6 +1333,12 @@ async fn cleanup_case(panic: bool) {
fixture.owner.runs.lock().await.quarantined.len(),
usize::from(panic)
);
+ assert_eq!(
+ fixture.owner.runs.lock().await.closing,
+ !panic,
+ "completion uncertainty quarantines only the session; advancing after SQL shutdown closes the owner"
+ );
+ assert_eq!(fixture.owner.stop.is_cancelled(), !panic);
let close = fixture.app.shutdown().await;
assert_eq!(close.is_err(), panic);
fixture.config.shutdown().await.unwrap();
@@ -1352,6 +1370,51 @@ async fn cleanup_case(panic: bool) {
} else {
drop(RootOwner::open(&root, &namespaces).unwrap());
}
+ let log = log_text(&mut capture);
+ let (stage, cause, consequence) = if panic {
+ (
+ "queue_cleanup_unconfirmed",
+ "cleanup_unconfirmed",
+ "session_quarantined",
+ )
+ } else {
+ (
+ "queue_advance_failed",
+ "store_connection_closed",
+ "owner_closing",
+ )
+ };
+ let event = log
+ .lines()
+ .find(|line| line.contains(&format!("stage=\"{stage}\"")))
+ .expect("actual fatal queue custody outcome must be logged");
+ assert!(event.contains("ERROR"));
+ assert!(event.contains(&format!("cause=\"{cause}\"")));
+ assert!(event.contains(&format!("consequence=\"{consequence}\"")));
+ for identity in [
+ &invocation.session_id,
+ &invocation.turn_id,
+ &invocation.run_id,
+ &invocation.invocation_id,
+ &activation,
+ ] {
+ assert!(
+ event.contains(identity),
+ "fatal event must retain original invocation/activation: {event}"
+ );
+ }
+ for secret in [
+ "controlled original model worker panic",
+ "ordinary model failure",
+ "private-fixture-key",
+ "after failed cleanup",
+ "Hello",
+ ] {
+ assert!(
+ !log.contains(secret),
+ "fatal evidence must exclude panic/error payloads, credentials and submitted text"
+ );
+ }
}
#[tokio::test]
async fn model_panic_and_terminal_sql_failure_quarantine_execution_and_keep_root() {
diff --git a/plugins/agent/native/src/owner/tests/selection.rs b/plugins/agent/native/src/owner/tests/selection.rs
index ca2ea4612f..1cbb19fa5d 100644
--- a/plugins/agent/native/src/owner/tests/selection.rs
+++ b/plugins/agent/native/src/owner/tests/selection.rs
@@ -547,7 +547,7 @@ 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 (mut capture, _log_guard) = 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;
@@ -579,7 +579,7 @@ 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 log = 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");
@@ -722,7 +722,7 @@ async fn unsupported_thinking_logs_original_provider_cause_without_admitting_exe
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 (mut capture, _log_guard) = 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;
@@ -762,7 +762,7 @@ async fn unsupported_thinking_logs_original_provider_cause_without_admitting_exe
Ok(_) => true,
Err(error) => panic!("observe actual provider listener: {error}"),
};
- let log = model_log_text(&mut capture);
+ let log = log_text(&mut capture);
assert_eq!(result.unwrap_err(), "agent_execution_unavailable");
assert!(
!facts.events.iter().any(|event| matches!(
diff --git a/plugins/agent/native/src/owner/tests/title.rs b/plugins/agent/native/src/owner/tests/title.rs
index ff1423bdf2..5d5db0a525 100644
--- a/plugins/agent/native/src/owner/tests/title.rs
+++ b/plugins/agent/native/src/owner/tests/title.rs
@@ -406,14 +406,23 @@ async fn title_host_shutdown_awaits_socket_cleanup_and_restart_does_not_replay()
#[tokio::test]
async fn title_settlement_fault_fences_owner_and_preserves_original_usage_for_recovery() {
tokio::time::timeout(Duration::from_secs(30), async {
+ const SQL_SECRET: &str = "SECRET_TITLE_SETTLEMENT_SQL_PAYLOAD";
+ let (mut capture, _log_guard) = log_capture();
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
- let f = fixture(&format!("http://{}/v1", listener.local_addr().unwrap())).await;
+ let base = format!("http://{}/v1", listener.local_addr().unwrap());
+ let f = fixture(&base).await;
+ let activation = f.owner.revision.activation.clone();
let session = automatic(&f).await;
- f.send(&session).await.unwrap();
+ let sent = f.send(&session).await.unwrap();
let ((_, mut main), (_, mut title)) = pair(&listener).await;
let raw = rusqlite::Connection::open(f.root.join(maka_event_log::root::ROOT_DATABASE)).unwrap();
- raw.execute_batch("CREATE TRIGGER hold_title_settlement BEFORE INSERT ON event_log WHEN NEW.kind='auxiliary_model_settled' BEGIN SELECT RAISE(ABORT, 'fixture title settlement failure'); END;").unwrap();
+ raw.execute_batch(&format!("CREATE TRIGGER hold_title_settlement BEFORE INSERT ON event_log WHEN NEW.kind='auxiliary_model_settled' BEGIN SELECT RAISE(ABORT, '{SQL_SECRET}'); END;")).unwrap();
response(&mut main).await;
+ let completed = loop {
+ let boundary = f.store.invocation_boundary(&session, sent["invocationId"].as_str().unwrap()).await.unwrap().unwrap().0;
+ if matches!(boundary.state, maka_event_log::turns::InvocationState::Ended { .. }) { break boundary; }
+ tokio::task::yield_now().await;
+ };
response_with_text(&mut title, "Uncommitted title").await;
f.drained().await;
assert!(f.owner.runs.lock().await.closing);
@@ -432,6 +441,29 @@ async fn title_settlement_fault_fences_owner_and_preserves_original_usage_for_re
assert_eq!(auxiliary.usage.input_tokens, Some(1));
assert_eq!(auxiliary.usage.output_tokens, Some(2));
f.config.shutdown().await.unwrap();
+ assert!(matches!(completed.state, maka_event_log::turns::InvocationState::Ended {
+ outcome: maka_agent_facts::event::InvocationOutcome::Completed, ..
+ }));
+ assert!(matches!(&auxiliary.origin, Origin::Auxiliary {
+ source: AuxiliarySource::SessionTitle { invocation }
+ } if invocation == &completed.invocation));
+ let log = log_text(&mut capture);
+ let settlement = log.lines().find(|line| line.contains("stage=\"auxiliary_model_settlement_failed\""))
+ .expect("typed SQL settlement failure must be logged before it becomes a persistence string");
+ assert!(settlement.contains("ERROR"));
+ assert!(settlement.contains("cause=\"store_sqlx\""));
+ assert!(settlement.contains("sql_code=1811"));
+ assert!(settlement.contains("request_id=") && settlement.contains(&auxiliary.request_id));
+ let fatal = log.lines().find(|line| line.contains("stage=\"title_worker_persistence_failed\""))
+ .expect("title worker must log its actual owner-closing consequence");
+ assert!(fatal.contains("ERROR"));
+ assert!(fatal.contains("consequence=\"owner_closing\""));
+ for identity in [&completed.invocation.session_id, &completed.invocation.turn_id, &completed.invocation.run_id, &completed.invocation.invocation_id, &activation] {
+ assert!(fatal.contains(identity), "title fatal event must retain its triggering invocation/activation: {fatal}");
+ }
+ for secret in [SQL_SECRET, "private-fixture-key", "Uncommitted title", "Hello", base.as_str()] {
+ assert!(!log.contains(secret), "typed cause and custody events must exclude SQL messages, credentials, input, title and URLs");
+ }
}).await.unwrap();
}
@@ -568,7 +600,7 @@ 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 (mut capture, _log_guard) = 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.
@@ -606,7 +638,7 @@ 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 log = 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");
diff --git a/plugins/agent/native/src/owner/title.rs b/plugins/agent/native/src/owner/title.rs
index d195959ef3..8c3a9e732e 100644
--- a/plugins/agent/native/src/owner/title.rs
+++ b/plugins/agent/native/src/owner/title.rs
@@ -29,6 +29,7 @@ use maka_agent_facts::{
use maka_agent_ports::session::title::Policy;
use maka_event_log::usage::AuxiliarySource;
use maka_model::{ModelRequest, prompt::Message};
+use tracing::Instrument;
#[cfg(test)]
pub(super) struct WriteGate {
@@ -77,29 +78,49 @@ impl Owner {
return;
};
let owner = self.clone();
- self.workers.spawn(async move {
- ticket.start();
- let result = std::panic::AssertUnwindSafe(owner.generate_title(invocation, model))
- .catch_unwind()
- .await
- .unwrap_or_else(|_| {
+ let span = tracing::error_span!("agent_title_worker", session_id = %invocation.session_id,
+ turn_id = %invocation.turn_id, run_id = %invocation.run_id,
+ invocation_id = %invocation.invocation_id, activation = %self.revision.activation);
+ self.workers.spawn(
+ async move {
+ ticket.start();
+ let completion =
+ std::panic::AssertUnwindSafe(owner.generate_title(invocation, model))
+ .catch_unwind()
+ .await;
+ let fatal = match &completion {
+ Err(_) => Some(("title_worker_panicked", "worker_panicked")),
+ Ok(Err(ToolError::Persistence(_))) => {
+ Some(("title_worker_persistence_failed", "persistence_failed"))
+ }
+ Ok(Err(ToolError::CleanupUnconfirmed(_))) => {
+ Some(("title_worker_cleanup_unconfirmed", "cleanup_unconfirmed"))
+ }
+ _ => None,
+ };
+ let result = completion.unwrap_or_else(|_| {
Err(ToolError::CleanupUnconfirmed(
"Auxiliary title worker disappeared".into(),
))
});
- let cleanup = match &result {
- Err(ToolError::CleanupUnconfirmed(error)) => Err(error.clone()),
- _ => Ok(()),
- };
- ticket.complete(cleanup);
- if matches!(
- result,
- Err(ToolError::Persistence(_) | ToolError::CleanupUnconfirmed(_))
- ) {
- owner.runs.lock().await.closing = true;
- owner.stop.cancel();
+ let cleanup = match &result {
+ Err(ToolError::CleanupUnconfirmed(error)) => Err(error.clone()),
+ _ => Ok(()),
+ };
+ ticket.complete(cleanup);
+ if let Some((stage, cause)) = fatal {
+ tracing::error!(
+ stage,
+ cause,
+ consequence = "owner_closing",
+ "Agent title worker failed"
+ );
+ owner.runs.lock().await.closing = true;
+ owner.stop.cancel();
+ }
}
- });
+ .instrument(span),
+ );
}
pub(super) async fn generate_title(
&self,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org