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/22 11:50:13 UTC
(maka) branch rust-preview updated: perf(event-log): group observation commits and index session materials
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 48da268e37 perf(event-log): group observation commits and index session materials
48da268e37 is described below
commit 48da268e37303c3d504a4595c3a83f28e10bd8b0
Author: Wang <m4...@gmail.com>
AuthorDate: Thu Oct 22 19:49:12 2026 +0800
perf(event-log): group observation commits and index session materials
---
.../migrations/0039_material_session_lookup.sql | 29 ++
platform/event-log/src/append.rs | 13 +
platform/event-log/src/append/observations.rs | 231 ++++++++++
.../event-log/src/append/observations/tests.rs | 483 +++++++++++++++++++++
.../event-log/src/bundle/validation/artifacts.rs | 2 +-
platform/event-log/src/connection.rs | 80 +++-
platform/event-log/src/context/cache.rs | 2 +-
platform/event-log/src/context/evidence/scan.rs | 2 +-
platform/event-log/src/context/safety.rs | 2 +-
platform/event-log/src/context/selection.rs | 2 +-
platform/event-log/src/context/usage.rs | 2 +-
platform/event-log/src/message_sources.rs | 53 ++-
platform/event-log/src/observation.rs | 8 +-
platform/event-log/src/observation/delivery.rs | 2 +-
platform/event-log/src/sessions/material.rs | 87 ++++
.../event-log/src/transcript/navigation/queries.rs | 23 +-
platform/event-log/src/workspace.rs | 2 +-
.../event-log/tests/migrations/material_session.rs | 236 ++++++++++
.../event-log/tests/migrations/prior_artifacts.rs | 14 +-
platform/event-log/tests/session_copy/mod.rs | 20 +
20 files changed, 1239 insertions(+), 54 deletions(-)
diff --git a/platform/event-log/migrations/0039_material_session_lookup.sql b/platform/event-log/migrations/0039_material_session_lookup.sql
new file mode 100644
index 0000000000..c738a0528f
--- /dev/null
+++ b/platform/event-log/migrations/0039_material_session_lookup.sql
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+
+-- Share the live-material key without changing inherited history ownership.
+DROP VIEW session_history_events;
+CREATE VIEW session_history_events AS
+ SELECT e.event_session AS owner_session_id,
+ 0 AS inherited, NULL AS archive_sequence, e.* FROM runtime_events e
+ UNION ALL
+ SELECT h.session_id, 1, h.archive_sequence, e.*
+ FROM session_history_members h JOIN runtime_events e ON e.sequence = h.sequence;
+
+DROP INDEX session_event_sequence;
diff --git a/platform/event-log/src/append.rs b/platform/event-log/src/append.rs
index e3d4b88bb0..ffd0def9f1 100644
--- a/platform/event-log/src/append.rs
+++ b/platform/event-log/src/append.rs
@@ -27,6 +27,7 @@ use sqlx::{Connection, SqliteConnection};
use crate::{EventLog, StoreError, sequence_number};
pub(crate) mod history;
+pub(crate) mod observations;
impl EventLog {
/// Commit ordered facts atomically, validating each against the preceding
@@ -34,6 +35,18 @@ impl EventLog {
/// until Session removal completes, after which all writes are rejected.
/// This must not enclose effects: live dispatch and outcome remain separate.
pub async fn append_batch(&self, events: &[EventWrite]) -> Result<Vec<u64>, CommitError> {
+ if !events.is_empty()
+ && events
+ .iter()
+ .all(|write| matches!(write.event().fact, Fact::ModelObserved { .. }))
+ {
+ self.validate_root().map_err(commit_error)?;
+ return self
+ .connection
+ .writer()
+ .observations(events.to_vec(), self.commits.clone())
+ .await;
+ }
self.append_batch_checked(events, None)
.await
.map_err(commit_error)
diff --git a/platform/event-log/src/append/observations.rs b/platform/event-log/src/append/observations.rs
new file mode 100644
index 0000000000..7053f91b78
--- /dev/null
+++ b/platform/event-log/src/append/observations.rs
@@ -0,0 +1,231 @@
+/*
+ * 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.
+ */
+
+//! Adjacent accepted observation requests share one durable receipt boundary.
+//! Savepoints isolate domain rejection; database failures retire the original owner.
+use std::sync::{
+ Arc,
+ atomic::{AtomicU64, Ordering},
+};
+
+use maka_agent_facts::event::{CommitError, EventWrite};
+use sqlx::{Connection, SqliteConnection};
+use tokio::sync::{oneshot, watch};
+
+use super::{AppendResult, commit_error};
+use crate::{EventLog, StoreError};
+
+pub(crate) struct Request {
+ events: Vec<EventWrite>,
+ commits: watch::Sender<u64>,
+ reply: oneshot::Sender<Result<Vec<u64>, CommitError>>,
+ failures: Arc<AtomicU64>,
+ outcome: Outcome,
+}
+
+enum Outcome {
+ Waiting,
+ Writing,
+ Prepared {
+ sequences: Vec<u64>,
+ inserted: Option<u64>,
+ },
+ Rejected(CommitError),
+}
+
+impl Request {
+ pub(crate) fn new(
+ events: Vec<EventWrite>,
+ commits: watch::Sender<u64>,
+ reply: oneshot::Sender<Result<Vec<u64>, CommitError>>,
+ failures: Arc<AtomicU64>,
+ ) -> Self {
+ Self {
+ events,
+ commits,
+ reply,
+ failures,
+ outcome: Outcome::Waiting,
+ }
+ }
+}
+
+struct Aborted {
+ message: String,
+ unknown: bool,
+}
+
+pub(crate) async fn commit(
+ connection: &mut SqliteConnection,
+ mut requests: Vec<Request>,
+ validate: &(impl Fn() -> Result<(), StoreError> + Sync),
+) -> bool {
+ let result = transaction(connection, &mut requests, validate).await;
+ for request in requests {
+ let outcome = match request.outcome {
+ Outcome::Rejected(error) => Err(error),
+ Outcome::Prepared {
+ sequences,
+ inserted,
+ } if result.is_ok() => {
+ if let Some(sequence) = inserted {
+ request.commits.send_replace(sequence);
+ }
+ Ok(sequences)
+ }
+ state => {
+ let failure = result.as_ref().expect_err("unprepared observation receipt");
+ if failure.unknown && !matches!(state, Outcome::Waiting) {
+ Err(CommitError::OutcomeUnknown(failure.message.clone()))
+ } else {
+ Err(CommitError::Rejected(failure.message.clone()))
+ }
+ }
+ };
+ if outcome.is_err() {
+ request.failures.fetch_add(1, Ordering::AcqRel);
+ }
+ let _ = request.reply.send(outcome);
+ }
+ result.is_ok()
+}
+
+async fn transaction(
+ connection: &mut SqliteConnection,
+ requests: &mut [Request],
+ validate: &(impl Fn() -> Result<(), StoreError> + Sync),
+) -> Result<(), Aborted> {
+ validate().map_err(|error| Aborted {
+ message: error.to_string(),
+ unknown: false,
+ })?;
+ let mut transaction = connection
+ .begin_with("BEGIN IMMEDIATE")
+ .await
+ .map_err(|error| Aborted {
+ message: error.to_string(),
+ unknown: false,
+ })?;
+ let prepared = prepare(&mut transaction, requests, validate).await;
+ match prepared {
+ Ok(true) => transaction.commit().await.map_err(|error| Aborted {
+ message: error.to_string(),
+ unknown: true,
+ }),
+ Ok(false) => transaction.rollback().await.map_err(|error| Aborted {
+ message: error.to_string(),
+ unknown: true,
+ }),
+ Err(error) => {
+ // No commit was attempted. Only a confirmed full rollback establishes
+ // rejection; a dropped transaction is never a rollback receipt.
+ let rollback = transaction.rollback().await;
+ Err(Aborted {
+ message: match &rollback {
+ Ok(()) => error.to_string(),
+ Err(rollback) => format!("{error}; rollback unconfirmed: {rollback}"),
+ },
+ unknown: rollback.is_err(),
+ })
+ }
+ }
+}
+
+async fn prepare(
+ connection: &mut SqliteConnection,
+ requests: &mut [Request],
+ validate: &(impl Fn() -> Result<(), StoreError> + Sync),
+) -> Result<bool, StoreError> {
+ let mut inserted_any = false;
+ for request in requests {
+ validate()?;
+ request.outcome = Outcome::Writing;
+ // Raw savepoints deliberately do not add SQLx Transaction depth. Any
+ // SQL/rollback/release failure abandons the whole group and this owner.
+ sqlx::query("SAVEPOINT observation_request")
+ .execute(&mut *connection)
+ .await?;
+ match prepare_request(connection, &request.events).await {
+ Ok((sequences, inserted)) => {
+ sqlx::query("RELEASE observation_request")
+ .execute(&mut *connection)
+ .await?;
+ inserted_any |= inserted.is_some();
+ request.outcome = Outcome::Prepared {
+ sequences,
+ inserted,
+ };
+ }
+ Err(error) if isolated_rejection(&error) => {
+ sqlx::query("ROLLBACK TO observation_request")
+ .execute(&mut *connection)
+ .await?;
+ sqlx::query("RELEASE observation_request")
+ .execute(&mut *connection)
+ .await?;
+ request.outcome = Outcome::Rejected(commit_error(error));
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ validate()?;
+ Ok(inserted_any)
+}
+
+async fn prepare_request(
+ connection: &mut SqliteConnection,
+ events: &[EventWrite],
+) -> Result<(Vec<u64>, Option<u64>), StoreError> {
+ let mut verification = crate::context::cache::Verification::default();
+ let mut sequences = Vec::with_capacity(events.len());
+ let mut inserted = None;
+ for event in events {
+ sequences.push(
+ match EventLog::append_verified(connection, event, &mut verification).await? {
+ AppendResult::Existing(sequence) => sequence,
+ AppendResult::Inserted(sequence) => {
+ inserted = Some(sequence);
+ sequence
+ }
+ },
+ );
+ }
+ crate::context::validate_batch(connection, events).await?;
+ Ok((sequences, inserted))
+}
+
+fn isolated_rejection(error: &StoreError) -> bool {
+ matches!(
+ error,
+ StoreError::Sealed
+ | StoreError::ImportedInvocation
+ | StoreError::EventConflict
+ | StoreError::SessionRetired
+ | StoreError::PermissionRevoked
+ | StoreError::InvalidTransition(_)
+ | StoreError::HistoryTurnMissing
+ | StoreError::HistoryTurnSplit
+ | StoreError::HistoryUnsettled
+ | StoreError::HistoryProviderEffectUnresolved
+ | StoreError::MaterialCollected
+ )
+}
+
+#[cfg(test)]
+mod tests;
diff --git a/platform/event-log/src/append/observations/tests.rs b/platform/event-log/src/append/observations/tests.rs
new file mode 100644
index 0000000000..f2c8ccef7c
--- /dev/null
+++ b/platform/event-log/src/append/observations/tests.rs
@@ -0,0 +1,483 @@
+/*
+ * 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 maka_agent_facts::{
+ event::{Fact, Invocation, InvocationInput, LogScope, RuntimeEvent},
+ model::{ModelEvent, TextKind},
+};
+use std::{sync::atomic::AtomicUsize, time::Duration};
+
+fn event(id: &str, fact: Fact) -> EventWrite {
+ let mut event = RuntimeEvent::new(
+ Invocation {
+ session_id: "session".into(),
+ turn_id: "turn".into(),
+ run_id: "run".into(),
+ invocation_id: "invocation".into(),
+ },
+ fact,
+ );
+ event.id = id.into();
+ EventWrite::plain(event).unwrap()
+}
+
+fn delta(id: &str, text: &str) -> EventWrite {
+ event(
+ id,
+ Fact::ModelObserved {
+ step_id: "step".into(),
+ event: ModelEvent::PartDelta {
+ id: "part".into(),
+ text: text.into(),
+ provider_options: None,
+ },
+ },
+ )
+}
+
+async fn open(path: &std::path::Path) -> Arc<EventLog> {
+ let log = Arc::new(EventLog::open(path).await.unwrap());
+ log.append_batch(&[
+ event(
+ "opened",
+ Fact::InvocationOpened {
+ configuration: None,
+ input: InvocationInput::Message {
+ source_messages: vec![],
+ content: "user".into(),
+ request_fingerprint: None,
+ },
+ },
+ ),
+ event(
+ "requested",
+ Fact::ModelRequested {
+ context_breakdown: None,
+ item_acceptance: false,
+ purpose: maka_agent_facts::context::ModelPurpose::Main,
+ context: None,
+ checkpoint_event_id: None,
+ step_id: "step".into(),
+ model_id: "model".into(),
+ source_scope: LogScope::Root,
+ source_high_water: 0,
+ source_digest: "source".into(),
+ effective_source_digest: None,
+ input_digest: "input".into(),
+ route_identity: "route".into(),
+ },
+ ),
+ event(
+ "part",
+ Fact::ModelObserved {
+ step_id: "step".into(),
+ event: ModelEvent::PartStarted {
+ id: "part".into(),
+ text_kind: TextKind::Text,
+ provider_options: None,
+ },
+ },
+ ),
+ ])
+ .await
+ .unwrap();
+ log
+}
+
+async fn gate(log: &Arc<EventLog>) -> (oneshot::Sender<()>, tokio::task::JoinHandle<()>) {
+ let (entered, reached) = oneshot::channel();
+ let (release, gate) = oneshot::channel();
+ let log = log.clone();
+ let job = tokio::spawn(async move {
+ log.connection
+ .run(move |_| {
+ Box::pin(async move {
+ entered.send(()).unwrap();
+ gate.await.unwrap();
+ Ok(())
+ })
+ })
+ .await
+ .unwrap();
+ });
+ tokio::time::timeout(Duration::from_secs(5), reached)
+ .await
+ .unwrap()
+ .unwrap();
+ (release, job)
+}
+
+async fn hook(log: &EventLog, callback: impl FnMut() -> bool + Send + 'static) {
+ log.connection
+ .run(move |db| {
+ Box::pin(async move {
+ db.lock_handle().await?.set_commit_hook(callback);
+ Ok(())
+ })
+ })
+ .await
+ .unwrap();
+}
+
+fn ids(path: &std::path::Path) -> Vec<String> {
+ let db = rusqlite::Connection::open(path).unwrap();
+ db.prepare("SELECT event_id FROM runtime_events ORDER BY sequence")
+ .unwrap()
+ .query_map([], |r| r.get(0))
+ .unwrap()
+ .collect::<Result<_, _>>()
+ .unwrap()
+}
+
+#[tokio::test]
+async fn neighbors_share_commit_and_isolate_bad_batch_and_conflicting_replay() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = open(&path).await;
+ let count = Arc::new(AtomicUsize::new(0));
+ let observed = count.clone();
+ hook(&log, move || {
+ observed.fetch_add(1, Ordering::SeqCst);
+ true
+ })
+ .await;
+ let before_failures = log.connection.writer().failures();
+ let (release, gate) = gate(&log).await;
+ let first = [delta("first", "one")];
+ let bad = [
+ delta("rolled-back", "bad prefix"),
+ event(
+ "bad-tail",
+ Fact::ModelObserved {
+ step_id: "missing".into(),
+ event: ModelEvent::PartDelta {
+ id: "part".into(),
+ text: "bad tail".into(),
+ provider_options: None,
+ },
+ },
+ ),
+ ];
+ let conflict = [delta("first", "different")];
+ let last = [delta("last", "two")];
+ let mut a = Box::pin(log.append_batch(&first));
+ let mut b = Box::pin(log.append_batch(&bad));
+ let mut replay = Box::pin(log.append_batch(&first));
+ let mut c = Box::pin(log.append_batch(&conflict));
+ let mut d = Box::pin(log.append_batch(&last));
+ assert!(futures_util::poll!(&mut a).is_pending());
+ assert!(futures_util::poll!(&mut b).is_pending());
+ assert!(futures_util::poll!(&mut replay).is_pending());
+ assert!(futures_util::poll!(&mut c).is_pending());
+ assert!(futures_util::poll!(&mut d).is_pending());
+ release.send(()).unwrap();
+ gate.await.unwrap();
+ assert_eq!(a.await.unwrap(), [4]);
+ assert!(matches!(b.await, Err(CommitError::Rejected(_))));
+ assert_eq!(replay.await.unwrap(), [4]);
+ assert!(matches!(c.await, Err(CommitError::Rejected(_))));
+ assert_eq!(d.await.unwrap(), [5]);
+ assert_eq!(count.load(Ordering::SeqCst), 1);
+ assert_eq!(log.connection.writer().failures() - before_failures, 2);
+ assert_eq!(ids(&path), ["opened", "requested", "part", "first", "last"]);
+ let mut commits = log.subscribe_commits();
+ commits.borrow_and_update();
+ assert_eq!(log.append_batch(&first).await.unwrap(), [4]);
+ assert!(!commits.has_changed().unwrap());
+ assert_eq!(*commits.borrow(), 5);
+ assert_eq!(
+ count.load(Ordering::SeqCst),
+ 1,
+ "replay has no durable write"
+ );
+ log.connection.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn commit_failure_preserves_domain_rejection_and_stops_pending_control() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = open(&path).await;
+ hook(&log, || false).await;
+ let (release, gate) = gate(&log).await;
+ let first = [delta("first", "one")];
+ let bad = [delta("part", "conflicting original")];
+ let last = [delta("last", "two")];
+ let mut a = Box::pin(log.append_batch(&first));
+ let mut b = Box::pin(log.append_batch(&bad));
+ let mut c = Box::pin(log.append_batch(&last));
+ assert!(futures_util::poll!(&mut a).is_pending());
+ assert!(futures_util::poll!(&mut b).is_pending());
+ assert!(futures_util::poll!(&mut c).is_pending());
+ let ran = Arc::new(AtomicUsize::new(0));
+ let called = ran.clone();
+ let mut control = Box::pin(log.connection.run(move |_| {
+ Box::pin(async move {
+ called.fetch_add(1, Ordering::SeqCst);
+ Ok(())
+ })
+ }));
+ assert!(futures_util::poll!(&mut control).is_pending());
+ release.send(()).unwrap();
+ gate.await.unwrap();
+ assert!(matches!(a.await, Err(CommitError::OutcomeUnknown(_))));
+ assert!(matches!(b.await, Err(CommitError::Rejected(_))));
+ assert!(matches!(c.await, Err(CommitError::OutcomeUnknown(_))));
+ assert!(matches!(control.await, Err(StoreError::OperationUnknown)));
+ assert_eq!(ran.load(Ordering::SeqCst), 0);
+ log.connection.shutdown().await.unwrap();
+ assert_eq!(ids(&path), ["opened", "requested", "part"]);
+ let reopened = EventLog::open(&path).await.unwrap();
+ assert_eq!(reopened.append_batch(&first).await.unwrap(), [4]);
+ reopened.close().await.unwrap();
+}
+
+#[tokio::test]
+async fn receipts_wait_for_commit_and_dropped_waiter_is_drained_on_shutdown() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = open(&path).await;
+ let (entered, reached) = oneshot::channel();
+ let mut entered = Some(entered);
+ let (release_commit, commit_gate) = std::sync::mpsc::channel();
+ hook(&log, move || {
+ entered.take().unwrap().send(()).unwrap();
+ commit_gate.recv_timeout(Duration::from_secs(10)).unwrap();
+ true
+ })
+ .await;
+ let (release, gate) = gate(&log).await;
+ let first = [delta("first", "one")];
+ let last = [delta("last", "two")];
+ let mut a = Box::pin(log.append_batch(&first));
+ let mut b = Box::pin(log.append_batch(&last));
+ assert!(futures_util::poll!(&mut a).is_pending());
+ assert!(futures_util::poll!(&mut b).is_pending());
+ let mut commits = log.subscribe_commits();
+ commits.borrow_and_update();
+ release.send(()).unwrap();
+ gate.await.unwrap();
+ tokio::time::timeout(Duration::from_secs(5), reached)
+ .await
+ .unwrap()
+ .unwrap();
+ assert!(futures_util::poll!(&mut a).is_pending());
+ assert!(futures_util::poll!(&mut b).is_pending());
+ assert!(!commits.has_changed().unwrap());
+ assert_eq!(ids(&path), ["opened", "requested", "part"]);
+ drop(a);
+ let mut close = Box::pin(log.connection.shutdown());
+ assert!(futures_util::poll!(&mut close).is_pending());
+ release_commit.send(()).unwrap();
+ assert_eq!(b.await.unwrap(), [5]);
+ close.await.unwrap();
+ assert_eq!(ids(&path), ["opened", "requested", "part", "first", "last"]);
+}
+
+#[tokio::test]
+async fn generic_control_keeps_its_original_position_between_observation_groups() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = open(&path).await;
+ let count = Arc::new(AtomicUsize::new(0));
+ let observed = count.clone();
+ hook(&log, move || {
+ observed.fetch_add(1, Ordering::SeqCst);
+ true
+ })
+ .await;
+ let (release, gate) = gate(&log).await;
+ let first = [delta("first", "one")];
+ let last = [delta("last", "two")];
+ let mut a = Box::pin(log.append_batch(&first));
+ assert!(futures_util::poll!(&mut a).is_pending());
+ let mut control = Box::pin(log.connection.run(|db| {
+ Box::pin(async move {
+ Ok(
+ sqlx::query_scalar::<_, i64>("SELECT max(sequence) FROM runtime_events")
+ .fetch_one(db)
+ .await?,
+ )
+ })
+ }));
+ assert!(futures_util::poll!(&mut control).is_pending());
+ let mut b = Box::pin(log.append_batch(&last));
+ assert!(futures_util::poll!(&mut b).is_pending());
+ let mut close = Box::pin(log.connection.shutdown());
+ assert!(futures_util::poll!(&mut close).is_pending());
+ release.send(()).unwrap();
+ gate.await.unwrap();
+ assert_eq!(a.await.unwrap(), [4]);
+ assert_eq!(control.await.unwrap(), 4);
+ assert_eq!(b.await.unwrap(), [5]);
+ close.await.unwrap();
+ assert_eq!(count.load(Ordering::SeqCst), 2);
+ assert_eq!(ids(&path), ["opened", "requested", "part", "first", "last"]);
+}
+
+#[tokio::test]
+async fn savepoint_failures_abort_the_group_and_close_without_running_control() {
+ unsafe extern "C" fn deny_savepoint(
+ mode: *mut std::ffi::c_void,
+ action: std::ffi::c_int,
+ operation: *const std::ffi::c_char,
+ _: *const std::ffi::c_char,
+ _: *const std::ffi::c_char,
+ _: *const std::ffi::c_char,
+ ) -> std::ffi::c_int {
+ if action == rusqlite::ffi::SQLITE_SAVEPOINT && !operation.is_null() {
+ // SAFETY: SQLite supplies a callback-local NUL-terminated operation.
+ let operation = unsafe { std::ffi::CStr::from_ptr(operation) }.to_bytes();
+ let denied = if mode as usize == 1 {
+ b"RELEASE".as_slice()
+ } else {
+ b"ROLLBACK".as_slice()
+ };
+ if operation == denied {
+ return rusqlite::ffi::SQLITE_DENY;
+ }
+ }
+ rusqlite::ffi::SQLITE_OK
+ }
+ for mode in [1_usize, 2] {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = open(&path).await;
+ log.connection
+ .run(move |db| {
+ Box::pin(async move {
+ let mut handle = db.lock_handle().await?;
+ // SAFETY: static callback; mode is an integer tag, never dereferenced.
+ // The exclusive handle is borrowed only for installing SQLite's authorizer.
+ assert_eq!(
+ unsafe {
+ rusqlite::ffi::sqlite3_set_authorizer(
+ handle.as_raw_handle().as_mut(),
+ Some(deny_savepoint),
+ mode as *mut std::ffi::c_void,
+ )
+ },
+ rusqlite::ffi::SQLITE_OK
+ );
+ Ok(())
+ })
+ })
+ .await
+ .unwrap();
+ let (release, gate) = gate(&log).await;
+ let first = [delta("first", "one")];
+ let bad = [delta("rolled-back", "prefix"), delta("part", "conflict")];
+ let last = [delta("last", "two")];
+ let mut a = Box::pin(log.append_batch(&first));
+ let mut b = Box::pin(log.append_batch(&bad));
+ let mut c = Box::pin(log.append_batch(&last));
+ assert!(futures_util::poll!(&mut a).is_pending());
+ assert!(futures_util::poll!(&mut b).is_pending());
+ assert!(futures_util::poll!(&mut c).is_pending());
+ let mut control = Box::pin(log.connection.run::<()>(|_| {
+ Box::pin(async {
+ panic!("failed savepoint must retire the owner before pending control")
+ })
+ }));
+ assert!(futures_util::poll!(&mut control).is_pending());
+ release.send(()).unwrap();
+ gate.await.unwrap();
+ for result in [a.await, b.await, c.await] {
+ assert!(
+ matches!(result, Err(CommitError::Rejected(_))),
+ "outer rollback confirmed: {result:?}"
+ );
+ }
+ assert!(matches!(control.await, Err(StoreError::OperationUnknown)));
+ log.connection.shutdown().await.unwrap();
+ assert_eq!(ids(&path), ["opened", "requested", "part"]);
+ let reopened = EventLog::open(&path).await.unwrap();
+ assert_eq!(reopened.append_batch(&first).await.unwrap(), [4]);
+ reopened.close().await.unwrap();
+ }
+}
+
+#[tokio::test]
+async fn sqlite_automatic_rollback_is_not_falsely_reported_as_confirmed_rollback() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = open(&path).await;
+ log.connection.run(|db| Box::pin(async move {
+ sqlx::raw_sql("CREATE TEMP TRIGGER rollback_observation BEFORE INSERT ON event_log
+ WHEN NEW.event_id='poison' BEGIN SELECT RAISE(ROLLBACK, 'injected storage failure'); END;")
+ .execute(db).await?;
+ Ok(())
+ })).await.unwrap();
+ let (release, gate) = gate(&log).await;
+ let first = [delta("first", "one")];
+ let poison = [delta("poison", "two")];
+ let last = [delta("last", "three")];
+ let mut a = Box::pin(log.append_batch(&first));
+ let mut b = Box::pin(log.append_batch(&poison));
+ let mut c = Box::pin(log.append_batch(&last));
+ assert!(futures_util::poll!(&mut a).is_pending());
+ assert!(futures_util::poll!(&mut b).is_pending());
+ assert!(futures_util::poll!(&mut c).is_pending());
+ release.send(()).unwrap();
+ gate.await.unwrap();
+ assert!(matches!(a.await, Err(CommitError::OutcomeUnknown(_))));
+ assert!(matches!(b.await, Err(CommitError::OutcomeUnknown(_))));
+ assert!(
+ matches!(c.await, Err(CommitError::Rejected(_))),
+ "unstarted request has no uncertain write"
+ );
+ log.connection.shutdown().await.unwrap();
+ assert_eq!(ids(&path), ["opened", "requested", "part"]);
+}
+
+#[tokio::test]
+async fn authority_lost_during_last_request_rolls_back_before_commit() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = open(&path).await;
+ let lock = dir.path().join("events.sqlite.writer.lock");
+ let displaced = dir.path().join("displaced.lock");
+ log.connection
+ .run(move |db| {
+ Box::pin(async move {
+ db.lock_handle().await?.set_update_hook(move |update| {
+ if update.table == "event_log" && update.rowid == 5 {
+ std::fs::rename(&lock, &displaced).unwrap();
+ }
+ });
+ Ok(())
+ })
+ })
+ .await
+ .unwrap();
+ let (release, gate) = gate(&log).await;
+ let first = [delta("first", "one")];
+ let last = [delta("last", "two")];
+ let mut a = Box::pin(log.append_batch(&first));
+ let mut b = Box::pin(log.append_batch(&last));
+ assert!(futures_util::poll!(&mut a).is_pending());
+ assert!(futures_util::poll!(&mut b).is_pending());
+ release.send(()).unwrap();
+ gate.await.unwrap();
+ assert!(matches!(a.await, Err(CommitError::Rejected(_))));
+ assert!(matches!(b.await, Err(CommitError::Rejected(_))));
+ log.connection.shutdown().await.unwrap();
+ assert_eq!(ids(&path), ["opened", "requested", "part"]);
+}
diff --git a/platform/event-log/src/bundle/validation/artifacts.rs b/platform/event-log/src/bundle/validation/artifacts.rs
index 7f3db6e02b..a90806d344 100644
--- a/platform/event-log/src/bundle/validation/artifacts.rs
+++ b/platform/event-log/src/bundle/validation/artifacts.rs
@@ -99,7 +99,7 @@ pub(super) async fn validate(
let row: Option<(i64, bool, bool, String)> = sqlx::query_as(
"SELECT sequence,revision,inherited,event_json FROM (
SELECT e.sequence,0 AS revision,0 AS inherited,e.event_json FROM runtime_events e
- WHERE json_extract(e.event_json,'$.invocation.session_id')=?1 AND e.sequence>?2
+ WHERE (e.event_json IS NOT NULL AND e.event_session=?1) AND e.sequence>?2
ORDER BY e.sequence LIMIT 1)
UNION ALL
SELECT sequence,revision,inherited,event_json FROM (
diff --git a/platform/event-log/src/connection.rs b/platform/event-log/src/connection.rs
index 721d27ffac..f6817ac0d6 100644
--- a/platform/event-log/src/connection.rs
+++ b/platform/event-log/src/connection.rs
@@ -53,6 +53,7 @@ type ReadJob = Box<
>;
enum Command {
Run(Job),
+ Observations(crate::append::observations::Request),
Shutdown,
}
@@ -77,6 +78,29 @@ pub(crate) struct WriterIngress {
}
impl WriterIngress {
+ pub(crate) async fn observations(
+ &self,
+ events: Vec<maka_agent_facts::event::EventWrite>,
+ commits: watch::Sender<u64>,
+ ) -> Result<Vec<u64>, maka_agent_facts::event::CommitError> {
+ use maka_agent_facts::event::CommitError;
+ let (reply, result) = oneshot::channel();
+ self.jobs
+ .send(Command::Observations(
+ crate::append::observations::Request::new(
+ events,
+ commits,
+ reply,
+ self.failures.clone(),
+ ),
+ ))
+ .await
+ .map_err(|_| CommitError::Rejected(StoreError::ConnectionClosed.to_string()))?;
+ result
+ .await
+ .map_err(|_| CommitError::OutcomeUnknown(StoreError::OperationUnknown.to_string()))?
+ }
+
pub(crate) fn failures(&self) -> u64 {
self.failures.load(Ordering::Acquire)
}
@@ -171,21 +195,47 @@ impl OwnedConnection {
.unwrap_or(Err(StoreError::OperationUnknown));
let success = initialized.is_ok();
// If open's waiter disappeared, close without accepting work.
+ let mut pending = None;
if ready.send(initialized).is_ok() && success {
- while let Some(command) = runtime.block_on(receiver.recv()) {
- let job = match command {
- Command::Run(job) => job,
- Command::Shutdown => {
- // Reject new sends, then drain every accepted job.
- receiver.close();
- continue;
- }
- };
- let executed = std::panic::catch_unwind(AssertUnwindSafe(|| {
- runtime.block_on(job(&mut connection, validate()))
- }));
- if executed.is_err() {
- // An unwound connection cannot accept further jobs.
+ while let Some(command) =
+ pending.take().or_else(|| runtime.block_on(receiver.recv()))
+ {
+ let executed =
+ std::panic::catch_unwind(AssertUnwindSafe(|| match command {
+ Command::Run(job) => {
+ runtime.block_on(job(&mut connection, validate()));
+ true
+ }
+ Command::Observations(first) => {
+ // Only the already queued adjacent prefix shares a commit.
+ // Keep a control boundary locally, ahead of later ingress.
+ let mut group = vec![first];
+ while group.len() < 32 {
+ match receiver.try_recv() {
+ Ok(Command::Observations(request)) => {
+ group.push(request)
+ }
+ Ok(boundary) => {
+ pending = Some(boundary);
+ break;
+ }
+ Err(_) => break,
+ }
+ }
+ runtime.block_on(crate::append::observations::commit(
+ &mut connection,
+ group,
+ &validate,
+ ))
+ }
+ Command::Shutdown => {
+ // Reject new sends, then drain every accepted job.
+ receiver.close();
+ true
+ }
+ }));
+ if !matches!(executed, Ok(true)) {
+ // Neither an unwound nor uncertain transaction may be reused.
break;
}
match runtime.block_on(
@@ -209,6 +259,8 @@ impl OwnedConnection {
// queued work, including sends holding an earlier capacity permit,
// before joining readers. None also accounts for those permits.
receiver.close();
+ queued_drop_panicked |=
+ std::panic::catch_unwind(AssertUnwindSafe(|| drop(pending))).is_err();
while let Some(command) = runtime.block_on(receiver.recv()) {
queued_drop_panicked |=
std::panic::catch_unwind(AssertUnwindSafe(|| drop(command))).is_err();
diff --git a/platform/event-log/src/context/cache.rs b/platform/event-log/src/context/cache.rs
index bdd16a1000..19a15e8e03 100644
--- a/platform/event-log/src/context/cache.rs
+++ b/platform/event-log/src/context/cache.rs
@@ -398,7 +398,7 @@ impl Verification {
pub(crate) async fn history_hint(connection: &mut SqliteConnection, session: &str) -> bool {
sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM runtime_events e
- WHERE json_extract(e.event_json, '$.invocation.session_id')=?1
+ WHERE (e.event_json IS NOT NULL AND e.event_session=?1)
AND e.kind <> 'invocation_opened')
OR EXISTS(SELECT 1 FROM session_history_members WHERE session_id=?1)",
)
diff --git a/platform/event-log/src/context/evidence/scan.rs b/platform/event-log/src/context/evidence/scan.rs
index bcbddb100a..7e2db6901a 100644
--- a/platform/event-log/src/context/evidence/scan.rs
+++ b/platform/event-log/src/context/evidence/scan.rs
@@ -119,7 +119,7 @@ fn digests(
let sql = format!(
"SELECT e.sequence AS sequence, {payload} FROM runtime_events e
WHERE e.sequence > ?1 AND e.sequence <= ?2
- AND json_extract(e.event_json,'$.invocation.session_id') = ?3 AND {filter}
+ AND (e.event_json IS NOT NULL AND e.event_session=?3) AND {filter}
UNION ALL SELECT h.sequence AS sequence, {payload} FROM session_history_members h
JOIN runtime_events e ON e.sequence = h.sequence
WHERE h.session_id = ?3 AND h.sequence > ?1 AND h.sequence <= ?2 AND {filter}
diff --git a/platform/event-log/src/context/safety.rs b/platform/event-log/src/context/safety.rs
index c81eaffbbf..a3301a9b07 100644
--- a/platform/event-log/src/context/safety.rs
+++ b/platform/event-log/src/context/safety.rs
@@ -276,7 +276,7 @@ pub(crate) async fn model_source_unchanged(
OR (json_extract(barrier.event_json,'$.fact.event.data.provider_options') IS NOT NULL
AND json_extract(barrier.event_json,'$.fact.event.data.provider_options') != '{{}}'))))
SELECT EXISTS(SELECT 1 FROM runtime_events e WHERE e.sequence > ?3
- AND json_extract(e.event_json,'$.invocation.session_id')=?5 AND {filter}
+ AND (e.event_json IS NOT NULL AND e.event_session=?5) AND {filter}
AND NOT (e.invocation_id=?1 AND e.kind IN ('tool_dispatched','tool_settled','tool_rejected')
AND e.operation_id IN (SELECT operation_id FROM independent))
AND NOT (e.invocation_id=?1 AND e.kind='tool_notified'
diff --git a/platform/event-log/src/context/selection.rs b/platform/event-log/src/context/selection.rs
index 4b52fd7000..5b3c9adcea 100644
--- a/platform/event-log/src/context/selection.rs
+++ b/platform/event-log/src/context/selection.rs
@@ -187,7 +187,7 @@ impl Selection {
sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT MAX(
(SELECT COALESCE(MAX(e.sequence),0) FROM runtime_events e WHERE e.sequence <= ?1
- AND json_extract(e.event_json,'$.invocation.session_id')=?2 AND {filter}),
+ AND (e.event_json IS NOT NULL AND e.event_session=?2) AND {filter}),
(SELECT COALESCE(MAX(h.sequence),0) FROM session_history_members h
JOIN runtime_events e ON e.sequence=h.sequence
WHERE h.session_id=?2 AND h.sequence <= ?1 AND {filter}))"
diff --git a/platform/event-log/src/context/usage.rs b/platform/event-log/src/context/usage.rs
index 394c6564ca..67199afb5a 100644
--- a/platform/event-log/src/context/usage.rs
+++ b/platform/event-log/src/context/usage.rs
@@ -161,7 +161,7 @@ async fn additions(
// Legacy interrupted fragments are discarded; marked retained progress
// was handled above. Completed Main output is already in newest usage.
const FILTER: &str = "
- e.sequence > ?1 AND json_extract(e.event_json,'$.invocation.session_id')=?2
+ e.sequence > ?1 AND (e.event_json IS NOT NULL AND e.event_session=?2)
AND e.kind IN ('invocation_opened','message_steered','model_requested',
'model_observed','executor_completed','tool_settled','tool_rejected','tool_notified',
'context_checkpoint_recorded','tool_result_archived')
diff --git a/platform/event-log/src/message_sources.rs b/platform/event-log/src/message_sources.rs
index 9231b2cb99..bea5929681 100644
--- a/platform/event-log/src/message_sources.rs
+++ b/platform/event-log/src/message_sources.rs
@@ -118,13 +118,25 @@ impl EventLog {
let session = session_id.to_owned();
self.connection.read(move |connection| Box::pin(async move {
let row: Option<(i64, String)> = sqlx::query_as(
- "SELECT e.sequence, e.event_json
- FROM runtime_events e WHERE e.kind='invocation_opened'
- AND json_extract(e.event_json,'$.fact.input.kind')='message'
- AND (json_extract(e.event_json,'$.invocation.session_id')=?1
- OR EXISTS(SELECT 1 FROM session_history_members h WHERE h.session_id=?1 AND h.sequence=e.sequence)
- OR EXISTS(SELECT 1 FROM session_revision_sources r WHERE r.session_id=?1 AND r.sequence=e.sequence))
- ORDER BY e.sequence DESC LIMIT 1"
+ "WITH candidates AS (
+ SELECT sequence FROM (
+ SELECT e.sequence FROM runtime_events e
+ WHERE json_extract(e.event_json,'$.invocation.session_id')=?1 AND e.kind='invocation_opened' AND json_extract(e.event_json,'$.fact.input.kind')='message'
+ ORDER BY e.sequence DESC LIMIT 1)
+ UNION ALL
+ SELECT sequence FROM (
+ SELECT h.sequence FROM session_history_members h
+ JOIN runtime_events e ON e.sequence=h.sequence
+ WHERE h.session_id=?1 AND e.kind='invocation_opened' AND json_extract(e.event_json,'$.fact.input.kind')='message'
+ ORDER BY h.sequence DESC LIMIT 1)
+ UNION ALL
+ SELECT sequence FROM (
+ SELECT r.sequence FROM session_revision_sources r
+ JOIN runtime_events e ON e.sequence=r.sequence
+ WHERE r.session_id=?1 AND e.kind='invocation_opened' AND json_extract(e.event_json,'$.fact.input.kind')='message'
+ ORDER BY r.sequence DESC LIMIT 1))
+ SELECT e.sequence,e.event_json FROM runtime_events e
+ WHERE e.sequence=(SELECT MAX(sequence) FROM candidates)"
).bind(session).fetch_optional(connection).await?;
row.map(|(sequence, json)| decode_delivery(sequence, json)).transpose()
})).await
@@ -300,14 +312,25 @@ pub(crate) async fn first_message_opening(
session: &str,
) -> Result<Option<StoredEvent>, StoreError> {
let row: Option<(i64, String, String)> = sqlx::query_as(
- "SELECT e.sequence, e.event_json, e.kind
- FROM runtime_events e WHERE ((e.kind='invocation_opened'
- AND json_extract(e.event_json,'$.fact.input.kind')='message')
- OR (e.kind='message_imported' AND json_extract(e.event_json,'$.fact.record.content.kind')='user'))
- AND (json_extract(e.event_json,'$.invocation.session_id')=?1
- OR EXISTS(SELECT 1 FROM session_history_members h WHERE h.session_id=?1 AND h.sequence=e.sequence)
- OR EXISTS(SELECT 1 FROM session_revision_sources r WHERE r.session_id=?1 AND r.sequence=e.sequence))
- ORDER BY e.sequence LIMIT 1"
+ "WITH candidates AS (
+ SELECT sequence FROM (
+ SELECT e.sequence FROM runtime_events e
+ WHERE e.event_json IS NOT NULL AND e.event_session=?1 AND ((e.kind='invocation_opened' AND json_extract(e.event_json,'$.fact.input.kind')='message') OR (e.kind='message_imported' AND json_extract(e.event_json,'$.fact.record.content.kind')='user'))
+ ORDER BY e.sequence ASC LIMIT 1)
+ UNION ALL
+ SELECT sequence FROM (
+ SELECT h.sequence FROM session_history_members h
+ JOIN runtime_events e ON e.sequence=h.sequence
+ WHERE h.session_id=?1 AND ((e.kind='invocation_opened' AND json_extract(e.event_json,'$.fact.input.kind')='message') OR (e.kind='message_imported' AND json_extract(e.event_json,'$.fact.record.content.kind')='user'))
+ ORDER BY h.sequence ASC LIMIT 1)
+ UNION ALL
+ SELECT sequence FROM (
+ SELECT r.sequence FROM session_revision_sources r
+ JOIN runtime_events e ON e.sequence=r.sequence
+ WHERE r.session_id=?1 AND ((e.kind='invocation_opened' AND json_extract(e.event_json,'$.fact.input.kind')='message') OR (e.kind='message_imported' AND json_extract(e.event_json,'$.fact.record.content.kind')='user'))
+ ORDER BY r.sequence ASC LIMIT 1))
+ SELECT e.sequence,e.event_json,e.kind FROM runtime_events e
+ WHERE e.sequence=(SELECT MIN(sequence) FROM candidates)"
).bind(session).fetch_optional(connection).await?;
match row {
Some((sequence, json, kind)) if kind == "invocation_opened" => {
diff --git a/platform/event-log/src/observation.rs b/platform/event-log/src/observation.rs
index 16fc851a10..e24e22c9bf 100644
--- a/platform/event-log/src/observation.rs
+++ b/platform/event-log/src/observation.rs
@@ -87,7 +87,7 @@ impl EventLog {
self.connection.read(move |connection| Box::pin(async move {
let row: Option<(i64, String)> = sqlx::query_as(
"SELECT sequence, event_json FROM runtime_events WHERE event_id = ?1 AND sequence <= ?2
- AND json_extract(event_json, '$.invocation.session_id') = ?3
+ AND (event_json IS NOT NULL AND event_session=?3)
AND json_extract(event_json, '$.invocation.turn_id') = ?4"
).bind(event_id).bind(through as i64).bind(invocation.session_id).bind(invocation.turn_id)
.fetch_optional(connection).await?;
@@ -110,9 +110,9 @@ impl EventLog {
let rows: Vec<(String, i64, i64, i64)> = sqlx::query_as(
"SELECT session.id, session.revision,
COALESCE((SELECT revision FROM message_queue_state WHERE session_id = session.id), 0),
- COALESCE((SELECT sequence FROM event_log INDEXED BY session_event_sequence
+ COALESCE((SELECT sequence FROM event_log INDEXED BY event_material_owner
WHERE invocation_id IS NOT NULL
- AND json_extract(event_json, '$.invocation.session_id') = session.id
+ AND (event_json IS NOT NULL AND event_session=session.id)
ORDER BY sequence DESC LIMIT 1), 0)
FROM session_control session WHERE session.id IN (SELECT value FROM json_each(?))"
).bind(sessions).fetch_all(connection).await?;
@@ -215,7 +215,7 @@ impl EventLog {
{
let mut rows = sqlx::query(
"SELECT sequence, length(CAST(event_json AS BLOB)), event_json FROM runtime_events
- WHERE json_extract(event_json, '$.invocation.session_id') = ?1
+ WHERE (event_json IS NOT NULL AND event_session=?1)
AND sequence > ?2 AND sequence <= ?3 ORDER BY sequence LIMIT ?4",
)
.bind(&session_id)
diff --git a/platform/event-log/src/observation/delivery.rs b/platform/event-log/src/observation/delivery.rs
index ebda52bdc6..47a27c7710 100644
--- a/platform/event-log/src/observation/delivery.rs
+++ b/platform/event-log/src/observation/delivery.rs
@@ -342,7 +342,7 @@ SELECT json_object(
END)) AS projected, invocation_id AS owner, sequence
FROM runtime_events
WHERE sequence > ?2 AND sequence <= ?3
-AND json_extract(event_json, '$.invocation.session_id') = ?1
+AND (event_json IS NOT NULL AND event_session=?1)
AND (kind IN ('invocation_ended', 'model_requested', 'model_completed', 'model_interrupted', 'context_checkpoint_recorded') OR NOT EXISTS (
SELECT 1 FROM runtime_events opening
WHERE opening.invocation_id = runtime_events.invocation_id
diff --git a/platform/event-log/src/sessions/material.rs b/platform/event-log/src/sessions/material.rs
index 8f2331f60f..e1e41f07b2 100644
--- a/platform/event-log/src/sessions/material.rs
+++ b/platform/event-log/src/sessions/material.rs
@@ -223,3 +223,90 @@ fn retained(event: &RuntimeEvent) -> Value {
};
json!({"id":event.id,"invocation":event.invocation,"recorded_at":event.recorded_at,"fact":fact})
}
+
+#[cfg(test)]
+mod cursor_tests {
+ use super::*;
+ use maka_agent_facts::event::{EventWrite, Invocation, InvocationInput};
+
+ #[tokio::test]
+ async fn observation_cursor_and_pages_exclude_newer_retained_proof() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = EventLog::open(&path).await.unwrap();
+ log.create_session("source", "create", &json!({}), 1)
+ .await
+ .unwrap();
+ let invocation = Invocation {
+ session_id: "source".into(),
+ turn_id: "turn".into(),
+ run_id: "run".into(),
+ invocation_id: "invocation".into(),
+ };
+ let opened = RuntimeEvent::new(
+ invocation.clone(),
+ Fact::InvocationOpened {
+ configuration: None,
+ input: InvocationInput::Message {
+ source_messages: vec![],
+ content: "live original".into(),
+ request_fingerprint: None,
+ },
+ },
+ );
+ let ended = RuntimeEvent::new(
+ invocation,
+ Fact::InvocationEnded {
+ outcome: InvocationOutcome::Completed,
+ },
+ );
+ assert_eq!(
+ log.append_batch(&[
+ EventWrite::plain(opened).unwrap(),
+ EventWrite::plain(ended.clone()).unwrap()
+ ])
+ .await
+ .unwrap(),
+ [1, 2]
+ );
+ // Exercise the query boundary independently of collection's traversal
+ // order: the newest own fact has proof but no live body.
+ let proof = serde_json::to_string(&retained(&ended)).unwrap();
+ let digest = content_digest(serde_json::to_string(&ended).unwrap().as_bytes());
+ log.connection.run(move |db| Box::pin(async move {
+ sqlx::query("UPDATE event_log SET retained_json=?,body_digest=?,event_json=NULL WHERE sequence=2")
+ .bind(proof).bind(digest).execute(db).await?;
+ Ok(())
+ })).await.unwrap();
+ for _ in 0..2 {
+ let versions = log
+ .observation_versions(&["source".into(), "missing".into()])
+ .await
+ .unwrap();
+ assert_eq!(versions.len(), 1);
+ assert_eq!(versions["source"].event, 1);
+ let page = log.session_events("source", 0, 2, 64, 65536).await.unwrap();
+ assert_eq!(
+ page.events.iter().map(|e| e.sequence).collect::<Vec<_>>(),
+ [1]
+ );
+ assert_eq!(page.through_sequence, 2);
+ assert!(page.next_after.is_none());
+ }
+ log.close().await.unwrap();
+ let log = EventLog::open(&path).await.unwrap();
+ assert_eq!(
+ log.observation_versions(&["source".into()]).await.unwrap()["source"].event,
+ 1
+ );
+ assert_eq!(
+ log.session_events("source", 1, 2, 64, 65536)
+ .await
+ .unwrap()
+ .events
+ .len(),
+ 0
+ );
+ log.close().await.unwrap();
+ }
+}
diff --git a/platform/event-log/src/transcript/navigation/queries.rs b/platform/event-log/src/transcript/navigation/queries.rs
index 038c4aebdc..9c6b166fbb 100644
--- a/platform/event-log/src/transcript/navigation/queries.rs
+++ b/platform/event-log/src/transcript/navigation/queries.rs
@@ -21,7 +21,7 @@ pub(super) const FENCE: &str = "
SELECT MAX(sequence) FROM (
SELECT MAX(event.sequence) AS sequence FROM runtime_events event
JOIN runtime_events opening ON opening.invocation_id = event.invocation_id AND opening.kind = 'invocation_opened'
- WHERE json_extract(event.event_json, '$.invocation.session_id') = ?1
+ WHERE (event.event_json IS NOT NULL AND event.event_session=?1)
AND json_extract(opening.event_json, '$.fact.input.kind') IN ('message', 'continuation', 'handoff')
UNION ALL
SELECT MAX(h.sequence) FROM session_history_members h
@@ -38,13 +38,20 @@ AND (row.source_sequence,row.item_index)>=(COALESCE(?3,1),COALESCE(?4,0))
ORDER BY row.source_sequence,row.item_index LIMIT 257";
pub(super) const LANDMARKS: &str = "
-WITH openings AS (
- SELECT opening.invocation_id, opening.sequence,
- ROW_NUMBER() OVER (PARTITION BY json_extract(opening.event_json, '$.invocation.turn_id') ORDER BY opening.sequence) AS first
- FROM session_history_events opening
- WHERE opening.owner_session_id = ?1
- AND opening.kind = 'invocation_opened' AND opening.sequence <= ?2
- AND json_extract(opening.event_json, '$.fact.input.kind') IN ('message', 'continuation')
+WITH sources AS (
+ SELECT opening.invocation_id,opening.sequence,json_extract(opening.event_json,'$.invocation.turn_id') AS turn_id
+ FROM runtime_events opening
+ WHERE json_extract(opening.event_json,'$.invocation.session_id')=?1
+ AND opening.kind='invocation_opened' AND opening.sequence<=?2
+ AND json_extract(opening.event_json,'$.fact.input.kind') IN ('message','continuation')
+ UNION ALL
+ SELECT opening.invocation_id,opening.sequence,json_extract(opening.event_json,'$.invocation.turn_id')
+ FROM session_history_members h JOIN runtime_events opening ON opening.sequence=h.sequence
+ WHERE h.session_id=?1 AND h.sequence<=?2 AND opening.kind='invocation_opened'
+ AND json_extract(opening.event_json,'$.fact.input.kind') IN ('message','continuation')
+), openings AS (
+ SELECT invocation_id,sequence,ROW_NUMBER() OVER (PARTITION BY turn_id ORDER BY sequence) AS first
+ FROM sources
), candidates AS (
SELECT invocation_id, sequence, ROW_NUMBER() OVER (ORDER BY sequence) - 1 AS rank, COUNT(*) OVER () AS total
FROM openings WHERE first = 1
diff --git a/platform/event-log/src/workspace.rs b/platform/event-log/src/workspace.rs
index 37a437b6f9..9018770676 100644
--- a/platform/event-log/src/workspace.rs
+++ b/platform/event-log/src/workspace.rs
@@ -68,7 +68,7 @@ impl EventLog {
let rows = sqlx::query(
"SELECT s.id, s.revision,
COALESCE((SELECT MAX(e.sequence) FROM runtime_events e WHERE
- json_extract(e.event_json, '$.invocation.session_id') = s.id), 0) AS sequence,
+ (e.event_json IS NOT NULL AND e.event_session=s.id)), 0) AS sequence,
EXISTS(SELECT 1 FROM shell_runs r WHERE r.session_id = s.id AND
json_extract(r.record_json, '$.state.outcome.kind') = 'orphaned') AS orphaned
FROM session_control s WHERE COALESCE(json_extract(s.configuration, '$.workspace.path'), json_extract(s.configuration, '$.workspace.hostCwd')) = ?
diff --git a/platform/event-log/tests/migrations/material_session.rs b/platform/event-log/tests/migrations/material_session.rs
new file mode 100644
index 0000000000..dc119afdd6
--- /dev/null
+++ b/platform/event-log/tests/migrations/material_session.rs
@@ -0,0 +1,236 @@
+/*
+ * 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 crate::sessions::{MaterialCollection, SessionRetirement};
+use maka_agent_facts::event::LogScope;
+
+fn snapshot(db: &Connection) -> Vec<Vec<String>> {
+ [
+ "SELECT json_array(sequence,event_id,invocation_id,kind,operation_id,event_json,retained_json,body_digest) FROM event_log ORDER BY sequence",
+ "SELECT json_array(owner_session_id,inherited,archive_sequence,sequence,event_id,event_json) FROM session_history_events ORDER BY owner_session_id,sequence,inherited",
+ "SELECT json_array(session_id,sequence,archive_sequence) FROM session_history_members ORDER BY session_id,sequence",
+ "SELECT json_array(session_id,sequence) FROM session_revision_sources ORDER BY session_id,sequence",
+ "SELECT json_array(version,description,success,hex(checksum)) FROM _sqlx_migrations WHERE version<=38 ORDER BY version",
+ ].into_iter().map(|sql| db.prepare(sql).unwrap().query_map([], |row| row.get(0)).unwrap()
+ .collect::<Result<Vec<String>, _>>().unwrap()).collect()
+}
+
+fn indexes(db: &Connection) -> std::collections::BTreeSet<String> {
+ db.prepare("SELECT name FROM sqlite_schema WHERE type='index'")
+ .unwrap()
+ .query_map([], |row| row.get(0))
+ .unwrap()
+ .collect::<Result<_, _>>()
+ .unwrap()
+}
+
+async fn retire(log: &EventLog, session: &str) {
+ let revision = log
+ .get_session::<Value>(session)
+ .await
+ .unwrap()
+ .unwrap()
+ .revision;
+ log.begin_session_removal(session, revision).await.unwrap();
+ assert_eq!(
+ log.finish_session_retirement(session).await.unwrap(),
+ SessionRetirement::Removed
+ );
+}
+
+async fn collect(log: &EventLog) {
+ for _ in 0..100 {
+ match log.collect_session_material(None).await.unwrap() {
+ MaterialCollection::Done => return,
+ MaterialCollection::Collected => {}
+ other => panic!("unexpected retained material: {other:?}"),
+ }
+ }
+ panic!("material collection failed to converge");
+}
+
+#[tokio::test]
+async fn migration_39_preserves_live_collected_and_copied_evidence_and_original_ledger() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("events.sqlite");
+ let log = open_at::<38>(&path).await;
+ log.connection
+ .run(|db| Box::pin(async move { crate::transcript::initialize_cache(db).await }))
+ .await
+ .unwrap();
+ seed_canonical_image(&log).await;
+ log.create_session("collected", "create-collected", &json!({}), 30)
+ .await
+ .unwrap();
+ let mut original = invocation();
+ original.session_id = "collected".into();
+ original.turn_id = "collected-turn".into();
+ original.run_id = "collected-run".into();
+ original.invocation_id = "collected-invocation".into();
+ for fact in [
+ Fact::InvocationOpened {
+ configuration: None,
+ input: InvocationInput::Message {
+ source_messages: vec![],
+ content: "private collected body".into(),
+ request_fingerprint: None,
+ },
+ },
+ Fact::InvocationEnded {
+ outcome: InvocationOutcome::Completed,
+ },
+ ] {
+ log.append(&EventWrite::plain(RuntimeEvent::new(original.clone(), fact)).unwrap())
+ .await
+ .unwrap();
+ }
+ retire(&log, "collected").await;
+ collect(&log).await;
+ let scoped = |id: &str| LogScope::Session { id: id.into() };
+ let native = log
+ .scoped_prefix(scoped("original"), 64, 1_000_000)
+ .await
+ .unwrap()
+ .digest;
+ let inherited = log
+ .scoped_prefix(scoped("copied"), 64, 1_000_000)
+ .await
+ .unwrap()
+ .digest;
+ assert_eq!(log.navigation_fence("original").await.unwrap(), Some(4));
+ assert_eq!(log.navigation_fence("copied").await.unwrap(), Some(4));
+ let versions = log
+ .observation_versions(&["original".into(), "copied".into()])
+ .await
+ .unwrap();
+ assert_eq!(versions["original"].event, 4);
+ assert_eq!(versions["copied"].event, 0);
+ log.close().await.unwrap();
+ let db = Connection::open(&path).unwrap();
+ let before = snapshot(&db);
+ let mut expected_indexes = indexes(&db);
+ assert!(expected_indexes.remove("session_event_sequence"));
+ assert_eq!(
+ db.query_row("SELECT max(version) FROM _sqlx_migrations", [], |r| r
+ .get::<_, i64>(0))
+ .unwrap(),
+ 38
+ );
+ assert_eq!(
+ db.query_row(
+ "SELECT count(*) FROM event_log WHERE retained_json IS NOT NULL",
+ [],
+ |r| r.get::<_, i64>(0)
+ )
+ .unwrap(),
+ 2
+ );
+ drop(db);
+ for _ in 0..2 {
+ let log = EventLog::open(&path).await.unwrap();
+ assert_eq!(
+ log.scoped_prefix(scoped("original"), 64, 1_000_000)
+ .await
+ .unwrap()
+ .digest,
+ native
+ );
+ assert_eq!(
+ log.scoped_prefix(scoped("copied"), 64, 1_000_000)
+ .await
+ .unwrap()
+ .digest,
+ inherited
+ );
+ let actual = log
+ .observation_versions(&["original".into(), "copied".into(), "missing".into()])
+ .await
+ .unwrap();
+ assert_eq!(actual["original"].event, 4);
+ assert_eq!(actual["copied"].event, 0);
+ assert_eq!(actual.len(), 2);
+ assert_eq!(log.navigation_fence("original").await.unwrap(), Some(4));
+ assert_eq!(log.navigation_fence("copied").await.unwrap(), Some(4));
+ assert!(
+ log.session_events("copied", 0, 6, 64, 1_000_000)
+ .await
+ .unwrap()
+ .events
+ .is_empty()
+ );
+ let own = log
+ .session_events("original", 0, 6, 64, 1_000_000)
+ .await
+ .unwrap();
+ assert_eq!(
+ own.events
+ .iter()
+ .map(|event| event.sequence)
+ .collect::<Vec<_>>(),
+ [1, 2, 3, 4]
+ );
+ let db = Connection::open(&path).unwrap();
+ assert_eq!(snapshot(&db), before);
+ assert_eq!(indexes(&db), expected_indexes);
+ assert_eq!(
+ db.query_row("SELECT max(version) FROM _sqlx_migrations", [], |r| r
+ .get::<_, i64>(0))
+ .unwrap(),
+ 39
+ );
+ drop(db);
+ log.close().await.unwrap();
+ }
+ let log = EventLog::open(&path).await.unwrap();
+ retire(&log, "original").await;
+ assert_eq!(
+ log.collect_session_material(None).await.unwrap(),
+ MaterialCollection::Retained("original".into())
+ );
+ assert_eq!(
+ log.scoped_prefix(scoped("copied"), 64, 1_000_000)
+ .await
+ .unwrap()
+ .digest,
+ inherited
+ );
+ retire(&log, "copied").await;
+ collect(&log).await;
+ log.close().await.unwrap();
+ let db = Connection::open(&path).unwrap();
+ assert_eq!(
+ db.query_row(
+ "SELECT count(*) FROM event_log WHERE event_json IS NOT NULL",
+ [],
+ |r| r.get::<_, i64>(0)
+ )
+ .unwrap(),
+ 0
+ );
+ assert_eq!(
+ db.query_row(
+ "SELECT count(*) FROM event_log WHERE retained_json IS NOT NULL",
+ [],
+ |r| r.get::<_, i64>(0)
+ )
+ .unwrap(),
+ 6
+ );
+}
diff --git a/platform/event-log/tests/migrations/prior_artifacts.rs b/platform/event-log/tests/migrations/prior_artifacts.rs
index 13f653b4b2..af1a0803b9 100644
--- a/platform/event-log/tests/migrations/prior_artifacts.rs
+++ b/platform/event-log/tests/migrations/prior_artifacts.rs
@@ -38,8 +38,9 @@ use sqlx::sqlite::SqliteConnectOptions;
// Run the real historical migrations, then use the real current writer APIs.
// No migration ledger entry or checksum is removed to manufacture an old schema.
async fn open_at<const VERSION: i64>(path: &Path) -> EventLog {
- let lease =
- Arc::new(crate::root::FileLease::acquire(&path.with_extension("writer.lock")).unwrap());
+ let mut lock_path = path.as_os_str().to_os_string();
+ lock_path.push(".writer.lock");
+ let lease = Arc::new(crate::root::FileLease::acquire(Path::new(&lock_path)).unwrap());
let connection = OwnedConnection::open(
SqliteConnectOptions::new()
.filename(path)
@@ -48,7 +49,7 @@ async fn open_at<const VERSION: i64>(path: &Path) -> EventLog {
move |connection| {
Box::pin(async move {
sqlx::raw_sql(
- "PRAGMA application_id = 1296124754; PRAGMA auto_vacuum = INCREMENTAL;",
+ "PRAGMA auto_vacuum = INCREMENTAL; PRAGMA application_id = 1296124754;",
)
.execute(&mut *connection)
.await?;
@@ -479,7 +480,7 @@ async fn migration_37_preserves_interaction_receipts_and_relations_without_json_
0
))
.unwrap(),
- 38
+ 39
);
for ((table, columns), original) in tables.iter().zip(original) {
assert_eq!(rows(&db, table, columns), original);
@@ -718,7 +719,7 @@ async fn migration_37_preserves_artifact_metadata_and_copied_image_provenance()
0
))
.unwrap(),
- 38
+ 39
);
assert_eq!(artifacts(&db), prior);
assert_eq!(mappings(&db), provenance);
@@ -788,3 +789,6 @@ async fn migration_37_preserves_artifact_metadata_and_copied_image_provenance()
drop(db);
log.close().await.unwrap();
}
+
+#[path = "material_session.rs"]
+mod material_session;
diff --git a/platform/event-log/tests/session_copy/mod.rs b/platform/event-log/tests/session_copy/mod.rs
index 185b2d47fb..13b119624f 100644
--- a/platform/event-log/tests/session_copy/mod.rs
+++ b/platform/event-log/tests/session_copy/mod.rs
@@ -589,6 +589,26 @@ async fn copies_own_history_and_files_without_replaying_execution_across_retries
);
assert_eq!(copy(&log, request).await, first);
for session in ["branch", "nested"] {
+ assert_eq!(
+ log.first_message_opening(session)
+ .await
+ .unwrap()
+ .unwrap()
+ .event
+ .invocation
+ .turn_id,
+ "first"
+ );
+ assert_eq!(
+ log.latest_message_opening(session)
+ .await
+ .unwrap()
+ .unwrap()
+ .event
+ .invocation
+ .turn_id,
+ "first"
+ );
let prefix = log
.scoped_prefix(LogScope::Session { id: session.into() }, 100, 65536)
.await
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org