diff --git a/crates/khive-mcp/docs/api/coordinator.md b/crates/khive-mcp/docs/api/coordinator.md index 22a5ebd77..318a5d5b2 100644 --- a/crates/khive-mcp/docs/api/coordinator.md +++ b/crates/khive-mcp/docs/api/coordinator.md @@ -42,7 +42,7 @@ operation is successful but its operation envelope also carries: "missing_backends": ["archive"], "backend_errors": { "archive": { - "kind": "backend_error", + "kind": "timeout", "message": "backend search timed out after 5000ms" } } @@ -69,6 +69,32 @@ through `backend_errors_truncated` and `backend_errors_omitted`. Backend ids and messages are credential-masked before exposure; changed backend ids carry a stable hash suffix and `backend_id_masked: true`, ids are capped at 256 Unicode scalar values, and messages are capped at 1,024 Unicode scalar values. +Each retained cause is typed as `timeout` for coordinator and typed runtime +deadline failures, or `backend_error` otherwise. A degraded-empty +`search_incomplete` error reports `retryable: true` only when every failed leg +is a timeout; any mixed or non-timeout failure keeps it false. This policy is +computed before diagnostic truncation, so an omitted cause cannot change the +classification. Every `retryable: true` error also carries `retry_after_ms`. +The server computes that pace from the full pre-truncation failure set as +2,000ms plus 250ms for each failed backend after the first, capped at 10,000ms. + +A conforming client that acts on `retryable: true` uses this published policy: + +- At most three total attempts per logical request (the first attempt plus no + more than two reissues). +- Before reissue number _n_ (starting at 1), wait a base of + `max(retry_after_ms, 2000) * 2^(n - 1)` milliseconds plus nonnegative random + jitter of at most half that base. Jitter must never shorten the server-named + delay. +- Key a circuit breaker by the failed backend set. Open it after three + consecutive all-timeout `search_incomplete` outcomes for that same set, + suppressing first attempts as well as retries for 30 seconds. Then admit one + half-open probe: a successful search closes the breaker; another all-timeout + outcome reopens it for 30 seconds. + +The server publishes and names this pace but cannot enforce client-side +admission. Clients that do not implement the complete budget, backoff, and +breaker policy must not act on `retryable: true` automatically. ## `t6d` — malformed `tags` must reject, not silently drop the filter diff --git a/crates/khive-mcp/src/coordinator.rs b/crates/khive-mcp/src/coordinator.rs index 46dc87a4e..e827d70e2 100644 --- a/crates/khive-mcp/src/coordinator.rs +++ b/crates/khive-mcp/src/coordinator.rs @@ -62,6 +62,47 @@ impl From for khive_runtime::RuntimeError { } } +/// Stable classification for a failed fan-out backend leg. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum BackendSearchFailureKind { + /// The backend operation failed for a non-timeout reason. + BackendError, + /// The backend exceeded the coordinator's per-request deadline. + Timeout, +} + +impl BackendSearchFailureKind { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::BackendError => "backend_error", + Self::Timeout => "timeout", + } + } +} + +/// Typed failure for one fan-out backend leg. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BackendSearchFailure { + pub kind: BackendSearchFailureKind, + pub message: String, +} + +impl BackendSearchFailure { + pub fn backend(message: impl Into) -> Self { + Self { + kind: BackendSearchFailureKind::BackendError, + message: message.into(), + } + } + + pub fn timeout(message: impl Into) -> Self { + Self { + kind: BackendSearchFailureKind::Timeout, + message: message.into(), + } + } +} + /// Per-backend contribution to a fan-out search. pub struct BackendSearchResult { pub backend_id: BackendId, @@ -71,8 +112,9 @@ pub struct BackendSearchResult { pub vector_selected: bool, /// Populated when this backend errored during the fan-out. A whole-backend /// failure (e.g. the text arm, or a fatal error before either arm ran) — - /// this backend contributed no hits at all. - pub error: Option, + /// this backend contributed no hits at all. The typed cause is what the + /// classifier reads; it never parses a rendered message. + pub error: Option, /// Populated when only the vector arm failed and the text arm still ran: /// `entity_hits` still carries the text arm's results, and `error` above /// stays `None`. @@ -379,7 +421,7 @@ pub(crate) mod tests { entity_hits: vec![], note_hits: vec![], vector_selected: true, - error: Some("injected search failure".to_string()), + error: Some(BackendSearchFailure::backend("injected search failure")), vector_error: None, }), ) diff --git a/crates/khive-mcp/src/server.rs b/crates/khive-mcp/src/server.rs index f5c126a03..fc5dd7d7c 100644 --- a/crates/khive-mcp/src/server.rs +++ b/crates/khive-mcp/src/server.rs @@ -45,7 +45,7 @@ use khive_types::RefusalReason; use khive_storage::{EdgeRelation, StorageCapability}; -use crate::coordinator::{CoordSearchResult, CoordinatorService}; +use crate::coordinator::{BackendSearchFailureKind, CoordSearchResult, CoordinatorService}; use crate::tools::request::RequestParams; const _: () = assert!( @@ -71,6 +71,20 @@ const _: () = assert!( "the search diagnostic budget must retain at least one backend cause" ); const MISSING_BACKEND_ERROR_MESSAGE: &str = "backend search failed without diagnostic detail"; +/// Server-named retry pace for ADR-130 Amendment 2. One failed backend uses +/// the published 2s floor; a wider all-timeout outage adds 250ms per extra +/// failed leg, capped at 10s. The full failure set (including omitted +/// diagnostics) controls the value. +const SEARCH_RETRY_AFTER_FLOOR_MS: u64 = 2_000; +const SEARCH_RETRY_AFTER_PER_FAILED_BACKEND_MS: u64 = 250; +const SEARCH_RETRY_AFTER_CEILING_MS: u64 = 10_000; + +fn search_retry_after_ms(failed_backend_count: usize) -> u64 { + let extra_backends = u64::try_from(failed_backend_count.saturating_sub(1)).unwrap_or(u64::MAX); + SEARCH_RETRY_AFTER_FLOOR_MS + .saturating_add(extra_backends.saturating_mul(SEARCH_RETRY_AFTER_PER_FAILED_BACKEND_MS)) + .min(SEARCH_RETRY_AFTER_CEILING_MS) +} /// Per-operation completeness discriminator for the `search` verb (ADR-130 /// §1). `SearchDegradation::status == None` means "not a search op" — no @@ -146,6 +160,7 @@ impl SearchArmParticipation { #[derive(Clone, Debug, PartialEq, Eq)] struct BackendErrorDiagnostic { + kind: BackendSearchFailureKind, message: String, backend_id_masked: bool, backend_id_truncated: bool, @@ -155,7 +170,9 @@ struct BackendErrorDiagnostic { #[derive(Debug, Default)] struct SearchDegradation { status: Option, + retryable: bool, arm_participation: Option, + retry_after_ms: Option, missing_backends: Vec, backend_errors: BTreeMap, backend_errors_omitted: usize, @@ -173,7 +190,9 @@ impl SearchDegradation { arm_participation.observe_result(result); Self { status: Some(SearchStatus::Complete), + retryable: false, arm_participation: Some(arm_participation), + retry_after_ms: None, missing_backends: Vec::new(), backend_errors: BTreeMap::new(), backend_errors_omitted: 0, @@ -237,18 +256,26 @@ impl SearchDegradation { .iter() .filter(|backend| backend.error.is_some()) .count(); + let retryable = failed_backend_count > 0 + && result + .per_backend + .iter() + .filter_map(|backend| backend.error.as_ref()) + .all(|failure| failure.kind == BackendSearchFailureKind::Timeout); + let retry_after_ms = retryable.then(|| search_retry_after_ms(failed_backend_count)); let mut candidates = BTreeMap::new(); - for (backend, error) in result + for (backend, failure) in result .per_backend .iter() - .filter_map(|backend| backend.error.as_deref().map(|error| (backend, error))) + .filter_map(|backend| backend.error.as_ref().map(|failure| (backend, failure))) { let (key, backend_id_masked, backend_id_truncated, backend_id_chars) = bounded_backend_error_key(backend.backend_id.as_str()); candidates.insert( key, BackendErrorDiagnostic { - message: bounded_backend_error_message(error), + kind: failure.kind, + message: bounded_backend_error_message(&failure.message), backend_id_masked, backend_id_truncated, backend_id_chars, @@ -265,7 +292,9 @@ impl SearchDegradation { candidate.insert(backend, diagnostic); let candidate_degradation = Self { status: Some(SearchStatus::Partial), + retryable, arm_participation: Some(arm_participation), + retry_after_ms, missing_backends: candidate.keys().cloned().collect(), backend_errors_omitted: failed_backend_count.saturating_sub(candidate.len()), backend_errors: candidate.clone(), @@ -308,7 +337,9 @@ impl SearchDegradation { } Self { status: Some(status), + retryable, arm_participation: Some(arm_participation), + retry_after_ms, missing_backends, backend_errors, backend_errors_omitted, @@ -406,7 +437,7 @@ fn backend_errors_value(errors: &BTreeMap) -> Va .iter() .map(|(backend, diagnostic)| { let mut value = json!({ - "kind": "backend_error", + "kind": diagnostic.kind.as_str(), "message": diagnostic.message, }); if diagnostic.backend_id_masked { @@ -423,16 +454,24 @@ fn backend_errors_value(errors: &BTreeMap) -> Va } fn search_diagnostic_value(degradation: &SearchDegradation) -> Value { + debug_assert_eq!( + degradation.retryable, + degradation.retry_after_ms.is_some(), + "retryable search failures must always carry a server-named pace" + ); let mut value = json!({ "kind": "search_incomplete", "message": "no-match was not established because selected backends failed", - "retryable": false, + "retryable": degradation.retryable, "missing_backends": degradation.missing_backends, "backend_errors": backend_errors_value(°radation.backend_errors), }); if let Some(participation) = degradation.arm_participation { value["arm_participation"] = search_arm_participation_value(participation); } + if let Some(retry_after_ms) = degradation.retry_after_ms { + value["retry_after_ms"] = json!(retry_after_ms); + } if degradation.backend_errors_omitted > 0 { value["backend_errors_truncated"] = Value::Bool(true); value["backend_errors_omitted"] = json!(degradation.backend_errors_omitted); @@ -2901,7 +2940,9 @@ fn ok_envelope(tool: String, success: OpSuccess) -> Value { } = success; let SearchDegradation { status, + retryable: _, arm_participation, + retry_after_ms: _, missing_backends, backend_errors, backend_errors_omitted, @@ -3295,7 +3336,16 @@ partial:true alias. Truncation is explicit through backend_errors_truncated and backend_errors_omitted. When a backend failure leaves no hit standing after filtering, the op instead fails outright with ok:false and error.kind="search_incomplete" while retaining the same diagnostics — that case -must not be read as "no results found." +must not be read as "no results found." Per-backend causes use kind="timeout" +for typed deadline failures and kind="backend_error" otherwise. The incomplete +error is retryable only when every failed backend leg timed out. Every such +error names retry_after_ms: 2000ms plus 250ms per additional failed backend, +capped at 10000ms and computed from the full pre-truncation failure set. +Conforming clients use at most three total attempts per logical request, +exponential backoff with nonnegative jitter (never shorter than the named +pace), and a 30s circuit breaker after three consecutive all-timeout outcomes +for the same backend set; the breaker suppresses first attempts too and admits +one half-open probe after the open interval. Verb discovery: install the `kg` / `gtd` plugins for usage skills. The verbs currently registered on this server (pack-derived) are listed below. Argument @@ -4954,6 +5004,7 @@ impl ServerHandler for KhiveMcpServer { #[cfg(test)] mod tests { use super::*; + use crate::coordinator::BackendSearchFailure; include!("server/plan_tests.rs"); use khive_runtime::Namespace; use khive_storage::{EventFilter, PageRequest}; @@ -7723,10 +7774,10 @@ mod tests { entity_hits: Vec::new(), note_hits: Vec::new(), vector_selected: true, - error: Some(format!( + error: Some(BackendSearchFailure::backend(format!( "backend failure {index}: {}", "\0\"\\".repeat(MAX_BACKEND_ERROR_MESSAGE_CHARS) - )), + ))), vector_error: None, }) .collect(); @@ -7845,7 +7896,7 @@ mod tests { entity_hits: Vec::new(), note_hits: Vec::new(), vector_selected: true, - error: Some("storage unavailable".to_string()), + error: Some(BackendSearchFailure::backend("storage unavailable")), vector_error: None, }], partial: true, @@ -7881,6 +7932,154 @@ mod tests { assert!(degradation.backend_errors[backend].backend_id_masked); } + fn degraded_search_result( + failures: impl IntoIterator, + ) -> CoordSearchResult { + CoordSearchResult { + entity_hits: Vec::new(), + note_hits: Vec::new(), + per_backend: failures + .into_iter() + .map( + |(backend_id, error)| crate::coordinator::BackendSearchResult { + backend_id: khive_runtime::BackendId::parse(backend_id) + .expect("valid backend id"), + entity_hits: Vec::new(), + note_hits: Vec::new(), + vector_selected: false, + error: Some(error), + vector_error: None, + }, + ) + .collect(), + partial: true, + entity_kinds: std::collections::HashMap::new(), + note_kinds: std::collections::HashMap::new(), + entity_created_at: std::collections::HashMap::new(), + note_created_at: std::collections::HashMap::new(), + note_versions: std::collections::HashMap::new(), + note_names: std::collections::HashMap::new(), + } + } + + #[test] + fn search_failure_classification_timeout_only_is_retryable_and_typed() { + let result = degraded_search_result([( + "archive".to_string(), + BackendSearchFailure::timeout("backend search timed out after 5000ms"), + )]); + + let diagnostic = + search_diagnostic_value(&SearchDegradation::from_result(&result, &json!([]))); + + assert_eq!(diagnostic["retryable"], json!(true)); + assert_eq!(diagnostic["retry_after_ms"], json!(2_000)); + assert_eq!( + diagnostic["backend_errors"]["archive"]["kind"], + json!("timeout") + ); + } + + #[test] + fn search_retry_pace_scales_with_the_full_failed_backend_set_and_is_capped() { + assert_eq!(search_retry_after_ms(1), 2_000); + assert_eq!(search_retry_after_ms(2), 2_250); + assert_eq!(search_retry_after_ms(9), 4_000); + assert_eq!(search_retry_after_ms(usize::MAX), 10_000); + } + + #[test] + fn search_failure_classification_does_not_parse_timeout_from_backend_error_text() { + let result = degraded_search_result([( + "archive".to_string(), + BackendSearchFailure::backend("backend search timed out after 5000ms"), + )]); + + let diagnostic = + search_diagnostic_value(&SearchDegradation::from_result(&result, &json!([]))); + + assert_eq!(diagnostic["retryable"], json!(false)); + assert!(diagnostic.get("retry_after_ms").is_none()); + assert_eq!( + diagnostic["backend_errors"]["archive"]["kind"], + json!("backend_error") + ); + } + + #[test] + fn search_failure_classification_mixed_is_not_retryable_and_keeps_each_kind() { + let result = degraded_search_result([ + ( + "archive".to_string(), + BackendSearchFailure::timeout("backend search timed out after 5000ms"), + ), + ( + "main".to_string(), + BackendSearchFailure::backend("storage unavailable"), + ), + ]); + + let diagnostic = + search_diagnostic_value(&SearchDegradation::from_result(&result, &json!([]))); + + assert_eq!(diagnostic["retryable"], json!(false)); + assert!(diagnostic.get("retry_after_ms").is_none()); + assert_eq!( + diagnostic["backend_errors"]["archive"]["kind"], + json!("timeout") + ); + assert_eq!( + diagnostic["backend_errors"]["main"]["kind"], + json!("backend_error") + ); + } + + #[test] + fn search_failure_classification_omitted_non_timeout_still_controls_retryability() { + let mut failures: Vec<(String, BackendSearchFailure)> = (0..MAX_BACKEND_ERROR_ENTRIES + 4) + .map(|index| { + ( + format!("backend-{index:03}"), + BackendSearchFailure::timeout("backend search timed out after 5000ms"), + ) + }) + .collect(); + failures.push(( + "zzzz-hidden-backend".to_string(), + BackendSearchFailure::backend("storage unavailable"), + )); + let degradation = + SearchDegradation::from_result(°raded_search_result(failures), &json!([])); + let diagnostic = search_diagnostic_value(°radation); + + assert!(degradation.backend_errors_omitted > 0); + assert!(!degradation + .backend_errors + .contains_key("zzzz-hidden-backend")); + assert!(degradation + .backend_errors + .values() + .all(|error| error.kind == BackendSearchFailureKind::Timeout)); + assert_eq!(diagnostic["retryable"], json!(false)); + } + + #[test] + fn search_failure_classification_all_timeouts_stays_retryable_when_truncated() { + let failures = (0..MAX_BACKEND_ERROR_ENTRIES + 5).map(|index| { + ( + format!("backend-{index:03}"), + BackendSearchFailure::timeout("backend search timed out after 5000ms"), + ) + }); + let degradation = + SearchDegradation::from_result(°raded_search_result(failures), &json!([])); + let diagnostic = search_diagnostic_value(°radation); + + assert!(degradation.backend_errors_omitted > 0); + assert_eq!(diagnostic["retryable"], json!(true)); + assert!(diagnostic["retry_after_ms"].as_u64().unwrap() > 2_000); + } + #[test] #[serial_test::serial(config_ledger)] fn frame_budget_omission_preserves_complete_search_status() { @@ -9314,6 +9513,7 @@ mod tests { result: json!([{"id": "11111111-1111-1111-1111-111111111111"}]), degradation: SearchDegradation { status: Some(SearchStatus::Partial), + retryable: false, arm_participation: Some(SearchArmParticipation { text: SearchArmEvidence { status: SearchArmStatus::Error, @@ -9324,10 +9524,12 @@ mod tests { candidate_count: 0, }, }), + retry_after_ms: None, missing_backends: vec!["archive".to_string()], backend_errors: BTreeMap::from([( "archive".to_string(), BackendErrorDiagnostic { + kind: BackendSearchFailureKind::BackendError, message: "storage unavailable".to_string(), backend_id_masked: false, backend_id_truncated: false, diff --git a/crates/kkernel/src/coordinator/dispatch.rs b/crates/kkernel/src/coordinator/dispatch.rs index 2c54088a9..ed663010b 100644 --- a/crates/kkernel/src/coordinator/dispatch.rs +++ b/crates/kkernel/src/coordinator/dispatch.rs @@ -10,7 +10,8 @@ use uuid::Uuid; use khive_pack_kg::handlers::{SearchSubstrate, ValidatedSearchRequest}; use khive_runtime::{ - BackendId, EdgeEndpointKind, KhiveRuntime, NoteSearchHit, Resolved, SearchHit, SearchSource, + BackendId, EdgeEndpointKind, KhiveRuntime, NoteSearchHit, Resolved, RuntimeError, SearchHit, + SearchSource, }; use khive_score::DeterministicScore; use khive_storage::EdgeRelation; @@ -74,20 +75,60 @@ pub(super) fn bounded_backend_cause_for_log(message: &str) -> String { bounded } +/// Stable classification for a failed fan-out backend leg. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum BackendSearchFailureKind { + BackendError, + Timeout, +} + +/// Typed failure for one fan-out backend leg. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BackendSearchFailure { + pub kind: BackendSearchFailureKind, + pub message: String, +} + +impl BackendSearchFailure { + fn timeout(timeout_ms: u64) -> Self { + Self { + kind: BackendSearchFailureKind::Timeout, + message: format!("backend search timed out after {timeout_ms}ms"), + } + } + + pub(super) fn from_runtime_error(error: RuntimeError) -> Self { + let kind = if matches!( + &error, + RuntimeError::Storage(khive_storage::StorageError::Timeout { .. }) + | RuntimeError::DeadlineExceeded { .. } + ) { + BackendSearchFailureKind::Timeout + } else { + BackendSearchFailureKind::BackendError + }; + Self { + kind, + message: error.to_string(), + } + } +} + /// Result of a single backend's entity-search contribution to a fan-out. /// /// `hits` may be empty when the backend returned no results. -/// `error` carries a whole-backend failure message (text arm, or a fatal -/// error before either arm ran); a backend that reported one is treated as -/// having contributed no hits at all. `vector_error` instead carries a -/// vector-arm-only failure: the text arm still ran and `hits` still carries -/// its results, so this backend is NOT `error`-failed. +/// `error` carries the backend-specific typed failure for a whole-backend +/// failure (text arm, or a fatal error before either arm ran); a backend that +/// reported one is treated as having contributed no hits at all. +/// `vector_error` instead carries a vector-arm-only failure: the text arm +/// still ran and `hits` still carries its results, so this backend is NOT +/// `error`-failed. #[derive(Debug)] pub struct BackendSearchResult { pub backend_id: BackendId, pub hits: Vec, pub note_hits: Vec, - pub error: Option, + pub error: Option, pub vector_error: Option, } @@ -618,7 +659,7 @@ impl SubstrateCoordinator { backend_id: backend_id.clone(), hits: vec![], note_hits: vec![], - error: Some(e.to_string()), + error: Some(BackendSearchFailure::from_runtime_error(e)), vector_error: None, }; return (vec![], vec![], vec![backend_result]); @@ -685,7 +726,7 @@ impl SubstrateCoordinator { backend_id: backend_id.clone(), hits: vec![], note_hits: vec![], - error: Some(e.to_string()), + error: Some(BackendSearchFailure::from_runtime_error(e)), vector_error: None, }; return (vec![], vec![], vec![backend_result]); @@ -705,7 +746,7 @@ impl SubstrateCoordinator { backend_id: backend_id.clone(), hits: vec![], note_hits: vec![], - error: Some(format!("backend search timed out after {timeout_ms}ms")), + error: Some(BackendSearchFailure::timeout(timeout_ms)), vector_error: None, }; return (vec![], vec![], vec![backend_result]); @@ -759,7 +800,7 @@ impl SubstrateCoordinator { backend_id: backend_id.clone(), hits: vec![], note_hits: vec![], - error: Some(e.to_string()), + error: Some(BackendSearchFailure::from_runtime_error(e)), vector_error: None, }; return (vec![], vec![], vec![backend_result]); @@ -779,7 +820,7 @@ impl SubstrateCoordinator { backend_id: backend_id.clone(), hits: vec![], note_hits: vec![], - error: Some(format!("backend search timed out after {timeout_ms}ms")), + error: Some(BackendSearchFailure::timeout(timeout_ms)), vector_error: None, }; return (vec![], vec![], vec![backend_result]); @@ -995,7 +1036,7 @@ impl SubstrateCoordinator { backend_id, hits: vec![], note_hits: vec![], - error: Some(e.to_string()), + error: Some(BackendSearchFailure::from_runtime_error(e)), vector_error: None, }); } @@ -1012,7 +1053,7 @@ impl SubstrateCoordinator { backend_id: joined_backend_id, hits: vec![], note_hits: vec![], - error: Some(error.to_string()), + error: Some(BackendSearchFailure::from_runtime_error(error)), vector_error: None, }); } @@ -1026,7 +1067,7 @@ impl SubstrateCoordinator { backend_id: joined_backend_id, hits: vec![], note_hits: vec![], - error: Some(format!("backend search timed out after {timeout_ms}ms")), + error: Some(BackendSearchFailure::timeout(timeout_ms)), vector_error: None, }); } @@ -1040,7 +1081,7 @@ impl SubstrateCoordinator { backend_id: joined_backend_id, hits: vec![], note_hits: vec![], - error: Some(format!("backend search timed out after {timeout_ms}ms")), + error: Some(BackendSearchFailure::timeout(timeout_ms)), vector_error: None, }); } diff --git a/crates/kkernel/src/coordinator/mod.rs b/crates/kkernel/src/coordinator/mod.rs index c92786071..566ccbdbe 100644 --- a/crates/kkernel/src/coordinator/mod.rs +++ b/crates/kkernel/src/coordinator/mod.rs @@ -5,7 +5,9 @@ mod locator; mod registry; pub mod service; -pub use dispatch::{BackendSearchResult, SubstrateCoordinator}; +pub use dispatch::{ + BackendSearchFailure, BackendSearchFailureKind, BackendSearchResult, SubstrateCoordinator, +}; pub use locator::LocatorCache; pub use registry::{BackendEntry, BackendRegistrationError, BackendRegistry}; pub use service::SubstrateCoordinatorService; diff --git a/crates/kkernel/src/coordinator/service.rs b/crates/kkernel/src/coordinator/service.rs index 6bfb11df2..f1fd11615 100644 --- a/crates/kkernel/src/coordinator/service.rs +++ b/crates/kkernel/src/coordinator/service.rs @@ -8,15 +8,16 @@ use async_trait::async_trait; use uuid::Uuid; use khive_mcp::coordinator::{ - BackendSearchResult as CoordBackendResult, CoordError, CoordLinkResult, CoordSearchResult, - CoordinatorService, + BackendSearchFailure as CoordBackendFailure, + BackendSearchFailureKind as CoordBackendFailureKind, BackendSearchResult as CoordBackendResult, + CoordError, CoordLinkResult, CoordSearchResult, CoordinatorService, }; use khive_pack_kg::handlers::ValidatedSearchRequest; use khive_runtime::BackendId; use khive_runtime::Namespace; use khive_storage::EdgeRelation; -use super::dispatch::SubstrateCoordinator; +use super::dispatch::{BackendSearchFailureKind, SubstrateCoordinator}; /// `CoordinatorService` wrapper around a [`SubstrateCoordinator`]. /// @@ -176,7 +177,15 @@ impl CoordinatorService for SubstrateCoordinatorService { entity_hits: r.hits, note_hits: r.note_hits, vector_selected, - error: r.error, + error: r.error.map(|failure| CoordBackendFailure { + kind: match failure.kind { + BackendSearchFailureKind::BackendError => { + CoordBackendFailureKind::BackendError + } + BackendSearchFailureKind::Timeout => CoordBackendFailureKind::Timeout, + }, + message: failure.message, + }), vector_error: r.vector_error, } }) diff --git a/crates/kkernel/src/coordinator/tests.rs b/crates/kkernel/src/coordinator/tests.rs index 3f75fb2a4..4fd2446a0 100644 --- a/crates/kkernel/src/coordinator/tests.rs +++ b/crates/kkernel/src/coordinator/tests.rs @@ -8,7 +8,7 @@ use uuid::Uuid; use khive_pack_kg::handlers::ValidatedSearchRequest; use khive_runtime::Namespace as RuntimeNamespace; use khive_runtime::{ - BackendId, KhiveRuntime, NoteSearchHit, PackRegistry, SearchHit, SearchSource, + BackendId, KhiveRuntime, NoteSearchHit, PackRegistry, RuntimeError, SearchHit, SearchSource, VerbRegistryBuilder, }; use khive_score::DeterministicScore; @@ -17,7 +17,10 @@ use khive_storage::EdgeRelation; use khive_types::{namespace::Namespace, SubstrateKind}; use super::dispatch::bounded_backend_cause_for_log; -use super::{BackendRegistry, LocatorCache, SubstrateCoordinator, SubstrateCoordinatorService}; +use super::{ + BackendRegistry, BackendSearchFailure, BackendSearchFailureKind, LocatorCache, + SubstrateCoordinator, SubstrateCoordinatorService, +}; fn memory_runtime() -> Arc { Arc::new(KhiveRuntime::memory().expect("memory runtime")) @@ -1023,6 +1026,27 @@ async fn fan_out_search_caps_merged_note_hits_at_limit() { // ---- MAJ-2: per-backend fan-out search timeout ---- +#[test] +fn backend_search_failure_classifies_typed_runtime_timeouts_without_message_matching() { + let storage_timeout = BackendSearchFailure::from_runtime_error(RuntimeError::Storage( + khive_storage::StorageError::Timeout { + operation: "fts_search".into(), + }, + )); + let deadline = BackendSearchFailure::from_runtime_error(RuntimeError::DeadlineExceeded { + operation: "search".to_string(), + budget_ms: 5_000, + elapsed_ms: 5_001, + }); + let internal = BackendSearchFailure::from_runtime_error(RuntimeError::Internal( + "backend search timed out after 5000ms".to_string(), + )); + + assert_eq!(storage_timeout.kind, BackendSearchFailureKind::Timeout); + assert_eq!(deadline.kind, BackendSearchFailureKind::Timeout); + assert_eq!(internal.kind, BackendSearchFailureKind::BackendError); +} + /// A hung backend's search task must not block the fan-out from returning a /// healthy sibling's results, and must surface a timeout-specific error for /// itself in its `BackendSearchResult`. @@ -1074,10 +1098,11 @@ async fn fan_out_search_hung_backend_times_out_sibling_still_returns() { .expect("hung backend must have a report entry"); let err = hung_report .error - .as_deref() + .as_ref() .expect("hung backend must carry an error"); + assert_eq!(err.kind, BackendSearchFailureKind::Timeout); assert!( - err.contains("timed out"), + err.message.contains("timed out"), "hung backend error must be timeout-specific, got: {err:?}" ); @@ -1124,8 +1149,9 @@ async fn fan_out_search_multiple_hung_backends_share_one_absolute_deadline() { assert_eq!(per_backend.len(), 3); assert!(per_backend.iter().all(|entry| entry .error - .as_deref() - .is_some_and(|error| error.contains("timed out")))); + .as_ref() + .is_some_and(|error| error.kind == BackendSearchFailureKind::Timeout + && error.message.contains("timed out")))); let elapsed = started.elapsed(); assert!( elapsed >= Duration::from_secs(5) && elapsed < Duration::from_secs(6), @@ -1165,8 +1191,9 @@ async fn fan_out_search_rejects_sibling_that_completed_during_interrupt_grace() assert!( report .error - .as_deref() - .is_some_and(|error| error.contains("timed out")), + .as_ref() + .is_some_and(|error| error.kind == BackendSearchFailureKind::Timeout + && error.message.contains("timed out")), "{backend} completed outside the absolute deadline but was accepted: {:?}", report.error ); @@ -1216,10 +1243,11 @@ async fn fan_out_search_single_backend_hung_backend_times_out_entity_substrate() assert_eq!(report.backend_id.as_str(), "hung"); let err = report .error - .as_deref() + .as_ref() .expect("hung single backend must carry an error"); + assert_eq!(err.kind, BackendSearchFailureKind::Timeout); assert!( - err.contains("timed out"), + err.message.contains("timed out"), "single-backend timeout error must be timeout-specific, got: {err:?}" ); } @@ -1314,8 +1342,8 @@ async fn fan_out_search_masks_real_authorization_cause_in_coordinator_warning() assert!( per_backend[0] .error - .as_deref() - .is_some_and(|error| error.contains("sk_live_")), + .as_ref() + .is_some_and(|error| error.message.contains("sk_live_")), "internal result should retain the raw cause until the MCP sanitizer" ); assert!( @@ -1360,10 +1388,11 @@ async fn fan_out_search_single_backend_hung_backend_times_out_note_substrate() { assert_eq!(report.backend_id.as_str(), "hung"); let err = report .error - .as_deref() + .as_ref() .expect("hung single backend must carry an error"); + assert_eq!(err.kind, BackendSearchFailureKind::Timeout); assert!( - err.contains("timed out"), + err.message.contains("timed out"), "single-backend timeout error must be timeout-specific, got: {err:?}" ); } @@ -1990,10 +2019,11 @@ async fn fan_out_panicked_backend_is_explicit_in_per_backend() { .expect("panicked backend remains identified"); let error = panicked .error - .as_deref() + .as_ref() .expect("panicked backend carries an explicit error"); + assert_eq!(error.kind, BackendSearchFailureKind::BackendError); assert!( - error.contains("join failed") && error.contains("panic"), + error.message.contains("join failed") && error.message.contains("panic"), "join error should identify the task panic, got {error:?}" ); assert!(logs.contains("backend search task failed")); @@ -3720,6 +3750,35 @@ async fn t7d_multi_backend_search_session_kind_routes_to_note_substrate() { // ---- MIN-1: SubstrateCoordinatorService hydration seam ---- +#[tokio::test(start_paused = true)] +#[serial_test::serial(config_ledger)] +async fn coordinator_service_preserves_timeout_failure_kind() { + use khive_mcp::coordinator::{ + BackendSearchFailureKind as CoordFailureKind, CoordinatorService, + }; + + let mut backend_reg = BackendRegistry::new(); + backend_reg.register(backend_id("hung"), memory_runtime()); + let service = SubstrateCoordinatorService::new( + SubstrateCoordinator::new(backend_reg).with_hanging_backend("hung"), + ); + let request = validated_kg_search(serde_json::json!({ + "kind": "entity", + "query": "typed timeout seam", + "limit": 10, + })); + + let result = service + .fan_out_search(&request, &Namespace::local(), &[]) + .await; + + let failure = result.per_backend[0] + .error + .as_ref() + .expect("hung backend must carry a typed failure"); + assert_eq!(failure.kind, CoordFailureKind::Timeout); +} + /// The `khive-mcp` row-shape parity test drives `MockCoordinator` with /// pre-populated `entity_kinds`/`note_kinds`/etc. maps, so it never runs /// `SubstrateCoordinatorService`'s own hydration — the per-hit @@ -3775,7 +3834,7 @@ async fn substrate_coordinator_service_hydrates_entity_and_note_metadata() { let entity_errors: Vec<&str> = entity_result .per_backend .iter() - .filter_map(|r| r.error.as_deref()) + .filter_map(|r| r.error.as_ref().map(|error| error.message.as_str())) .collect(); assert!( entity_errors.is_empty(), @@ -3813,7 +3872,7 @@ async fn substrate_coordinator_service_hydrates_entity_and_note_metadata() { let note_errors: Vec<&str> = note_result .per_backend .iter() - .filter_map(|r| r.error.as_deref()) + .filter_map(|r| r.error.as_ref().map(|error| error.message.as_str())) .collect(); assert!( note_errors.is_empty(), diff --git a/docs/guide/api-reference.md b/docs/guide/api-reference.md index 4859451e2..a33512926 100644 --- a/docs/guide/api-reference.md +++ b/docs/guide/api-reference.md @@ -481,6 +481,22 @@ presentation and response-frame compaction. If no hit survives filtering, the operation is `ok: false` with `error.kind="search_incomplete"`; the structured error carries the same diagnostics. `backend_errors_truncated` plus `backend_errors_omitted` explicitly report causes omitted by safety bounds. +When every failed leg is structurally classified as `timeout`, the error has +`retryable: true` and a positive `retry_after_ms`; any non-timeout leg keeps +`retryable: false` and omits `retry_after_ms`. Classification and pacing use the +full pre-truncation failure set. The server pace is 2,000ms plus 250ms for each +failed backend after the first, capped at 10,000ms. + +Clients that act on `retryable: true` are expected to use at most three total +attempts per logical request. Before retry _n_ (starting at 1), wait +`max(retry_after_ms, 2000) * 2^(n - 1)` milliseconds plus nonnegative random +jitter no greater than half that base; never retry sooner than the server's +value. A breaker keyed by the failed backend set opens after three consecutive +all-timeout outcomes, suppresses both first attempts and retries for 30 seconds, +then admits one half-open probe. Success closes the breaker; another +all-timeout outcome reopens it for 30 seconds. A client that does not implement +the complete attempt budget, backoff, and breaker policy must not retry this +error automatically. Every successful KG search also carries `arm_participation` beside `result`; `search_incomplete` carries the same object inside `error`: