diff --git a/TASKS.md b/TASKS.md index a7d4fdc..c067b31 100644 --- a/TASKS.md +++ b/TASKS.md @@ -62,9 +62,11 @@ Progress: - `ExecutionConfig.response_cache` is now part of the public operation model, and `admin-api` validates the safe Community baseline: only explicit `REST GET` operations without `auth_profile_ref` - cache boundary and namespace rules are now documented for `platform / coordination cache` versus `response cache` - runtime executor now performs actual response cache reads/writes for eligible `REST GET` operations with keys isolated by `workspace + agent + operation + request fingerprint` + - mcp-server published tool catalog now uses shared coordination snapshots across instances, while preserving the same `workspace + agent` isolation model - pending: - preserve the same isolation model while extending response cache beyond the first `REST GET` hot path - - wire replay guard / coordination state into additional concrete runtime and transport paths + - wire replay guard into the first real token or nonce flow once commercial machine-token issuance stops being a `Community` stub + - wire coordination state into additional runtime or transport paths beyond the shared published catalog snapshot ## Planned diff --git a/apps/mcp-server/src/app.rs b/apps/mcp-server/src/app.rs index 71bab9f..52df5fe 100644 --- a/apps/mcp-server/src/app.rs +++ b/apps/mcp-server/src/app.rs @@ -20,9 +20,10 @@ use axum::{ }; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use crank_core::{ - AsyncJobHandle, AsyncJobId, AuthProfile, InvocationLevel, InvocationLog, InvocationLogId, - InvocationSource, InvocationStatus, JobStatus, OperationSecurityLevel, PlatformApiKeyScope, - SecretId, StreamSession, StreamSessionId, StreamStatus, + AsyncJobHandle, AsyncJobId, AuthProfile, CoordinationStateStore, InvocationLevel, + InvocationLog, InvocationLogId, InvocationSource, InvocationStatus, JobStatus, + OperationSecurityLevel, PlatformApiKeyScope, SecretId, StreamSession, StreamSessionId, + StreamStatus, }; use crank_registry::{ CreateAsyncJobRequest, CreateInvocationLogRequest, CreateStreamSessionRequest, @@ -134,12 +135,13 @@ pub fn build_app( secret_crypto: SecretCrypto, runtime: RuntimeExecutor, api_rate_limiter: RequestRateLimiter, + coordination_store: Arc, sessions: SharedSessionStore, credential_verifier: SharedMachineCredentialVerifier, ) -> Router { let state = Arc::new(AppState { registry: registry.clone(), - catalog: PublishedToolCatalog::new(registry, refresh_interval), + catalog: PublishedToolCatalog::new(registry, refresh_interval, coordination_store), runtime, api_rate_limiter, secret_crypto, diff --git a/apps/mcp-server/src/catalog.rs b/apps/mcp-server/src/catalog.rs index 02521b9..f3f73bd 100644 --- a/apps/mcp-server/src/catalog.rs +++ b/apps/mcp-server/src/catalog.rs @@ -4,7 +4,9 @@ use std::{ time::{Duration, Instant}, }; +use crank_core::{CacheScope, CoordinationStateStore, CoordinationStateValue}; use crank_registry::{PostgresRegistry, PublishedAgentTool, RegistryError}; +use serde::{Deserialize, Serialize}; use tokio::sync::RwLock; use tracing::info; @@ -12,6 +14,7 @@ use tracing::info; pub struct PublishedToolCatalog { registry: PostgresRegistry, refresh_interval: Duration, + coordination_store: Arc, cached: Arc>>, } @@ -27,11 +30,21 @@ struct CachedCatalog { tools: Vec, } +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +struct CatalogSnapshot { + tools: Vec, +} + impl PublishedToolCatalog { - pub fn new(registry: PostgresRegistry, refresh_interval: Duration) -> Self { + pub fn new( + registry: PostgresRegistry, + refresh_interval: Duration, + coordination_store: Arc, + ) -> Self { Self { registry, refresh_interval, + coordination_store, cached: Arc::new(RwLock::new(HashMap::new())), } } @@ -68,6 +81,18 @@ impl PublishedToolCatalog { return Ok(()); } + if let Some(tools) = self.load_shared_snapshot(workspace_slug, agent_slug).await { + let mut guard = self.cached.write().await; + guard.insert( + key, + CachedCatalog { + loaded_at: Some(Instant::now()), + tools, + }, + ); + return Ok(()); + } + let tools = match self .registry .get_published_agent_tools_by_slug(workspace_slug, agent_slug) @@ -77,6 +102,8 @@ impl PublishedToolCatalog { Err(RegistryError::PublishedAgentNotFound { .. }) => Vec::new(), Err(error) => return Err(error), }; + self.store_shared_snapshot(workspace_slug, agent_slug, &tools) + .await; let mut guard = self.cached.write().await; let previous_count = guard .get(&key) @@ -104,6 +131,57 @@ impl PublishedToolCatalog { Ok(()) } + + async fn load_shared_snapshot( + &self, + workspace_slug: &str, + agent_slug: &str, + ) -> Option> { + if self.refresh_interval.is_zero() { + return None; + } + + let key = catalog_snapshot_key(workspace_slug, agent_slug); + let value = match self + .coordination_store + .get_value(CacheScope::Coordination, &key) + .await + { + Ok(value) => value?, + Err(_) => return None, + }; + serde_json::from_value::(value.payload) + .ok() + .map(|snapshot| snapshot.tools) + } + + async fn store_shared_snapshot( + &self, + workspace_slug: &str, + agent_slug: &str, + tools: &[PublishedAgentTool], + ) { + let Some(ttl) = catalog_snapshot_ttl(self.refresh_interval) else { + return; + }; + + let payload = match serde_json::to_value(CatalogSnapshot { + tools: tools.to_vec(), + }) { + Ok(payload) => payload, + Err(_) => return, + }; + let key = catalog_snapshot_key(workspace_slug, agent_slug); + let _ = self + .coordination_store + .put_value( + CacheScope::Coordination, + &key, + CoordinationStateValue { payload }, + ttl, + ) + .await; + } } impl CatalogKey { @@ -114,3 +192,15 @@ impl CatalogKey { } } } + +fn catalog_snapshot_key(workspace_slug: &str, agent_slug: &str) -> String { + format!("published_catalog:{workspace_slug}:{agent_slug}") +} + +fn catalog_snapshot_ttl(refresh_interval: Duration) -> Option { + if refresh_interval.is_zero() { + return None; + } + + refresh_interval.checked_mul(2).or(Some(refresh_interval)) +} diff --git a/apps/mcp-server/src/main.rs b/apps/mcp-server/src/main.rs index 1ded0ed..e79744c 100644 --- a/apps/mcp-server/src/main.rs +++ b/apps/mcp-server/src/main.rs @@ -67,6 +67,7 @@ async fn main() -> Result<(), Box> { } else { RequestRateLimiter::new(api_rate_limit) }, + cache_stores.coordination.clone(), std::sync::Arc::new(session_store), std::sync::Arc::new(CommunityMachineCredentialVerifier), ); @@ -162,7 +163,8 @@ mod tests { SaveAgentBindingsRequest, }; use crank_runtime::{ - RequestRateLimitConfig, RequestRateLimiter, RuntimeExecutor, SecretCrypto, + InMemoryCoordinationStateStore, RequestRateLimitConfig, RequestRateLimiter, + RuntimeExecutor, SecretCrypto, }; use crank_schema::{Schema, SchemaKind}; use futures_util::stream; @@ -181,6 +183,7 @@ mod tests { MachineCredentialVerifierError, SharedMachineCredentialVerifier, VerifiedMachineCredential, }, + catalog::PublishedToolCatalog, session::{InMemorySessionStore, SharedSessionStore, TransportSessionStore}, }; @@ -276,6 +279,7 @@ mod tests { SecretCrypto::new("test-master-key").unwrap(), RuntimeExecutor::new(), RequestRateLimiter::new(rate_limit_config), + std::sync::Arc::new(InMemoryCoordinationStateStore::default()), sessions, credential_verifier, ) @@ -1298,6 +1302,63 @@ mod tests { ); } + #[tokio::test] + async fn shares_published_catalog_snapshot_across_instances() { + let registry = test_registry().await; + let upstream_base_url = spawn_upstream_server().await; + let operation = test_operation(&upstream_base_url, "crm_catalog_shared"); + let coordination_store = std::sync::Arc::new(InMemoryCoordinationStateStore::default()); + + registry + .create_operation(&test_workspace_id(), &operation, Some("alice")) + .await + .unwrap(); + registry + .publish_operation(PublishRequest { + workspace_id: &test_workspace_id(), + operation_id: &operation.id, + version: 1, + published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(), + published_by: Some("alice"), + }) + .await + .unwrap(); + publish_agent_for_operation(®istry, &operation, "sales-shared-catalog").await; + + let catalog_a = PublishedToolCatalog::new( + registry.clone(), + Duration::from_secs(60), + coordination_store.clone(), + ); + let tools_a = catalog_a + .list_tools(test_workspace_slug(), "sales-shared-catalog") + .await + .unwrap(); + assert_eq!(tools_a.len(), 1); + + registry + .unpublish_agent( + &test_workspace_id(), + &test_agent_id("sales-shared-catalog"), + &OffsetDateTime::parse("2026-03-26T10:05:00Z", &Rfc3339).unwrap(), + ) + .await + .unwrap(); + + let catalog_b = PublishedToolCatalog::new( + registry.clone(), + Duration::from_secs(60), + coordination_store, + ); + let tools_b = catalog_b + .list_tools(test_workspace_slug(), "sales-shared-catalog") + .await + .unwrap(); + + assert_eq!(tools_b.len(), 1); + assert_eq!(tools_b[0].tool_name, operation.name); + } + #[tokio::test] async fn lists_only_bound_tools_for_agent_context() { let registry = test_registry().await; diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index 528c998..51d584c 100644 --- a/docs/implementation-plan.md +++ b/docs/implementation-plan.md @@ -197,6 +197,7 @@ - platform / coordination state не смешивался с response cache; - response cache по умолчанию изолировался по `workspace + agent + operation + request fingerprint`; - внешний cache backend можно было безопасно шарить между несколькими рабочими областями и агентами. +- shared coordination cache уже применяется не только для rate limiting, но и для multi-instance snapshots published MCP catalogs. ## 14. Этап 11. Live staging and demo readiness diff --git a/docs/runtime-config.md b/docs/runtime-config.md index 820371c..1b305b8 100644 --- a/docs/runtime-config.md +++ b/docs/runtime-config.md @@ -178,6 +178,7 @@ Demo/deployment: - ingress rate limiting; - replay guard; - ephemeral coordination state; +- shared snapshots published MCP catalogs between instances; - будущие short-lived / one-time token helpers. Второй контур хранит только кэшируемые ответы операций.