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/16 15:36:23 UTC
(maka) branch rust-preview updated: perf(runtime): keep long histories responsive during preparation
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 73dd759480 perf(runtime): keep long histories responsive during preparation
73dd759480 is described below
commit 73dd7594803a539d18472e9209765ae3dcc080f1
Author: Wang <m4...@gmail.com>
AuthorDate: Fri Oct 16 23:35:53 2026 +0800
perf(runtime): keep long histories responsive during preparation
Use indexed admission and transcript evidence lookups for both native model steps and external tools. Move pure context and pruning snapshots to independent readers, interrupt abandoned SQLite work, and drain transaction cleanup before connection reuse.
Add reproducible release probes for long histories, single invocations, external tools, and durable cancellation. Validate complete message output and work equivalence across alternating measurements.
Validation: 238 event-log tests, 61 engine tests, Clippy, and five-pair release comparisons passed.
Generated-by: Codex
---
.../migrations/0026_admission_covering_lookup.sql | 42 +---
.../migrations/0027_model_step_lookup.sql | 43 +---
.../migrations/0028_executor_tool_lookup.sql | 37 +---
platform/event-log/src/archive/candidates.rs | 2 +-
platform/event-log/src/connection.rs | 95 ++++++++-
platform/event-log/src/connection/cancellation.rs | 237 +++++++++++++++++++++
platform/event-log/src/context/read.rs | 2 +-
platform/event-log/src/transcript.rs | 35 +--
platform/event-log/src/transcript/evidence.rs | 88 ++++++--
platform/event-log/src/transcript/tools.rs | 17 +-
platform/event-log/tests/transcript_index.rs | 4 +
.../tests/transcript_index/performance.rs | 128 +++++++++++
.../tests/transcript_index/performance_executor.rs | 155 ++++++++++++++
plugins/agent/engine/src/compact.rs | 11 +-
plugins/agent/engine/src/model_attempt.rs | 37 ++--
plugins/agent/engine/src/prune.rs | 11 +-
plugins/agent/engine/src/steps.rs | 13 +-
.../engine/tests/integration/long_task_perf.rs | 35 ++-
.../tests/integration/long_task_responsiveness.rs | 205 ++++++++++++++++++
plugins/agent/engine/tests/integration/main.rs | 1 +
scripts/rust/README.md | 32 +++
scripts/rust/README.zh-CN.md | 24 +++
scripts/rust/benchmark-long-task.py | 30 ++-
23 files changed, 1089 insertions(+), 195 deletions(-)
diff --git a/plugins/agent/engine/tests/integration/main.rs b/platform/event-log/migrations/0026_admission_covering_lookup.sql
similarity index 52%
copy from plugins/agent/engine/tests/integration/main.rs
copy to platform/event-log/migrations/0026_admission_covering_lookup.sql
index 3d4e869598..f5ffc6972a 100644
--- a/plugins/agent/engine/tests/integration/main.rs
+++ b/platform/event-log/migrations/0026_admission_covering_lookup.sql
@@ -17,37 +17,13 @@
* under the License.
*/
-mod agent_loop;
-mod attachment_projection;
-mod auto_context;
-mod code_mode;
-mod code_mode_lifetime;
-mod code_mode_media;
-mod context_compaction;
-mod context_overflow;
-mod context_stop;
-mod dynamic_tools;
-mod executor;
-mod frozen_projection;
-mod handoff;
-mod handoff_compaction;
-mod image_projection;
-mod lifetime;
-mod long_task_perf;
-mod model_interruption;
-mod pruning;
-mod session_import;
-mod stale_pruning;
-mod startup_recovery;
-mod startup_recovery_limits;
+-- runtime_events excludes collected bodies. Put that predicate in the indexes
+-- so proving every old opening is sealed never loads its event_json payload.
+CREATE INDEX admission_openings ON event_log(
+ json_extract(event_json, '$.invocation.session_id'), invocation_id
+) WHERE kind = 'invocation_opened' AND event_json IS NOT NULL;
-mod support {
- pub(crate) mod agent_loop;
- pub(crate) mod code_mode;
- pub(crate) mod code_mode_lifetime;
- pub(crate) mod context;
- pub(crate) mod http;
- pub(crate) mod image_projection;
- pub(crate) mod invocation;
- pub(crate) mod recovery;
-}
+-- Uniqueness already follows from invocation_boundary. Retaining that planner
+-- fact makes SQLite choose this covering subset over the older boundary index.
+CREATE UNIQUE INDEX admission_terminals ON event_log(invocation_id, kind)
+WHERE kind = 'invocation_ended' AND event_json IS NOT NULL;
diff --git a/plugins/agent/engine/tests/integration/main.rs b/platform/event-log/migrations/0027_model_step_lookup.sql
similarity index 52%
copy from plugins/agent/engine/tests/integration/main.rs
copy to platform/event-log/migrations/0027_model_step_lookup.sql
index 3d4e869598..5a429fd905 100644
--- a/plugins/agent/engine/tests/integration/main.rs
+++ b/platform/event-log/migrations/0027_model_step_lookup.sql
@@ -17,37 +17,14 @@
* under the License.
*/
-mod agent_loop;
-mod attachment_projection;
-mod auto_context;
-mod code_mode;
-mod code_mode_lifetime;
-mod code_mode_media;
-mod context_compaction;
-mod context_overflow;
-mod context_stop;
-mod dynamic_tools;
-mod executor;
-mod frozen_projection;
-mod handoff;
-mod handoff_compaction;
-mod image_projection;
-mod lifetime;
-mod long_task_perf;
-mod model_interruption;
-mod pruning;
-mod session_import;
-mod stale_pruning;
-mod startup_recovery;
-mod startup_recovery_limits;
+-- Keep the existing catalog paths while allowing single-kind lookups to use
+-- the same partial index, including an absent executor in a long model run.
+DROP INDEX catalog_model_boundaries;
+CREATE INDEX catalog_model_boundaries ON event_log(invocation_id, kind, sequence)
+WHERE kind = 'model_requested' OR kind = 'model_completed'
+ OR kind = 'model_interrupted' OR kind = 'executor_started';
-mod support {
- pub(crate) mod agent_loop;
- pub(crate) mod code_mode;
- pub(crate) mod code_mode_lifetime;
- pub(crate) mod context;
- pub(crate) mod http;
- pub(crate) mod image_projection;
- pub(crate) mod invocation;
- pub(crate) mod recovery;
-}
+-- Resolve each request's terminal without scanning the containing invocation.
+-- The body predicate preserves runtime_events' exclusion of collected records.
+CREATE INDEX model_step_terminals ON event_log(invocation_id, operation_id, sequence)
+WHERE kind IN ('model_completed', 'model_interrupted') AND event_json IS NOT NULL;
diff --git a/plugins/agent/engine/tests/integration/main.rs b/platform/event-log/migrations/0028_executor_tool_lookup.sql
similarity index 52%
copy from plugins/agent/engine/tests/integration/main.rs
copy to platform/event-log/migrations/0028_executor_tool_lookup.sql
index 3d4e869598..aa46248731 100644
--- a/plugins/agent/engine/tests/integration/main.rs
+++ b/platform/event-log/migrations/0028_executor_tool_lookup.sql
@@ -17,37 +17,6 @@
* under the License.
*/
-mod agent_loop;
-mod attachment_projection;
-mod auto_context;
-mod code_mode;
-mod code_mode_lifetime;
-mod code_mode_media;
-mod context_compaction;
-mod context_overflow;
-mod context_stop;
-mod dynamic_tools;
-mod executor;
-mod frozen_projection;
-mod handoff;
-mod handoff_compaction;
-mod image_projection;
-mod lifetime;
-mod long_task_perf;
-mod model_interruption;
-mod pruning;
-mod session_import;
-mod stale_pruning;
-mod startup_recovery;
-mod startup_recovery_limits;
-
-mod support {
- pub(crate) mod agent_loop;
- pub(crate) mod code_mode;
- pub(crate) mod code_mode_lifetime;
- pub(crate) mod context;
- pub(crate) mod http;
- pub(crate) mod image_projection;
- pub(crate) mod invocation;
- pub(crate) mod recovery;
-}
+CREATE INDEX executor_tool_observations ON event_log (
+ invocation_id, json_extract(event_json, '$.fact.output.toolCallId'), sequence
+) WHERE kind IN ('executor_observed', 'executor_completed');
diff --git a/platform/event-log/src/archive/candidates.rs b/platform/event-log/src/archive/candidates.rs
index 05a1c21bad..d118a6f180 100644
--- a/platform/event-log/src/archive/candidates.rs
+++ b/platform/event-log/src/archive/candidates.rs
@@ -41,7 +41,7 @@ impl EventLog {
}
let (session, current) = (session.to_owned(), current.map(str::to_owned));
let cursor = cursor.cloned();
- self.connection.run(move |connection| Box::pin(async move {
+ self.connection.read_interruptible(move |connection| Box::pin(async move {
let mut tx = connection.begin().await?;
if let Some(id) = ¤t {
let live: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM runtime_events o WHERE invocation_id=? AND kind='invocation_opened'
diff --git a/platform/event-log/src/connection.rs b/platform/event-log/src/connection.rs
index ae6670ee75..4e33b26c15 100644
--- a/platform/event-log/src/connection.rs
+++ b/platform/event-log/src/connection.rs
@@ -17,9 +17,19 @@
* under the License.
*/
-use std::{future::Future, panic::AssertUnwindSafe, pin::Pin, sync::Arc};
+use std::{
+ future::Future,
+ panic::AssertUnwindSafe,
+ pin::Pin,
+ sync::{
+ Arc,
+ atomic::{AtomicBool, Ordering},
+ },
+};
mod authority;
+#[cfg(test)]
+mod cancellation;
pub use authority::ConnectionAuthority;
use futures_util::{FutureExt, future::Shared};
@@ -283,6 +293,27 @@ impl OwnedConnection {
operation: impl for<'c> FnOnce(&'c mut SqliteConnection) -> BoxFuture<'c, Result<T, StoreError>>
+ Send
+ 'static,
+ ) -> Result<T, StoreError> {
+ self.read_job(operation, false).await
+ }
+
+ /// Pure database reads may stop when their observer disappears. The owner
+ /// still awaits operation and transaction cleanup before reusing a reader.
+ pub(crate) async fn read_interruptible<T: Send + 'static>(
+ &self,
+ operation: impl for<'c> FnOnce(&'c mut SqliteConnection) -> BoxFuture<'c, Result<T, StoreError>>
+ + Send
+ + 'static,
+ ) -> Result<T, StoreError> {
+ self.read_job(operation, true).await
+ }
+
+ async fn read_job<T: Send + 'static>(
+ &self,
+ operation: impl for<'c> FnOnce(&'c mut SqliteConnection) -> BoxFuture<'c, Result<T, StoreError>>
+ + Send
+ + 'static,
+ interruptible: bool,
) -> Result<T, StoreError> {
let permit = self
.readers
@@ -290,23 +321,69 @@ impl OwnedConnection {
.acquire_owned()
.await
.map_err(|_| StoreError::ConnectionClosed)?;
- let (reply, result) = oneshot::channel();
+ let (mut reply, result) = oneshot::channel();
self.read_jobs
.send(Box::new(move |connection| {
Box::pin(async move {
let _permit = permit;
+ let mut clean = true;
let executed = match connection {
Ok(connection) => {
- AssertUnwindSafe(async { operation(connection).await })
- .catch_unwind()
- .await
+ let cancel = Arc::new(AtomicBool::new(false));
+ if interruptible {
+ match connection.lock_handle().await {
+ Ok(mut locked) => {
+ let cancel = cancel.clone();
+ // Remember cancellation between statements, then consume
+ // it once so transaction rollback can run normally.
+ locked.set_progress_handler(1000, move || {
+ !cancel.swap(false, Ordering::Relaxed)
+ });
+ }
+ Err(error) => {
+ let _ = reply.send(Err(error.into()));
+ return false;
+ }
+ }
+ }
+ let executed = {
+ let operation =
+ AssertUnwindSafe(async { operation(&mut *connection).await })
+ .catch_unwind();
+ tokio::pin!(operation);
+ tokio::select! {
+ result = &mut operation => result,
+ () = reply.closed(), if interruptible => {
+ cancel.store(true, Ordering::Relaxed);
+ operation.await
+ }
+ }
+ };
+ if interruptible {
+ // The handle lock follows queued Transaction::drop rollbacks.
+ clean = match connection.lock_handle().await {
+ Ok(mut locked) => {
+ locked.remove_progress_handler();
+ // SAFETY: the SQLx guard excludes worker FFI calls;
+ // this only reads the live handle's transaction state.
+ unsafe {
+ rusqlite::ffi::sqlite3_get_autocommit(
+ locked.as_raw_handle().as_ptr(),
+ ) != 0
+ }
+ }
+ Err(_) => false,
+ };
+ }
+ executed
}
Err(error) => Ok(Err(error)),
};
- let reusable = !matches!(
- &executed,
- Err(_) | Ok(Err(StoreError::Sqlx(sqlx::Error::WorkerCrashed)))
- );
+ let reusable = clean
+ && !matches!(
+ &executed,
+ Err(_) | Ok(Err(StoreError::Sqlx(sqlx::Error::WorkerCrashed)))
+ );
let result = executed.unwrap_or(Err(StoreError::OperationUnknown));
let _ = reply.send(result);
reusable
diff --git a/platform/event-log/src/connection/cancellation.rs b/platform/event-log/src/connection/cancellation.rs
new file mode 100644
index 0000000000..d269466cbb
--- /dev/null
+++ b/platform/event-log/src/connection/cancellation.rs
@@ -0,0 +1,237 @@
+/*
+ * 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.
+ */
+
+use super::*;
+use std::time::Duration;
+
+const LONG_QUERY: &str = "WITH RECURSIVE n(x) AS (VALUES(0) UNION ALL SELECT x+1 FROM n WHERE x<5000000) SELECT sum(x) FROM n";
+
+async fn owner(directory: &std::path::Path) -> Arc<OwnedConnection> {
+ let lease = Arc::new(crate::root::FileLease::acquire(&directory.join("writer.lock")).unwrap());
+ Arc::new(OwnedConnection::open(
+ SqliteConnectOptions::new().filename(directory.join("cancel.sqlite")).create_if_missing(true),
+ ConnectionAuthority::Writer { lease, root: None },
+ |connection| Box::pin(async move {
+ sqlx::raw_sql("PRAGMA journal_mode=WAL; CREATE TABLE facts(value INTEGER); INSERT INTO facts VALUES(1);").execute(connection).await?;
+ Ok(())
+ }),
+ ).await.unwrap())
+}
+
+// Exercise both cancellation between statements and inside a nested SQL UDF.
+// Both readers must be reclaimed, retain their connection settings, and leave
+// their old transactions before observing the writer's subsequent commit.
+#[tokio::test]
+async fn cancelled_reads_reclaim_both_slots_and_rollback_before_reuse() {
+ for nested in [false, true] {
+ let directory = tempfile::tempdir().unwrap();
+ let owner = owner(directory.path()).await;
+ let entered = Arc::new(Semaphore::new(0));
+ let release = Arc::new(Semaphore::new(0));
+ let mut callers = Vec::new();
+ let mut completions = Vec::new();
+ let mut udf_releases = Vec::new();
+ for _ in 0..2 {
+ let reader = owner.clone();
+ let entered = entered.clone();
+ let release = release.clone();
+ let (udf_release, udf_gate) = std::sync::mpsc::channel();
+ udf_releases.push(udf_release);
+ let (done, completed) = oneshot::channel();
+ completions.push(completed);
+ callers.push(tokio::spawn(async move {
+ reader
+ .read_interruptible(move |connection| {
+ Box::pin(async move {
+ sqlx::query("PRAGMA cache_size=-1234")
+ .execute(&mut *connection)
+ .await?;
+ if nested {
+ let mut locked = connection.lock_handle().await?;
+ // SAFETY: SQLx excludes all worker calls during registration;
+ // the borrowed wrapper does not own or close the handle.
+ let borrowed = unsafe {
+ rusqlite::Connection::from_handle(
+ locked.as_raw_handle().as_ptr(),
+ )
+ }?;
+ let entered = entered.clone();
+ borrowed.create_scalar_function(
+ "nested_read",
+ 0,
+ rusqlite::functions::FunctionFlags::SQLITE_UTF8,
+ move |context| {
+ entered.add_permits(1);
+ udf_gate.recv_timeout(Duration::from_secs(10)).unwrap();
+ // SAFETY: synchronous read-only use within this SQLite callback.
+ let connection = unsafe { context.get_connection() }?;
+ let result = connection.query_row(LONG_QUERY, [], |row| row.get::<_, i64>(0));
+ assert!(matches!(&result, Err(rusqlite::Error::SqliteFailure(error, _)) if error.extended_code == 9));
+ result
+ },
+ )?;
+ }
+ let mut tx = connection.begin().await?;
+ let count: i64 = sqlx::query_scalar("SELECT count(*) FROM facts")
+ .fetch_one(&mut *tx)
+ .await?;
+ assert_eq!(count, 1);
+ if !nested {
+ entered.add_permits(1);
+ release.acquire().await.unwrap().forget();
+ }
+ let result = sqlx::query_scalar::<_, i64>(if nested {
+ "SELECT nested_read()"
+ } else {
+ LONG_QUERY
+ })
+ .fetch_one(&mut *tx)
+ .await;
+ let error = result.as_ref().unwrap_err().as_database_error().unwrap();
+ assert_eq!(error.code().as_deref(), Some(if nested { "1" } else { "9" }));
+ assert_eq!(error.message(), "interrupted");
+ let _ = done.send(());
+ result?;
+ tx.commit().await?;
+ Ok(())
+ })
+ })
+ .await
+ }));
+ }
+ tokio::time::timeout(Duration::from_secs(5), entered.acquire_many(2))
+ .await
+ .unwrap()
+ .unwrap()
+ .forget();
+ for caller in callers {
+ caller.abort();
+ assert!(caller.await.unwrap_err().is_cancelled());
+ }
+ // Writer remains independent while both readers are paused in snapshots.
+ owner
+ .run(|connection| {
+ Box::pin(async move {
+ sqlx::query("INSERT INTO facts VALUES(2)")
+ .execute(connection)
+ .await?;
+ Ok(())
+ })
+ })
+ .await
+ .unwrap();
+ release.add_permits(2);
+ for gate in udf_releases {
+ let _ = gate.send(());
+ }
+ for done in completions {
+ tokio::time::timeout(Duration::from_secs(5), done)
+ .await
+ .unwrap()
+ .unwrap();
+ }
+ let occupied = Arc::new(Semaphore::new(0));
+ let release = Arc::new(Semaphore::new(0));
+ let mut readers = Vec::new();
+ for _ in 0..2 {
+ let owner = owner.clone();
+ let occupied = occupied.clone();
+ let release = release.clone();
+ readers.push(tokio::spawn(async move {
+ owner
+ .read_interruptible(move |connection| {
+ Box::pin(async move {
+ let marker: i64 = sqlx::query_scalar("PRAGMA cache_size")
+ .fetch_one(&mut *connection)
+ .await?;
+ let mut tx = connection.begin().await?;
+ let count: i64 = sqlx::query_scalar("SELECT count(*) FROM facts")
+ .fetch_one(&mut *tx)
+ .await?;
+ occupied.add_permits(1);
+ release.acquire().await.unwrap().forget();
+ tx.commit().await?;
+ Ok((marker, count))
+ })
+ })
+ .await
+ }));
+ }
+ tokio::time::timeout(Duration::from_secs(5), occupied.acquire_many(2))
+ .await
+ .unwrap()
+ .unwrap()
+ .forget();
+ release.add_permits(2);
+ for reader in readers {
+ assert_eq!(reader.await.unwrap().unwrap(), (-1234, 2));
+ }
+ owner.shutdown().await.unwrap();
+ }
+}
+
+#[tokio::test]
+async fn cancelled_read_still_drains_before_shutdown_releases_authority() {
+ let directory = tempfile::tempdir().unwrap();
+ let owner = owner(directory.path()).await;
+ let entered = Arc::new(Semaphore::new(0));
+ let release = Arc::new(Semaphore::new(0));
+ let reader = owner.clone();
+ let entered_job = entered.clone();
+ let release_job = release.clone();
+ let caller = tokio::spawn(async move {
+ reader
+ .read_interruptible(move |connection| {
+ Box::pin(async move {
+ let mut tx = connection.begin().await?;
+ sqlx::query("SELECT * FROM facts").execute(&mut *tx).await?;
+ entered_job.add_permits(1);
+ release_job.acquire().await.unwrap().forget();
+ let result = sqlx::query_scalar::<_, i64>(LONG_QUERY)
+ .fetch_one(&mut *tx)
+ .await;
+ assert!(matches!(&result, Err(sqlx::Error::Database(error)) if error.code().as_deref() == Some("9")));
+ Ok(())
+ })
+ })
+ .await
+ });
+ tokio::time::timeout(Duration::from_secs(5), entered.acquire())
+ .await
+ .unwrap()
+ .unwrap()
+ .forget();
+ caller.abort();
+ assert!(caller.await.unwrap_err().is_cancelled());
+ let closing = owner.clone();
+ let shutdown = tokio::spawn(async move { closing.shutdown().await });
+ assert!(
+ tokio::time::timeout(Duration::from_millis(50), owner.closed.clone())
+ .await
+ .is_err()
+ );
+ assert!(crate::root::FileLease::acquire(&directory.path().join("writer.lock")).is_err());
+ release.add_permits(1);
+ tokio::time::timeout(Duration::from_secs(5), shutdown)
+ .await
+ .unwrap()
+ .unwrap()
+ .unwrap();
+ assert!(crate::root::FileLease::acquire(&directory.path().join("writer.lock")).is_ok());
+}
diff --git a/platform/event-log/src/context/read.rs b/platform/event-log/src/context/read.rs
index b88aab6a31..ba05d18740 100644
--- a/platform/event-log/src/context/read.rs
+++ b/platform/event-log/src/context/read.rs
@@ -87,7 +87,7 @@ impl EventLog {
crate::sessions::validate_id(session)?;
let (session, current) = (session.to_owned(), current.map(str::to_owned));
self.connection
- .run(move |connection| {
+ .read_interruptible(move |connection| {
Box::pin(async move {
let mut tx = connection.begin().await?;
let opening =
diff --git a/platform/event-log/src/transcript.rs b/platform/event-log/src/transcript.rs
index 09d5aae3a0..ce279a1d0f 100644
--- a/platform/event-log/src/transcript.rs
+++ b/platform/event-log/src/transcript.rs
@@ -81,28 +81,37 @@ impl EventLog {
// Read only small boundary headers here. Canonical payloads are selected
// and budget-checked separately, one boundary at a time.
let boundaries = {
- let records = sqlx::query(
- "SELECT sequence, invocation_id, operation_id, kind FROM session_history_events runtime_events
- WHERE owner_session_id = ?1
- AND sequence > ?2 AND sequence <= ?3
- AND (kind = 'context_checkpoint_recorded' OR NOT EXISTS (SELECT 1 FROM runtime_events opening
- WHERE opening.invocation_id = runtime_events.invocation_id
+ // The shared history view requires a whole-remainder sort here.
+ // Merge the two ordered ownership paths before applying the limit.
+ let eligible = "(e.kind = 'context_checkpoint_recorded' OR NOT EXISTS (SELECT 1 FROM runtime_events opening
+ WHERE opening.invocation_id = e.invocation_id
AND opening.kind = 'invocation_opened'
AND json_extract(opening.event_json, '$.fact.input.kind') = 'context_compact'))
- AND (kind NOT IN ('model_completed', 'model_interrupted') OR
+ AND (e.kind NOT IN ('model_completed', 'model_interrupted') OR
EXISTS (SELECT 1 FROM runtime_events request
JOIN runtime_events opening ON opening.invocation_id = request.invocation_id
AND opening.kind = 'invocation_opened'
- WHERE request.kind = 'model_requested' AND request.invocation_id = runtime_events.invocation_id
- AND request.operation_id = runtime_events.operation_id
+ WHERE request.kind = 'model_requested' AND request.invocation_id = e.invocation_id
+ AND request.operation_id = e.operation_id
AND json_extract(opening.event_json, '$.fact.input.kind') IN ('message', 'continuation', 'handoff')
AND json_extract(request.event_json, '$.fact.purpose') = 'main'))
- AND (kind IN ('message_imported', 'invocation_opened', 'message_steered', 'model_completed',
+ AND (e.kind IN ('message_imported', 'invocation_opened', 'message_steered', 'model_completed',
'model_interrupted', 'invocation_ended', 'context_checkpoint_recorded',
'tool_dispatched', 'tool_rejected', 'tool_settled', 'executor_completed')
- OR (kind = 'executor_observed' AND json_extract(event_json, '$.fact.output.type') IN ('tool_start','tool_result')))
- ORDER BY sequence LIMIT ?4",
- )
+ OR (e.kind = 'executor_observed' AND json_extract(e.event_json, '$.fact.output.type') IN ('tool_start','tool_result')))";
+ let sql = format!(
+ "SELECT e.sequence AS sequence, e.invocation_id, e.operation_id, e.kind
+ FROM runtime_events e
+ WHERE json_extract(e.event_json, '$.invocation.session_id') = ?1
+ AND e.sequence > ?2 AND e.sequence <= ?3 AND {eligible}
+ UNION ALL
+ SELECT h.sequence AS sequence, e.invocation_id, e.operation_id, e.kind
+ FROM session_history_members h CROSS JOIN runtime_events e ON e.sequence = h.sequence
+ WHERE h.session_id = ?1 AND h.sequence > ?2 AND h.sequence <= ?3 AND {eligible}
+ ORDER BY sequence LIMIT ?4"
+ );
+ // Only the static predicate is interpolated; all inputs remain bound.
+ let records = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(&session)
.bind(progress)
.bind(fence)
diff --git a/platform/event-log/src/transcript/evidence.rs b/platform/event-log/src/transcript/evidence.rs
index 5adaf63b66..e7bcc8bb1c 100644
--- a/platform/event-log/src/transcript/evidence.rs
+++ b/platform/event-log/src/transcript/evidence.rs
@@ -48,6 +48,7 @@ pub(super) async fn selected(
through: i64,
boundary: bool,
) -> Result<Vec<StoredEvent>, StoreError> {
+ let mut executor_boundary = false;
if boundary {
let kind: String = sqlx::query_scalar(
"SELECT kind FROM runtime_events WHERE sequence = ? AND invocation_id = ?",
@@ -56,6 +57,7 @@ pub(super) async fn selected(
.bind(invocation)
.fetch_one(&mut *tx)
.await?;
+ executor_boundary = kind == "executor_observed";
if matches!(
kind.as_str(),
"tool_dispatched" | "tool_rejected" | "tool_settled"
@@ -67,7 +69,25 @@ pub(super) async fn selected(
"SELECT EXISTS(SELECT 1 FROM runtime_events WHERE invocation_id = ?1 AND sequence <= ?2 AND kind = 'executor_started')"
).bind(invocation).bind(through).fetch_one(&mut *tx).await?;
if external {
- return read(tx, sqlx::query(
+ let sql = if executor_boundary {
+ // A tool boundary needs only its own activity. Preserve SQLite's
+ // native JSON equality, including malformed non-string IDs.
+ "SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+ WHERE invocation_id = ?1 AND sequence <= ?2 AND kind = 'invocation_opened'
+ UNION
+ SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+ WHERE invocation_id = ?1 AND sequence <= ?2 AND kind = 'executor_started'
+ UNION
+ SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+ WHERE invocation_id = ?1 AND sequence <= ?2 AND ?3 AND sequence = ?2
+ UNION
+ SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+ WHERE invocation_id = ?1 AND sequence <= ?2
+ AND kind IN ('executor_observed','executor_completed')
+ AND json_extract(event_json, '$.fact.output.toolCallId') =
+ (SELECT json_extract(event_json, '$.fact.output.toolCallId') FROM runtime_events WHERE sequence = ?2)
+ ORDER BY sequence LIMIT ?4"
+ } else {
"SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
WHERE invocation_id = ?1 AND sequence <= ?2 AND (
kind IN ('invocation_opened','executor_started')
@@ -77,7 +97,16 @@ pub(super) async fn selected(
OR json_extract(event_json, '$.fact.output.toolCallId') =
(SELECT json_extract(event_json, '$.fact.output.toolCallId') FROM runtime_events WHERE sequence = ?2))))
ORDER BY sequence LIMIT ?4"
- ).bind(invocation).bind(through).bind(boundary).bind((MAX_EVENTS + 1) as i64)).await;
+ };
+ return read(
+ tx,
+ sqlx::query(sql)
+ .bind(invocation)
+ .bind(through)
+ .bind(boundary)
+ .bind((MAX_EVENTS + 1) as i64),
+ )
+ .await;
}
// A failed/cancelled terminal can seal a request without a separate model
// interruption (for example after a model-observation commit failure).
@@ -87,24 +116,55 @@ pub(super) async fn selected(
None
};
let step = step.or(unresolved.as_deref());
- // Bound count before retrieving payloads, then bound cumulative UTF-8 bytes
- // from SQLite lengths before any event JSON is deserialized.
+ let ids = serde_json::to_string(if boundary {
+ std::slice::from_ref(&through)
+ } else {
+ &[]
+ })?;
+ read_step(tx, invocation, step, through, &ids, false).await
+}
+
+/// Merge only the opening, selected model step, and explicit boundary/ancestry.
+/// UNION removes overlapping identities before the bounded reader counts them.
+pub(super) async fn read_step(
+ tx: &mut SqliteConnection,
+ invocation: &str,
+ step: Option<&str>,
+ through: i64,
+ boundary_ids: &str,
+ include_completed: bool,
+) -> Result<Vec<StoredEvent>, StoreError> {
+ // CAST matches the observation index; the original equality also preserves
+ // selection semantics for malformed non-string step IDs.
let query = sqlx::query(
"SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
- WHERE invocation_id = ?1 AND sequence <= ?3 AND (
- kind = 'invocation_opened'
- OR (?4 AND sequence = ?3)
- OR (?2 IS NOT NULL AND (
- (kind = 'model_requested' AND operation_id = ?2)
- OR (kind = 'model_observed' AND json_extract(event_json, '$.fact.step_id') = ?2)
- )))
- ORDER BY sequence LIMIT ?5",
+WHERE invocation_id = ?1 AND sequence <= ?3 AND kind = 'invocation_opened'
+UNION
+SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+WHERE invocation_id = ?1 AND sequence <= ?3
+ AND sequence IN (SELECT value FROM json_each(?4))
+UNION
+SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+WHERE invocation_id = ?1 AND sequence <= ?3
+ AND ?2 IS NOT NULL AND kind = 'model_requested' AND operation_id = ?2
+UNION
+SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+WHERE invocation_id = ?1 AND sequence <= ?3
+ AND ?6 AND ?2 IS NOT NULL AND kind = 'model_completed' AND operation_id = ?2
+UNION
+SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+WHERE invocation_id = ?1 AND sequence <= ?3 AND ?2 IS NOT NULL
+ AND kind = 'model_observed'
+ AND CAST(json_extract(event_json, '$.fact.step_id') AS TEXT) = ?2
+ AND json_extract(event_json, '$.fact.step_id') = ?2
+ORDER BY sequence LIMIT ?5",
)
.bind(invocation)
.bind(step)
.bind(through)
- .bind(boundary)
- .bind((MAX_EVENTS + 1) as i64);
+ .bind(boundary_ids)
+ .bind((MAX_EVENTS + 1) as i64)
+ .bind(include_completed);
read(tx, query).await
}
diff --git a/platform/event-log/src/transcript/tools.rs b/platform/event-log/src/transcript/tools.rs
index 1336ccbe8c..e1892ea465 100644
--- a/platform/event-log/src/transcript/tools.rs
+++ b/platform/event-log/src/transcript/tools.rs
@@ -116,20 +116,5 @@ pub(super) async fn selected(
}
}
let ids = serde_json::to_string(&ids)?;
- let query = sqlx::query(
- "SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
- WHERE invocation_id = ?1 AND sequence <= ?3 AND (
- kind = 'invocation_opened' OR sequence IN (SELECT value FROM json_each(?4))
- OR (?2 IS NOT NULL AND (
- (kind IN ('model_requested', 'model_completed') AND operation_id = ?2)
- OR (kind = 'model_observed' AND json_extract(event_json, '$.fact.step_id') = ?2)
- )))
- ORDER BY sequence LIMIT ?5",
- )
- .bind(invocation)
- .bind(step)
- .bind(through)
- .bind(ids)
- .bind((evidence::MAX_EVENTS + 1) as i64);
- evidence::read(tx, query).await
+ evidence::read_step(tx, invocation, step.as_deref(), through, &ids, true).await
}
diff --git a/platform/event-log/tests/transcript_index.rs b/platform/event-log/tests/transcript_index.rs
index 5326502a53..424df7aecb 100644
--- a/platform/event-log/tests/transcript_index.rs
+++ b/platform/event-log/tests/transcript_index.rs
@@ -40,6 +40,10 @@ mod fixtures;
mod history;
#[path = "transcript_index/image.rs"]
mod image;
+#[path = "transcript_index/performance.rs"]
+mod performance;
+#[path = "transcript_index/performance_executor.rs"]
+mod performance_executor;
#[path = "transcript_index/source.rs"]
mod source;
use fixtures::*;
diff --git a/platform/event-log/tests/transcript_index/performance.rs b/platform/event-log/tests/transcript_index/performance.rs
new file mode 100644
index 0000000000..0af9b47074
--- /dev/null
+++ b/platform/event-log/tests/transcript_index/performance.rs
@@ -0,0 +1,128 @@
+/*
+ * 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.
+ */
+
+//! Opt-in release probe for many model steps inside one physical invocation.
+use super::*;
+use maka_agent_facts::model::{ModelFinishReason, assembly::StepBuilder};
+use serde_json::json;
+use std::time::Instant;
+
+#[tokio::test]
+#[ignore = "release performance probe for one long invocation"]
+#[allow(clippy::assertions_on_constants)]
+async fn one_invocation_transcript() {
+ assert!(!cfg!(debug_assertions), "run with --release");
+ let sizes = std::env::var("MAKA_TRANSCRIPT_BENCH_STEPS").unwrap_or("128,512".into());
+ let repetitions: usize = std::env::var("MAKA_TRANSCRIPT_BENCH_REPETITIONS")
+ .unwrap_or("5".into())
+ .parse()
+ .unwrap();
+ assert!(repetitions > 0);
+ for steps in sizes.split(',').map(|size| size.parse::<usize>().unwrap()) {
+ assert!(steps > 0);
+ let fixture = tempfile::tempdir().unwrap();
+ let database = fixture.path().join("source.sqlite");
+ let log = EventLog::open(&database).await.unwrap();
+ log.append(&event("a", opening())).await.unwrap();
+ for step in 0..steps {
+ let name = format!("s-{step}");
+ let request = request(&name);
+ let Fact::ModelRequested { step_id, .. } = &request else {
+ unreachable!()
+ };
+ let step_id = step_id.clone();
+ let mut writes = vec![event("a", request)];
+ let text = format!("one-invocation-step-{step}. ");
+ let mut observations = vec![ModelEvent::PartStarted {
+ id: "text".into(),
+ text_kind: TextKind::Text,
+ provider_options: None,
+ }];
+ observations.extend((0..13).map(|_| ModelEvent::PartDelta {
+ id: "text".into(),
+ text: text.clone(),
+ provider_options: None,
+ }));
+ observations.push(ModelEvent::PartFinished {
+ id: "text".into(),
+ provider_options: None,
+ });
+ observations.push(ModelEvent::Finished {
+ reason: ModelFinishReason::Stop,
+ usage: ModelUsage::default(),
+ provider_options: None,
+ });
+ let mut builder = StepBuilder::for_step(&step_id).unwrap();
+ for observation in observations {
+ builder.push(observation.clone()).unwrap();
+ writes.push(event("a", observe(&name, observation)));
+ }
+ writes.push(event(
+ "a",
+ Fact::ModelCompleted {
+ step_id,
+ output: builder.finish().unwrap(),
+ },
+ ));
+ assert_eq!(writes.len(), 18);
+ log.append_batch(&writes).await.unwrap();
+ }
+ let through = end(&log, "a", InvocationOutcome::Completed).await;
+ assert_eq!(through, (steps * 18 + 2) as u64);
+ log.close().await.unwrap();
+ for repetition in 0..repetitions {
+ let sample = tempfile::tempdir().unwrap();
+ let path = sample.path().join("sample.sqlite");
+ std::fs::copy(&database, &path).unwrap();
+ let log = EventLog::open(&path).await.unwrap();
+ let started = Instant::now();
+ let mut batches = 1usize;
+ while !log.prepare_transcript("a", through, 32).await.unwrap() {
+ batches += 1;
+ }
+ let transcript_prepare_ms = started.elapsed().as_secs_f64() * 1000.0;
+ let db = Connection::open(&path).unwrap();
+ let texts: Vec<String> = db
+ .prepare(
+ "SELECT json_extract(payload, '$.text') FROM transcript_rows
+ WHERE session_id='a' AND json_extract(payload, '$.type')='assistant'
+ ORDER BY sequence",
+ )
+ .unwrap()
+ .query_map([], |row| row.get(0))
+ .unwrap()
+ .collect::<Result<_, _>>()
+ .unwrap();
+ assert_eq!(texts.len(), steps);
+ for (step, text) in texts.iter().enumerate() {
+ assert_eq!(text, &format!("one-invocation-step-{step}. ").repeat(13));
+ }
+ assert_eq!(progress(&db, "a"), through as i64);
+ println!(
+ "{}",
+ json!({"benchmark":"one_invocation_transcript", "shape":"one_invocation",
+ "model_steps":steps, "history_events":through, "repetition":repetition,
+ "transcript_prepare_ms":transcript_prepare_ms, "preparation_calls":batches,
+ "assistant_messages":texts.len(), "errors":0})
+ );
+ drop(db);
+ log.close().await.unwrap();
+ }
+ }
+}
diff --git a/platform/event-log/tests/transcript_index/performance_executor.rs b/platform/event-log/tests/transcript_index/performance_executor.rs
new file mode 100644
index 0000000000..c69c30bdc3
--- /dev/null
+++ b/platform/event-log/tests/transcript_index/performance_executor.rs
@@ -0,0 +1,155 @@
+/*
+ * 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.
+ */
+
+//! Opt-in release probe for sequential tools inside one external invocation.
+use super::*;
+use maka_agent_facts::execution::{
+ ApprovalPolicy, BehaviorId, CollaborationMode, InvocationConfiguration, SandboxMode, ToolMode,
+ WorkspaceOrigin,
+};
+use maka_agent_facts::executor::{Binding, Output};
+use serde_json::json;
+use std::time::Instant;
+
+#[tokio::test]
+#[ignore = "release performance probe for a long external invocation"]
+#[allow(clippy::assertions_on_constants)]
+async fn executor_invocation_transcript() {
+ assert!(!cfg!(debug_assertions), "run with --release");
+ let sizes = std::env::var("MAKA_EXECUTOR_BENCH_STEPS").unwrap_or("128,512".into());
+ let repetitions: usize = std::env::var("MAKA_EXECUTOR_BENCH_REPETITIONS")
+ .unwrap_or("5".into())
+ .parse()
+ .unwrap();
+ assert!(repetitions > 0);
+ for count in sizes.split(',').map(|size| size.parse::<usize>().unwrap()) {
+ assert!((1..=2048).contains(&count));
+ let fixture = tempfile::tempdir().unwrap();
+ let database = fixture.path().join("source.sqlite");
+ let log = EventLog::open(&database).await.unwrap();
+ let mut opening = opening();
+ let Fact::InvocationOpened { configuration, .. } = &mut opening else {
+ unreachable!()
+ };
+ *configuration = Some(Box::new(InvocationConfiguration {
+ compaction_prompt: None,
+ workspace_origin: WorkspaceOrigin::Selected,
+ approval_policy: ApprovalPolicy::OnRequest,
+ boundary_revision: 0,
+ cwd: fixture.path().to_string_lossy().into_owned(),
+ workspace_identity: None,
+ model: None,
+ editing_tools: None,
+ tool_discovery: false,
+ tool_composition: None,
+ system_prompt: None,
+ thinking_level: None,
+ sandbox_mode: SandboxMode::WorkspaceWrite,
+ collaboration_mode: CollaborationMode::Agent,
+ orchestration_mode: BehaviorId::default(),
+ tool_mode: ToolMode::Direct,
+ }));
+ log.append(&event("a", opening)).await.unwrap();
+ log.append(&event(
+ "a",
+ Fact::ExecutorStarted {
+ settings: Default::default(),
+ binding: Binding {
+ executor_id: "example".to_owned().try_into().unwrap(),
+ package_id: "example".into(),
+ entry_id: "example".into(),
+ activation: "activation".into(),
+ },
+ },
+ ))
+ .await
+ .unwrap();
+ for tool in 0..count {
+ let id = format!("external-{tool}");
+ let outputs = [
+ Output::ToolStart {
+ tool_call_id: id.clone(),
+ name: "Read".into(),
+ input: json!({"path":format!("file-{tool}")}),
+ },
+ Output::ToolProgress {
+ tool_call_id: id.clone(),
+ text: format!("progress-{tool}"),
+ },
+ Output::ToolResult {
+ tool_call_id: id,
+ text: format!("result-{tool}"),
+ is_error: false,
+ },
+ ];
+ let writes = outputs
+ .into_iter()
+ .map(|output| event("a", Fact::ExecutorObserved { output }))
+ .collect::<Vec<_>>();
+ log.append_batch(&writes).await.unwrap();
+ }
+ log.append(&event(
+ "a",
+ Fact::ExecutorCompleted {
+ text: "external complete".into(),
+ },
+ ))
+ .await
+ .unwrap();
+ let through = end(&log, "a", InvocationOutcome::Completed).await;
+ assert_eq!(through, (count * 3 + 4) as u64);
+ let prefix = log.prefix(10_000, 16 * 1024 * 1024).await.unwrap();
+ let mut view = maka_presentation::InvocationView::new(16 * 1024 * 1024).unwrap();
+ let expected = prefix
+ .events
+ .iter()
+ .flat_map(|row| view.push(row).unwrap())
+ .map(|row| serde_json::to_value(row.message).unwrap())
+ .collect::<Vec<_>>();
+ assert_eq!(expected.len(), count * 2 + 4);
+ log.close().await.unwrap();
+ for repetition in 0..repetitions {
+ let sample = tempfile::tempdir().unwrap();
+ let path = sample.path().join("sample.sqlite");
+ std::fs::copy(&database, &path).unwrap();
+ let log = EventLog::open(&path).await.unwrap();
+ let started = Instant::now();
+ let mut batches = 1usize;
+ while !log.prepare_transcript("a", through, 32).await.unwrap() {
+ batches += 1;
+ }
+ let transcript_prepare_ms = started.elapsed().as_secs_f64() * 1000.0;
+ let db = Connection::open(&path).unwrap();
+ let rows: Vec<Value> = db.prepare("SELECT CAST(payload AS TEXT) FROM transcript_rows WHERE session_id='a' ORDER BY sequence").unwrap()
+ .query_map([], |row| row.get::<_, String>(0)).unwrap()
+ .map(|row| serde_json::from_str(&row.unwrap()).unwrap()).collect();
+ assert_eq!(rows, expected);
+ assert_eq!(progress(&db, "a"), through as i64);
+ println!(
+ "{}",
+ json!({"benchmark":"executor_invocation_transcript", "shape":"external_invocation",
+ "tool_calls":count, "history_events":through, "repetition":repetition,
+ "transcript_prepare_ms":transcript_prepare_ms, "prepare_calls":batches,
+ "messages":rows.len(), "errors":0})
+ );
+ drop(db);
+ log.close().await.unwrap();
+ }
+ }
+}
diff --git a/plugins/agent/engine/src/compact.rs b/plugins/agent/engine/src/compact.rs
index c5d0b1e120..b47eacc9b6 100644
--- a/plugins/agent/engine/src/compact.rs
+++ b/plugins/agent/engine/src/compact.rs
@@ -65,14 +65,15 @@ pub(super) async fn capture(
continuation_base: Option<u64>,
) -> Result<Option<Job>, RunError> {
input.refresh_model(cancellation).await?;
- let source = inner
- .log
- .prepare_context_compaction(
+ let source = tokio::select! {
+ biased;
+ _ = cancellation.cancelled() => return Err(RunError::Cancelled),
+ source = inner.log.prepare_context_compaction(
&input.invocation.session_id,
Some(&input.invocation.invocation_id),
mode,
- )
- .await?;
+ ) => source?,
+ };
let has_new_history = source
.tail
.iter()
diff --git a/plugins/agent/engine/src/model_attempt.rs b/plugins/agent/engine/src/model_attempt.rs
index 106b5107f2..3fbb6e735d 100644
--- a/plugins/agent/engine/src/model_attempt.rs
+++ b/plugins/agent/engine/src/model_attempt.rs
@@ -203,23 +203,26 @@ pub(super) async fn read_source(
inner: &Inner,
input: &RunContext,
prior_unknown: bool,
+ cancellation: &CancellationToken,
) -> Result<ModelContextSource, RunError> {
- if prior_unknown {
- Ok(inner
- .log
- .read_manual_message_context(
- &input.invocation.session_id,
- &input.invocation.invocation_id,
- )
- .await?)
- } else {
- Ok(inner
- .log
- .read_model_context(
- &input.invocation.session_id,
- Some(&input.invocation.invocation_id),
- )
- .await?)
+ // Only abandon the read reply. Its snapshot remains owned by the log's
+ // independent reader until completion; no append or external effect is raced.
+ tokio::select! {
+ biased;
+ _ = cancellation.cancelled() => Err(RunError::Cancelled),
+ source = async {
+ if prior_unknown {
+ inner.log.read_manual_message_context(
+ &input.invocation.session_id,
+ &input.invocation.invocation_id,
+ ).await
+ } else {
+ inner.log.read_model_context(
+ &input.invocation.session_id,
+ Some(&input.invocation.invocation_id),
+ ).await
+ }
+ } => Ok(source?),
}
}
@@ -306,7 +309,7 @@ pub(super) async fn execute_main(
// Accepted progress replaces the frozen retry prompt. Release it
// before materializing the next canonical source.
drop(retry.take());
- let context = read_source(inner, input, prior_unknown).await?;
+ let context = read_source(inner, input, prior_unknown, cancellation).await?;
high_water = context.source_evidence.high_water;
source = RequestSource::from(&context);
let prompt = surface.apply(
diff --git a/plugins/agent/engine/src/prune.rs b/plugins/agent/engine/src/prune.rs
index 40fc5e2c28..87777b5636 100644
--- a/plugins/agent/engine/src/prune.rs
+++ b/plugins/agent/engine/src/prune.rs
@@ -34,16 +34,17 @@ pub(super) async fn run(
if cancellation.is_cancelled() {
return Err(RunError::Cancelled);
}
- let batch = inner
- .log
- .prepare_prune_candidates(
+ let batch = tokio::select! {
+ biased;
+ _ = cancellation.cancelled() => return Err(RunError::Cancelled),
+ batch = inner.log.prepare_prune_candidates(
&input.invocation.session_id,
Some(&input.invocation.invocation_id),
128,
4 * 1024 * 1024,
cursor.as_ref(),
- )
- .await?;
+ ) => batch?,
+ };
for candidate in batch.candidates {
let Some(placeholder) = ArchivedPlaceholder::prepare(
candidate.event_id,
diff --git a/plugins/agent/engine/src/steps.rs b/plugins/agent/engine/src/steps.rs
index af30276f09..3251d16bee 100644
--- a/plugins/agent/engine/src/steps.rs
+++ b/plugins/agent/engine/src/steps.rs
@@ -80,7 +80,9 @@ pub(super) async fn run(
if !tools.code_idle() {
return None;
}
- let context = model_attempt::read_source(inner, input, false).await.ok()?;
+ let context = model_attempt::read_source(inner, input, false, cancellation)
+ .await
+ .ok()?;
// Read the live source, but project from the successor's point of
// view: manual replay cuts distinguish current from inherited work.
let prompt = model_attempt::prompt(
@@ -144,7 +146,8 @@ pub(super) async fn run(
{
compaction = CompactionBudget::Reshaped;
}
- let mut context = model_attempt::read_source(inner, input, prior_unknown).await?;
+ let mut context =
+ model_attempt::read_source(inner, input, prior_unknown, cancellation).await?;
let mut prompt = model_attempt::prompt(
inner,
input,
@@ -184,7 +187,8 @@ pub(super) async fn run(
// Compaction makes its own logical requests. The next Main
// must observe any provider change made meanwhile.
input.refresh_model(cancellation).await?;
- context = model_attempt::read_source(inner, input, prior_unknown).await?;
+ context =
+ model_attempt::read_source(inner, input, prior_unknown, cancellation).await?;
prompt = model_attempt::prompt(
inner,
input,
@@ -311,7 +315,8 @@ pub(super) async fn run(
return Ok(maka_agent_facts::event::InvocationOutcome::Completed);
}
if !cancellation.is_cancelled() && lane.needs_confirmation() {
- let context = model_attempt::read_source(inner, input, prior_unknown).await?;
+ let context =
+ model_attempt::read_source(inner, input, prior_unknown, cancellation).await?;
let replay = model_attempt::prompt(
inner,
input,
diff --git a/plugins/agent/engine/tests/integration/long_task_perf.rs b/plugins/agent/engine/tests/integration/long_task_perf.rs
index 3c70710705..1f464bdb6e 100644
--- a/plugins/agent/engine/tests/integration/long_task_perf.rs
+++ b/plugins/agent/engine/tests/integration/long_task_perf.rs
@@ -110,7 +110,7 @@ async fn seed(
] {
writes.push(EventWrite::plain(RuntimeEvent::new(invocation.clone(), fact)).unwrap());
}
- let output = if tool_count > 0 {
+ let output: maka_agent_facts::model::ModelStep = if tool_count > 0 {
serde_json::from_value(
json!({"parts": calls, "finish_reason":"tool-calls", "usage":{}}),
)
@@ -153,6 +153,37 @@ async fn seed(
}
builder.finish().unwrap()
};
+ if tool_count > 0 {
+ for part in &output.parts {
+ let maka_agent_facts::model::ModelPart::ToolCall { call } = part else {
+ unreachable!()
+ };
+ writes.push(
+ EventWrite::plain(RuntimeEvent::new(
+ invocation.clone(),
+ Fact::ModelObserved {
+ step_id: step.clone(),
+ event: ModelEvent::ToolCall(call.clone()),
+ },
+ ))
+ .unwrap(),
+ );
+ }
+ writes.push(
+ EventWrite::plain(RuntimeEvent::new(
+ invocation.clone(),
+ Fact::ModelObserved {
+ step_id: step.clone(),
+ event: ModelEvent::Finished {
+ reason: ModelFinishReason::ToolCalls,
+ usage: ModelUsage::default(),
+ provider_options: None,
+ },
+ },
+ ))
+ .unwrap(),
+ );
+ }
writes.push(
EventWrite::plain(RuntimeEvent::new(
invocation.clone(),
@@ -206,7 +237,7 @@ async fn seed(
if tool_count == 0 {
20
} else {
- 4 + 2 * tool_count
+ 5 + 3 * tool_count
}
);
event_count += writes.len();
diff --git a/plugins/agent/engine/tests/integration/long_task_responsiveness.rs b/plugins/agent/engine/tests/integration/long_task_responsiveness.rs
new file mode 100644
index 0000000000..e5980a72ed
--- /dev/null
+++ b/plugins/agent/engine/tests/integration/long_task_responsiveness.rs
@@ -0,0 +1,205 @@
+/*
+ * 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.
+ */
+
+//! Release-only history/cancellation probe over explicitly selected disposable
+//! long_task_perf fixture databases. Preparation and warm reads are separate.
+use crate::support::context as fixture;
+use maka_agent_engine::RunError;
+use maka_agent_facts::event::InvocationOutcome;
+use maka_agent_ports::session::history::{Page, Position};
+use maka_event_log::EventLog;
+use serde_json::{Value, json};
+use std::{
+ path::PathBuf,
+ sync::{
+ Arc,
+ atomic::{AtomicUsize, Ordering},
+ },
+ time::{Duration, Instant},
+};
+use tokio::{
+ io::{AsyncReadExt, AsyncWriteExt},
+ net::TcpListener,
+};
+use tokio_util::sync::CancellationToken;
+
+async fn tail(log: &EventLog, through: u64, turns: u64) -> (f64, usize) {
+ let started = Instant::now();
+ let page = log
+ .history_text(
+ "session",
+ through,
+ Position::Tail,
+ false,
+ 16 * 1024,
+ 32,
+ 64 * 1024,
+ )
+ .await
+ .unwrap();
+ let elapsed = started.elapsed().as_secs_f64() * 1000.0;
+ let Page::Ready { chunks, .. } = page else {
+ panic!("prepared history returned Preparing")
+ };
+ assert!(!chunks.is_empty());
+ assert!(chunks.iter().any(|chunk| {
+ chunk
+ .text
+ .contains(&format!("question-history-{}", turns - 1))
+ }));
+ (elapsed, chunks.len())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[ignore = "release responsiveness probe over disposable long-history fixtures"]
+#[allow(clippy::assertions_on_constants)]
+async fn long_history_responsiveness() {
+ assert!(!cfg!(debug_assertions), "run with --release");
+ let paths = std::env::var("MAKA_RESPONSIVENESS_FIXTURES")
+ .expect("explicit comma-separated fixture databases required");
+ let repetitions: usize = std::env::var("MAKA_RESPONSIVENESS_REPETITIONS")
+ .unwrap_or("5".into())
+ .parse()
+ .unwrap();
+ assert!(repetitions > 0);
+ for source in paths.split(',').map(PathBuf::from) {
+ let source = source.canonicalize().unwrap();
+ let temp = std::env::temp_dir().canonicalize().unwrap();
+ assert!(
+ source.starts_with(&temp)
+ || source.starts_with(PathBuf::from("/tmp").canonicalize().unwrap())
+ );
+ let metadata: Value =
+ serde_json::from_slice(&std::fs::read(source.with_extension("json")).unwrap()).unwrap();
+ let turns = metadata["settings"]["turns"].as_u64().unwrap();
+ assert_eq!(metadata["settings"]["checkpoint"], true);
+ let prepared_directory = tempfile::tempdir().unwrap();
+ let prepared = prepared_directory.path().join("prepared.sqlite");
+ std::fs::copy(&source, &prepared).unwrap();
+ let log = EventLog::open(&prepared).await.unwrap();
+ if log.get_session::<Value>("session").await.unwrap().is_none() {
+ log.create_session("session", "responsiveness", &json!({}), 1)
+ .await
+ .unwrap();
+ }
+ let through = *log.subscribe_commits().borrow();
+ let started = Instant::now();
+ let mut preparation_calls = 0;
+ loop {
+ preparation_calls += 1;
+ if log
+ .prepare_transcript("session", through, 32)
+ .await
+ .unwrap()
+ {
+ break;
+ }
+ }
+ let prepare_ms = started.elapsed().as_secs_f64() * 1000.0;
+ let (first_page_ms, chunks) = tail(&log, through, turns).await;
+ println!(
+ "{}",
+ json!({"benchmark":"long_history_preparation", "history_turns":turns,
+ "transcript_prepare_ms":prepare_ms, "preparation_calls":preparation_calls,
+ "first_page_ms":first_page_ms, "chunks":chunks, "errors":0})
+ );
+ log.close().await.unwrap();
+ for phase in ["after_admission_10ms", "model"] {
+ for repetition in 0..repetitions {
+ let directory = tempfile::tempdir().unwrap();
+ let database = directory.path().join("events.sqlite");
+ std::fs::copy(&prepared, &database).unwrap();
+ let log = Arc::new(EventLog::open(&database).await.unwrap());
+ let worker = fixture::engine(log.clone());
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let base = format!("http://{}/v1", listener.local_addr().unwrap());
+ let requests = Arc::new(AtomicUsize::new(0));
+ let observed = requests.clone();
+ let (ready, received) = tokio::sync::oneshot::channel();
+ let server = tokio::spawn(async move {
+ let (mut socket, _) = listener.accept().await.unwrap();
+ fixture::read_request(&mut socket).await;
+ observed.fetch_add(1, Ordering::Relaxed);
+ socket.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n").await.unwrap();
+ let _ = ready.send(());
+ let mut bytes = Vec::new();
+ let _ = socket.read_to_end(&mut bytes).await;
+ });
+ let started = Instant::now();
+ let running = worker
+ .start(
+ fixture::input(&base, "responsive", false),
+ CancellationToken::new(),
+ )
+ .await
+ .unwrap();
+ let invocation = running.invocation().clone();
+ if phase == "model" {
+ tokio::time::timeout(Duration::from_secs(300), received)
+ .await
+ .unwrap()
+ .unwrap();
+ } else {
+ // Deliberate timed cancellation stimulus, not a readiness sleep.
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ }
+ let cancel_at_ms = started.elapsed().as_secs_f64() * 1000.0;
+ let history = tail(&log, through, turns);
+ let cancellation = async move {
+ let started = Instant::now();
+ running.cancel();
+ let completion =
+ tokio::time::timeout(Duration::from_secs(300), running.completion())
+ .await
+ .unwrap();
+ let stop_ms = started.elapsed().as_secs_f64() * 1000.0;
+ assert!(matches!(
+ completion.result,
+ Err(RunError::Cancelled)
+ | Err(RunError::Model(maka_model::ModelError::Cancelled))
+ ));
+ assert!(completion.cleanup.is_ok());
+ stop_ms
+ };
+ let ((history_ms, chunks), stop_ms) = tokio::join!(history, cancellation);
+ worker.drain().await;
+ server.abort();
+ let _ = server.await;
+ let (boundary, _) = log
+ .invocation_boundary("session", &invocation.invocation_id)
+ .await
+ .unwrap()
+ .unwrap();
+ assert!(matches!(
+ boundary.state.terminal_outcome(),
+ Some(InvocationOutcome::Cancelled { .. })
+ ));
+ println!(
+ "{}",
+ json!({"benchmark":"long_history_responsiveness", "history_turns":turns,
+ "phase":phase, "repetition":repetition, "cancel_at_ms":cancel_at_ms,
+ "stop_ms":stop_ms, "concurrent_history_ms":history_ms, "chunks":chunks,
+ "model_requests":requests.load(Ordering::Relaxed), "outcome":"cancelled", "errors":0})
+ );
+ drop(worker);
+ Arc::try_unwrap(log).ok().unwrap().close().await.unwrap();
+ }
+ }
+ }
+}
diff --git a/plugins/agent/engine/tests/integration/main.rs b/plugins/agent/engine/tests/integration/main.rs
index 3d4e869598..d2d696998f 100644
--- a/plugins/agent/engine/tests/integration/main.rs
+++ b/plugins/agent/engine/tests/integration/main.rs
@@ -34,6 +34,7 @@ mod handoff_compaction;
mod image_projection;
mod lifetime;
mod long_task_perf;
+mod long_task_responsiveness;
mod model_interruption;
mod pruning;
mod session_import;
diff --git a/scripts/rust/README.md b/scripts/rust/README.md
index b90a2625ea..6db441d505 100644
--- a/scripts/rust/README.md
+++ b/scripts/rust/README.md
@@ -71,3 +71,35 @@ not Host IPC. The [Engine fixture](../../plugins/agent/engine/tests/integration/
and [prune fixture](../../platform/event-log/tests/archive/performance.rs) expose
workload settings for sparse tools, dense tools and projection sizes. Run without
competing builds; collect CPU profiles separately from timing comparisons.
+
+For 1,000 / 10,000 / 100,000-turn scale probes, set `MAKA_ENGINE_PERF_FIXTURE_DIR`
+to a disposable directory under the system temporary directory and
+`MAKA_ENGINE_PERF_FIXTURE_TAG` to an explicit revision label. The Engine fixture
+uses real append validation and compacts every 1,000 turns. It caches closed
+SQLite snapshots using all workload settings and a hash of its own source;
+changing the fixture invalidates the cache. Run `--kind engine --checkpoint
+--sizes 1000,10000,100000` against retained executables with identical fixture
+source. Seeding is deliberately separate from request latency and can take much
+longer than the measured run.
+
+For history and cancellation, set `MAKA_RESPONSIVENESS_FIXTURES` to a
+comma-separated list of those disposable `.sqlite` snapshots and use
+`--kind responsive`. This checks cold transcript preparation, a real history
+page, and cancellation both 10 ms after admission and after model HTTP receipt.
+The first trigger does not guarantee the Engine is still preparing; inspect the
+recorded model request count. Stop timing includes Engine completion and cleanup,
+with a verified durable cancelled outcome, but excludes closing the EventLog and
+draining abandoned read snapshots. It measures the store/Engine APIs, not GUI or
+Host IPC latency. Context/prune reads interrupt abandoned SQLite work through a
+progress handler and await transaction cleanup before reuse. Separate behavior
+tests cover reclaiming both readers and transaction recovery; arbitrary
+synchronous Rust computation is not immediately interruptible.
+
+`--kind transcript --sizes 128,512` uses retained release
+`maka-event-log --test transcript_index` executables to check many model steps
+inside one invocation. It verifies every assistant message after preparing each
+fresh copy; this workload complements many-turn Engine fixtures.
+
+`--kind executor --sizes 128,512` uses the same transcript executable for
+sequential external tools. It compares every persisted message with canonical
+replay, including progress, results, and final invocation state.
diff --git a/scripts/rust/README.zh-CN.md b/scripts/rust/README.zh-CN.md
index ef309f1066..06631c3901 100644
--- a/scripts/rust/README.zh-CN.md
+++ b/scripts/rust/README.zh-CN.md
@@ -65,3 +65,27 @@ python3 scripts/rust/benchmark-long-task.py --kind engine \
Host IPC。[Engine 夹具](../../plugins/agent/engine/tests/integration/long_task_perf.rs)
和 [prune 夹具](../../platform/event-log/tests/archive/performance.rs) 提供工具稀疏、
工具密集及投影大小等参数。测量时不要同时构建；CPU 剖析与耗时对比分开运行。
+
+测量 1,000 / 10,000 / 100,000 轮时，将 `MAKA_ENGINE_PERF_FIXTURE_DIR` 指向
+系统临时目录内的一次性目录，并用 `MAKA_ENGINE_PERF_FIXTURE_TAG` 明确标记版本。
+Engine 夹具走真实 append 校验，每 1,000 轮执行真实压缩；关闭后的 SQLite 快照
+按全部负载参数及夹具源码哈希缓存，修改夹具会使缓存失效。对夹具源码一致的保留
+可执行文件运行 `--kind engine --checkpoint --sizes 1000,10000,100000`。
+造数据与请求耗时分开，前者可能远长于实际测量过程。
+
+历史与取消测试使用 `--kind responsive`，并将 `MAKA_RESPONSIVENESS_FIXTURES`
+设为上述一次性 `.sqlite` 快照路径，以逗号分隔。它验证冷 transcript 准备、真实
+历史页，以及 admission 完成后 10 ms 和模型 HTTP 请求到达后的取消。第一个触发点
+不保证引擎仍处于准备阶段，需核对模型请求计数。停止耗时包括 Engine 完成和清理，
+并核对持久化的取消结果；不包含 EventLog 关闭时等待被放弃读取结束的时间。
+这些是存储／Engine API 指标，不是 GUI 或 Host IPC 延迟。上下文与裁剪读取在
+调用方取消后通过 SQLite progress handler 中断，并等事务清理完成后复用连接；
+双 reader 回收及事务恢复另有行为测试，不代表任意同步 Rust 计算都可立即中断。
+
+`--kind transcript --sizes 128,512` 使用保留的 release
+`maka-event-log --test transcript_index` 可执行文件，补测同一 invocation 内的
+大量模型步骤。每个样本从新副本生成 transcript，并逐条核对助手消息内容。
+
+`--kind executor --sizes 128,512` 用同一 transcript 可执行文件测量外部执行器
+的连续工具调用，并将全部持久化消息与 canonical replay 对比，覆盖工具进度、
+结果和最终 invocation 状态。
diff --git a/scripts/rust/benchmark-long-task.py b/scripts/rust/benchmark-long-task.py
index 603de23f49..288e1502ea 100644
--- a/scripts/rust/benchmark-long-task.py
+++ b/scripts/rust/benchmark-long-task.py
@@ -41,16 +41,20 @@ def records(kind, output):
for line in output.splitlines():
if line.startswith('{'):
row = json.loads(line)
- if row.get("benchmark") in ("engine_long_history", "archive_context") and not row.get("summary"):
+ if row.get("benchmark") in ("engine_long_history", "archive_context",
+ "long_history_preparation", "long_history_responsiveness",
+ "one_invocation_transcript", "executor_invocation_transcript") and not row.get("summary"):
if kind == "context":
row["history_turns"] = row["archived_count"]
+ elif kind in ("transcript", "executor"):
+ row["history_turns"] = row["model_steps" if kind == "transcript" else "tool_calls"]
result.append(row)
return result
def main():
parser = argparse.ArgumentParser(description=__doc__)
- parser.add_argument("--kind", choices=["prune", "engine", "context"], required=True)
+ parser.add_argument("--kind", choices=["prune", "engine", "context", "responsive", "transcript", "executor"], required=True)
parser.add_argument("--baseline", type=Path, required=True)
parser.add_argument("--candidate", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
@@ -65,15 +69,21 @@ def main():
args.output.mkdir(parents=True, exist_ok=True)
environment = os.environ.copy()
prefix = {"prune": "MAKA_PRUNE_BENCH", "engine": "MAKA_ENGINE_PERF",
- "context": "MAKA_CONTEXT_BENCH"}[args.kind]
- environment[prefix + ("_ARCHIVES" if args.kind == "context" else "_TURNS")] = args.sizes
+ "context": "MAKA_CONTEXT_BENCH", "responsive": "MAKA_RESPONSIVENESS",
+ "transcript": "MAKA_TRANSCRIPT_BENCH", "executor": "MAKA_EXECUTOR_BENCH"}[args.kind]
+ if args.kind != "responsive":
+ suffix = {"context": "_ARCHIVES", "transcript": "_STEPS", "executor": "_STEPS"}.get(args.kind, "_TURNS")
+ environment[prefix + suffix] = args.sizes
environment[prefix + "_REPETITIONS"] = "1"
if args.kind == "engine":
environment[prefix + "_CHECKPOINT"] = str(int(args.checkpoint))
test = {"prune": "performance::long_history_prune",
"engine": "long_task_perf::engine_long_history",
- "context": "context_performance::archive_heavy_context"}[args.kind]
- evidence = {"kind": args.kind, "sizes": args.sizes, "checkpoint": args.checkpoint,
+ "context": "context_performance::archive_heavy_context",
+ "responsive": "long_task_responsiveness::long_history_responsiveness",
+ "transcript": "performance::one_invocation_transcript",
+ "executor": "performance_executor::executor_invocation_transcript"}[args.kind]
+ evidence = {"kind": args.kind, "sizes": None if args.kind == "responsive" else args.sizes, "checkpoint": args.checkpoint,
"runs": args.runs, "cpu_count": os.cpu_count(),
"load_average": list(os.getloadavg()),
"environment": {k: v for k, v in environment.items() if k.startswith(prefix)},
@@ -95,6 +105,8 @@ def main():
"exit_code": process.returncode, "log": str(output)})
result_file.write_text(json.dumps(evidence, indent=2) + "\n")
raise SystemExit(f"benchmark failed; inspect {output}")
+ if args.kind == "responsive":
+ evidence["sizes"] = ",".join(map(str, sorted({row["history_turns"] for row in rows})))
work = [{key: value for key, value in row.items()
if not key.endswith("_ms") and key not in
("repetition", "database_bytes", "database_wal_bytes", "fixture_cached")}
@@ -109,7 +121,7 @@ def main():
for row in rows:
sample = dict(row, pair=pair + 1, revision=label)
evidence["samples"].append(sample)
- key = (label, row["history_turns"], row.get("shape", "engine"),
+ key = (label, row["history_turns"], row.get("shape", row.get("phase", row.get("benchmark", "engine"))),
row.get("archived", args.checkpoint))
groups.setdefault(key, []).append(row)
result_file.write_text(json.dumps(evidence, indent=2) + "\n")
@@ -118,7 +130,9 @@ def main():
summary = []
for key, rows in groups.items():
metrics = {}
- for metric in ["prune_ms", "context_ms", "request_receipt_ms", "engine_completion_ms"]:
+ for metric in ["prune_ms", "context_ms", "request_receipt_ms", "engine_completion_ms",
+ "transcript_prepare_ms", "first_page_ms", "stop_ms", "concurrent_history_ms",
+ "store_open_ms", "cancel_at_ms"]:
if metric in rows[0]:
values = [row[metric] for row in rows]
metrics[metric] = dict(n=len(values), median=statistics.median(values),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org