perf(runtime): reduce database-bound turn latency

crates/astra-turn-core/src/pipeline/session.rs

@@ -31,6 +31,7 @@ use crate::recovery_state::RecoveryState;
 use crate::session_latches::SessionLatches;
 use crate::shadow_diff::{ShadowDiffResult, diff_pipeline_outputs};
 use crate::working_memory::WorkingMemoryState;
+use std::sync::Arc;
 
 /// Per-turn input provided by the agentic loop to `PipelineSession::run_turn()`.
 pub struct TurnInput<'a> {
@@ -77,6 +78,7 @@ pub struct ShadowTurnOutput {
 /// `run_turn()` before each LLM request and `record_feedback()` after.
 pub struct PipelineSession {
     pipeline: ContextPipeline,
+    static_sections: Option<Arc<StaticSections>>,
     pub stats: PipelineStats,
     pub latches: SessionLatches,
     pub emergent: EmergentContext,
@@ -168,6 +170,7 @@ impl PipelineSession {
     ) -> Self {
         Self {
             pipeline: ContextPipeline::new(config),
+            static_sections: None,
             stats: PipelineStats::default(),
             latches: SessionLatches::default(),
             emergent: EmergentContext::default(),
@@ -193,6 +196,7 @@ impl PipelineSession {
     ) -> Self {
         Self {
             pipeline: ContextPipeline::new(config),
+            static_sections: None,
             stats,
             latches: SessionLatches::default(),
             emergent: EmergentContext::default(),
@@ -224,6 +228,7 @@ impl PipelineSession {
     ) -> Self {
         Self {
             pipeline: ContextPipeline::new(config),
+            static_sections: None,
             stats,
             latches,
             emergent: EmergentContext::default(),
@@ -246,6 +251,18 @@ impl PipelineSession {
         self.turns_completed
     }
 
+    /// Return the immutable prompt sections owned by this pipeline session,
+    /// building them once on first use. The returned `Arc` lets the caller
+    /// borrow the sections while mutably advancing the rest of the session.
+    pub fn static_sections_or_init(
+        &mut self,
+        init: impl FnOnce() -> StaticSections,
+    ) -> Arc<StaticSections> {
+        self.static_sections
+            .get_or_insert_with(|| Arc::new(init()))
+            .clone()
+    }
+
     /// Run the pipeline for one turn. Returns the serialized provider request
     /// and associated metadata, or an abort if the session is in an
     /// unrecoverable error state.
@@ -847,6 +864,7 @@ impl PipelineSession {
 
         Self {
             pipeline: ContextPipeline::new(config),
+            static_sections: None,
             stats,
             latches,
             emergent,
@@ -1078,6 +1096,32 @@ mod tests {
         assert_eq!(actual.cache_eligible_tokens, 7_225);
     }
 
+    #[test]
+    fn static_sections_cache_is_scoped_to_pipeline_session() {
+        let mut first_session = PipelineSession::new(PipelineConfig::default());
+        let first = first_session.static_sections_or_init(|| {
+            let mut sections = StaticSections::test_default();
+            sections.core_rules.text = "first session".into();
+            sections
+        });
+        let reused = first_session.static_sections_or_init(|| {
+            panic!("a pipeline session must build static sections only once")
+        });
+
+        assert!(Arc::ptr_eq(&first, &reused));
+        assert_eq!(reused.core_rules.text, "first session");
+
+        let mut second_session = PipelineSession::new(PipelineConfig::default());
+        let second = second_session.static_sections_or_init(|| {
+            let mut sections = StaticSections::test_default();
+            sections.core_rules.text = "second session".into();
+            sections
+        });
+
+        assert!(!Arc::ptr_eq(&first, &second));
+        assert_eq!(second.core_rules.text, "second session");
+    }
+
     fn test_external() -> ExternalSources {
         ExternalSources {
             memory_entries: vec![],

crates/runtime/src/data_layer/storage.rs

@@ -3,12 +3,12 @@ pub use astra_services::storage::*;
 use std::time::Duration;
 
 use serde_json::Value;
-use sqlx::{MySql, query};
+use sqlx::{MySql, QueryBuilder, Row, query};
 
 use astra_core::canonical_names::{
     metadata_duration_ms, metadata_tool_call_id, metadata_tool_name,
 };
-use astra_core::matrixone_statement_with_null_shape;
+use astra_core::{matrixone_null_shape_comment, matrixone_statement_with_null_shape};
 use astra_turn_core::contracts::{
     TurnCoreEventRecord, TurnDecisionAuditRecord, TurnSkillSelectionRecord, TurnToolEventRecord,
 };
@@ -211,68 +211,94 @@ fn trace_event_insert_values(event: &TraceEvent) -> Result<TraceEventInsertValue
     })
 }
 
-pub(crate) async fn insert_trace_event(
+pub(crate) async fn insert_trace_events(
     tx: &mut sqlx::Transaction<'_, MySql>,
-    event: &TraceEvent,
-) -> Result<bool, sqlx::Error> {
-    let values = trace_event_insert_values(event)?;
-    let insert_sql = matrixone_statement_with_null_shape(
+    events: &[TraceEvent],
+) -> Result<(u64, Option<String>), sqlx::Error> {
+    let Some(first) = events.first() else {
+        return Ok((0, None));
+    };
+    if events
+        .iter()
+        .any(|event| event.user_id != first.user_id || event.session_id != first.session_id)
+    {
+        return Err(sqlx::Error::Protocol(
+            "trace event batch must belong to one user session".to_string(),
+        ));
+    }
+
+    let prepared = events
+        .iter()
+        .enumerate()
+        .map(|(index, event)| trace_event_insert_values(event).map(|values| (index, values)))
+        .collect::<Result<Vec<_>, _>>()?;
+    if prepared.is_empty() {
+        return Ok((0, None));
+    }
+
+    let mut insert = QueryBuilder::<MySql>::new(
         "INSERT IGNORE INTO agent_events \
          (event_id, session_id, user_id, agent_id, agent_version, event_type, content, \
           parent_event_id, causal_chain_id, run_id, parent_run_id, turn_id, turn_seq, \
           round_index, tool_call_id, parent_agent_id, trace_kind, token_usage, \
           llm_model_used, reasoning_content, token_input, token_output, token_total, \
-          meta_tool_name, meta_duration_ms, metadata, created_at) \
-         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
-        values.nullable_shape(event),
+          meta_tool_name, meta_duration_ms, metadata, created_at) ",
     );
-    let result = query(&insert_sql)
-        .bind(&event.event_id)
-        .bind(&event.session_id)
-        .bind(&event.user_id)
-        .bind(event.agent_id.as_deref().unwrap_or("astra-server"))
-        .bind(env!("CARGO_PKG_VERSION"))
-        .bind(&event.event_type)
-        .bind(&event.content)
-        .bind(&event.parent_event_id)
-        .bind(&event.causal_chain_id)
-        .bind(&event.run_id)
-        .bind(&event.parent_run_id)
-        .bind(&event.turn_id)
-        .bind(event.turn_seq)
-        .bind(event.round_index)
-        .bind(&event.tool_call_id)
-        .bind(&event.parent_agent_id)
-        .bind(&event.trace_kind)
-        .bind(values.token_usage_json)
-        .bind(&event.llm_model_used)
-        .bind(&event.reasoning_content)
-        .bind(values.token_input)
-        .bind(values.token_output)
-        .bind(values.token_total)
-        .bind(&event.meta_tool_name)
-        .bind(event.meta_duration_ms)
-        .bind(values.metadata_json)
-        .bind(values.created_at)
-        .execute(&mut **tx)
-        .await?;
-    let inserted = result.rows_affected() > 0;
-    if inserted {
-        insert_agent_event_edges(
-            &mut **tx,
-            &event.user_id,
-            &event.session_id,
-            &event.event_id,
-            event.parent_event_id.as_deref(),
-            event
-                .parent_event_id
-                .as_ref()
-                .map(|id| std::slice::from_ref(id))
-                .unwrap_or(&[]),
-        )
-        .await?;
-    }
-    Ok(inserted)
+    insert.push_values(prepared.iter(), |mut row, (index, values)| {
+        let event = &events[*index];
+        row.push_bind(&event.event_id)
+            .push_bind(&event.session_id)
+            .push_bind(&event.user_id)
+            .push_bind(event.agent_id.as_deref().unwrap_or("astra-server"))
+            .push_bind(env!("CARGO_PKG_VERSION"))
+            .push_bind(&event.event_type)
+            .push_bind(&event.content)
+            .push_bind(&event.parent_event_id)
+            .push_bind(&event.causal_chain_id)
+            .push_bind(&event.run_id)
+            .push_bind(&event.parent_run_id)
+            .push_bind(&event.turn_id)
+            .push_bind(event.turn_seq)
+            .push_bind(event.round_index)
+            .push_bind(&event.tool_call_id)
+            .push_bind(&event.parent_agent_id)
+            .push_bind(&event.trace_kind)
+            .push_bind(&values.token_usage_json)
+            .push_bind(&event.llm_model_used)
+            .push_bind(&event.reasoning_content)
+            .push_bind(values.token_input)
+            .push_bind(values.token_output)
+            .push_bind(values.token_total)
+            .push_bind(&event.meta_tool_name)
+            .push_bind(event.meta_duration_ms)
+            .push_bind(&values.metadata_json)
+            .push_bind(&values.created_at);
+    });
+    insert.push(matrixone_null_shape_comment(
+        prepared
+            .iter()
+            .flat_map(|(index, values)| values.nullable_shape(&events[*index])),
+    ));
+    let inserted = insert.build().execute(&mut **tx).await?.rows_affected();
+
+    // Both immutable event rows and their edge projection are idempotent
+    // inserts. Sending the complete deterministic batch lets MatrixOne report
+    // the number of newly inserted events and removes the preceding existence
+    // read. Replayed edges are ignored by their unique key, while a partial
+    // replay still repairs any missing edge rows.
+    let edge_inputs = events
+        .iter()
+        .map(|event| astra_services::storage::AgentEventEdgeInsert {
+            user_id: &event.user_id,
+            session_id: &event.session_id,
+            child_event_id: &event.event_id,
+            primary_parent_event_id: event.parent_event_id.as_deref(),
+            parent_event_ids: event.parent_event_id.as_slice(),
+        })
+        .collect::<Vec<_>>();
+    astra_services::storage::insert_agent_event_edges_batch(&mut **tx, &edge_inputs).await?;
+
+    Ok((inserted, events.last().map(|event| event.event_id.clone())))
 }
 
 pub(crate) async fn insert_core_turn_event(

crates/runtime/src/server/agent_binding_skill_runtime.rs

@@ -690,6 +690,66 @@ pub(crate) async fn prepare_agent_binding_skill_resolver(
     build_resolver_from_catalogs(server_id, endpoint_url, authorization, catalogs)
 }
 
+pub(crate) fn prepare_agent_binding_skill_resolver_from_snapshot(
+    server_id: &str,
+    endpoint_url: &str,
+    authorization: &str,
+    agent_binding_ids: &[String],
+    snapshot_catalogs: &[astra_services::runs::RuntimeCapabilitySkillCatalogSnapshotRequest],
+) -> Result<PreparedAgentBindingSkills, (StatusCode, Json<ErrorResponse>)> {
+    validate_skill_endpoint(endpoint_url)?;
+    let mut catalogs_by_binding = HashMap::with_capacity(snapshot_catalogs.len());
+    for catalog in snapshot_catalogs {
+        if catalog.agent_binding_id.trim().is_empty()
+            || catalog.agent_binding_id != catalog.agent_binding_id.trim()
+            || catalogs_by_binding
+                .insert(catalog.agent_binding_id.clone(), &catalog.skills)
+                .is_some()
+        {
+            return Err(skill_error(
+                StatusCode::BAD_REQUEST,
+                "runtime capability discovery snapshot has an invalid or duplicate agent_binding_id",
+                "agent_binding_discovery_snapshot_invalid",
+            ));
+        }
+    }
+    if catalogs_by_binding.len() != agent_binding_ids.len()
+        || catalogs_by_binding
+            .keys()
+            .any(|id| !agent_binding_ids.contains(id))
+    {
+        return Err(skill_error(
+            StatusCode::BAD_REQUEST,
+            "runtime capability discovery snapshot does not match the Agent Binding Set",
+            "agent_binding_discovery_snapshot_invalid",
+        ));
+    }
+    let mut catalogs = Vec::with_capacity(agent_binding_ids.len());
+    for agent_binding_id in agent_binding_ids {
+        let Some(raw_skills) = catalogs_by_binding.get(agent_binding_id) else {
+            return Err(skill_error(
+                StatusCode::BAD_REQUEST,
+                "runtime capability discovery snapshot is missing an Agent Binding skill catalog",
+                "agent_binding_discovery_snapshot_invalid",
+            ));
+        };
+        let mut skills = Vec::with_capacity(raw_skills.len());
+        for raw_skill in *raw_skills {
+            let skill =
+                serde_json::from_value::<DiscoveredSkill>(raw_skill.clone()).map_err(|error| {
+                    skill_error(
+                        StatusCode::BAD_REQUEST,
+                        format!("runtime capability discovery Skill is invalid: {error}"),
+                        "agent_binding_discovery_snapshot_invalid",
+                    )
+                })?;
+            skills.push(skill);
+        }
+        catalogs.push((Some(agent_binding_id.clone()), skills));
+    }
+    build_resolver_from_catalogs(server_id, endpoint_url, authorization, catalogs)
+}
+
 pub(crate) async fn prepare_runtime_skill_resolver(
     server_id: &str,
     endpoint_url: &str,
@@ -720,6 +780,31 @@ mod tests {
         }
     }
 
+    #[test]
+    fn prepares_agent_binding_skills_from_frozen_snapshot() {
+        let prepared = prepare_agent_binding_skill_resolver_from_snapshot(
+            "skills",
+            "https://catalog.example.test/api/v1/skills/http",
+            "Bearer runtime-grant",
+            &["binding_1".to_string()],
+            &[
+                astra_services::runs::RuntimeCapabilitySkillCatalogSnapshotRequest {
+                    agent_binding_id: "binding_1".to_string(),
+                    skills: vec![json!({
+                        "name": "pdf",
+                        "description": "Read PDF files",
+                        "allowed_tools": ["file_download"]
+                    })],
+                },
+            ],
+        )
+        .expect("frozen Skill discovery snapshot should prepare");
+
+        assert_eq!(prepared.catalogs.len(), 1);
+        assert_eq!(prepared.catalogs[0].agent_binding_id, "binding_1");
+        assert!(prepared.resolver.is_some());
+    }
+
     #[test]
     fn build_resolver_maps_discovered_skill_to_lazy_catalog_entry() {
         let resolver = build_resolver(