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/19 17:10:02 UTC
(maka) branch rust-preview updated: fix(event-log): bound bundle validation cursor work
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 da94c0df6b fix(event-log): bound bundle validation cursor work
da94c0df6b is described below
commit da94c0df6b330b79d2eb8a706910ae53dcb182f0
Author: Wang <m4...@gmail.com>
AuthorDate: Tue Oct 20 01:09:55 2026 +0800
fix(event-log): bound bundle validation cursor work
Use the staged source-order index and merge the first eligible owned,
inherited and revision event at each Artifact-validation cursor. Preserve
original and relocated causal replay and cover revision attachment
authority through import and reopen.
Generated-by: Codex
---
platform/event-log/src/bundle/validation.rs | 2 +-
.../event-log/src/bundle/validation/artifacts.rs | 20 +-
platform/event-log/tests/bundle/image.rs | 204 +++++++++++++++++++++
3 files changed, 221 insertions(+), 5 deletions(-)
diff --git a/platform/event-log/src/bundle/validation.rs b/platform/event-log/src/bundle/validation.rs
index e961f48164..7ede24ecb5 100644
--- a/platform/event-log/src/bundle/validation.rs
+++ b/platform/event-log/src/bundle/validation.rs
@@ -91,7 +91,7 @@ async fn replay(
let mut after = 0i64;
loop {
let row: Option<String> = sqlx::query_scalar(
- "SELECT record_json FROM frames WHERE kind='event' AND source_sequence>? ORDER BY source_sequence LIMIT 1"
+ "SELECT record_json FROM frames INDEXED BY source_events WHERE kind='event' AND source_sequence>? ORDER BY source_sequence LIMIT 1"
).bind(after).fetch_optional(&mut *staged).await?;
let Some(row) = row else {
break;
diff --git a/platform/event-log/src/bundle/validation/artifacts.rs b/platform/event-log/src/bundle/validation/artifacts.rs
index bec7b73a5c..7f3db6e02b 100644
--- a/platform/event-log/src/bundle/validation/artifacts.rs
+++ b/platform/event-log/src/bundle/validation/artifacts.rs
@@ -94,12 +94,24 @@ pub(super) async fn validate(
for session in &inventory.sessions {
let mut after = (0i64, false);
loop {
+ // Bound each ownership branch before merging so every cursor step
+ // sorts at most three candidates instead of the remaining history.
let row: Option<(i64, bool, bool, String)> = sqlx::query_as(
- "SELECT e.sequence,0 AS revision,e.inherited,e.event_json FROM session_history_events e
- WHERE e.owner_session_id=?1 AND (e.sequence,0)>(?2,?3)
+ "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
+ ORDER BY e.sequence LIMIT 1)
UNION ALL
- SELECT e.sequence,1 AS revision,1 AS inherited,e.event_json FROM session_revision_sources r
- JOIN runtime_events e ON e.sequence=r.sequence WHERE r.session_id=?1 AND (e.sequence,1)>(?2,?3)
+ SELECT sequence,revision,inherited,event_json FROM (
+ SELECT e.sequence,0 AS revision,1 AS inherited,e.event_json FROM session_history_members h
+ JOIN runtime_events e ON e.sequence=h.sequence WHERE h.session_id=?1 AND h.sequence>?2
+ ORDER BY h.sequence LIMIT 1)
+ UNION ALL
+ SELECT sequence,revision,inherited,event_json FROM (
+ SELECT r.sequence,1 AS revision,1 AS inherited,e.event_json FROM session_revision_sources r
+ JOIN runtime_events e ON e.sequence=r.sequence
+ WHERE r.session_id=?1 AND r.sequence>=?2 AND (r.sequence>?2 OR ?3=0)
+ ORDER BY r.sequence LIMIT 1)
ORDER BY sequence,revision LIMIT 1"
).bind(&session.id).bind(after.0).bind(after.1).fetch_optional(&mut *original).await?;
let Some((sequence, revision, inherited, json)) = row else {
diff --git a/platform/event-log/tests/bundle/image.rs b/platform/event-log/tests/bundle/image.rs
index a8450475e5..ee886beaaa 100644
--- a/platform/event-log/tests/bundle/image.rs
+++ b/platform/event-log/tests/bundle/image.rs
@@ -381,6 +381,210 @@ async fn copied_image_ownership_survives_transfer_without_importing_parent_catal
.is_err()
);
destination.close().await.unwrap();
+ // Revision material belongs to the excluded editable input, even when
+ // preparation replaced its attachment with a different upload.
+ use maka_agent_facts::attachment::{AttachmentKind, AttachmentRef};
+ use maka_agent_facts::input::{DeliveredMessage, MessageInput};
+ use maka_agent_facts::message::{MessageDisposition, Placement, RootSourceMessage};
+ upload(&source, "editable-only").await;
+ upload(&source, "prepared-only").await;
+ let attachment = |id: &str| AttachmentRef {
+ kind: AttachmentKind::Image,
+ name: "image.gif".into(),
+ mime_type: "image/gif".into(),
+ bytes: GIF.len() as u64,
+ storage_ref: StorageRef::SessionFile {
+ session_id: "images".into(),
+ relative_path: id.into(),
+ },
+ };
+ let mut unprepared = MessageInput::from("original editable image");
+ unprepared.attachments = Some(vec![attachment("editable-only")]);
+ let mut prepared = MessageInput::from("prepared image");
+ prepared.attachments = Some(vec![attachment("prepared-only")]);
+ let editable = invocation("images", "editable-run");
+ let opening = RuntimeEvent::new(
+ editable.clone(),
+ Fact::InvocationOpened {
+ configuration: None,
+ input: InvocationInput::Message {
+ content: prepared.clone(),
+ source_messages: vec![RootSourceMessage {
+ message: DeliveredMessage {
+ origin: None,
+ message_id: "editable-message".into(),
+ content: prepared,
+ submitted_content_digest: unprepared.content_digest().unwrap(),
+ },
+ unprepared_content: unprepared,
+ submitted_placement: Placement::CurrentTurn,
+ disposition: MessageDisposition::TurnStarted,
+ submitted_intent: None,
+ }],
+ request_fingerprint: None,
+ },
+ },
+ );
+ let source_sequence = source
+ .append(&EventWrite::plain(opening.clone()).unwrap())
+ .await
+ .unwrap();
+ end(&source, &editable).await;
+ let revision = source
+ .get_session::<Value>("images")
+ .await
+ .unwrap()
+ .unwrap()
+ .revision;
+ source
+ .copy_session(
+ SessionCopy {
+ source_session_id: "images".into(),
+ target_session_id: "revision".into(),
+ expected_source_revision: revision,
+ purpose: CopyPurpose::Revision {
+ turn_id: editable.turn_id.clone(),
+ },
+ },
+ &Value::Null,
+ 2,
+ )
+ .await
+ .unwrap();
+ let bytes = export(&source, "revision").await;
+ let summary = bundle::inspect(bytes.as_slice()).await.unwrap();
+ let records = super::frames::records(&bytes, &summary.digest);
+ let revision_sources: Vec<_> = records
+ .iter()
+ .filter(|r| r["kind"] == "revision_source" && r["session"] == "revision")
+ .collect();
+ assert_eq!(revision_sources.len(), 1);
+ assert_eq!(revision_sources[0]["sequence"], source_sequence);
+ assert!(!records.iter().any(|r| r["kind"] == "member"
+ && r["session"] == "revision"
+ && r["sequence"] == source_sequence));
+ let mut retained: Vec<_> = records
+ .iter()
+ .filter(|r| r["kind"] == "history_artifact" && r["session"] == "revision")
+ .map(|r| r["source_artifact"].as_str().unwrap())
+ .collect();
+ retained.sort_unstable();
+ assert_eq!(retained, ["borrowed", "editable-only"]);
+ let artifacts: Vec<_> = records
+ .iter()
+ .filter(|r| r["kind"] == "blob" && r["resource"] == "artifact")
+ .collect();
+ assert_eq!(artifacts.len(), 2);
+ assert!(
+ artifacts
+ .iter()
+ .all(|r| r["metadata"]["sessionId"] == "revision")
+ );
+ let mut staged = StagedBundle::read(bytes.as_slice()).await.unwrap();
+ staged.validate_history().await.unwrap();
+ staged.close().await.unwrap();
+
+ let mut removed = 0;
+ let missing_mapping = super::frames::rewrite_records(&bytes, |record| {
+ if record["kind"] == "history_artifact"
+ && record["session"] == "revision"
+ && record["source_artifact"] == "editable-only"
+ {
+ removed += 1;
+ return false;
+ }
+ true
+ });
+ assert_eq!(removed, 1);
+ let mut staged = StagedBundle::read(missing_mapping.as_slice())
+ .await
+ .unwrap();
+ assert!(matches!(
+ staged.validate_history().await,
+ Err(bundle::BundleError::Store(maka_event_log::StoreError::InvalidTransition(reason)))
+ if reason == "bundle copied reference lacks its retained Artifact"
+ ));
+ staged.close().await.unwrap();
+
+ let path = dir.path().join("revision-destination.sqlite");
+ let mut destination = EventLog::open(&path).await.unwrap();
+ padding(&destination, "revision-padding", 9).await;
+ import(&destination, &bytes).await;
+ for reopened in [false, true] {
+ let prefix = destination.prefix(1000, 8 * 1024 * 1024).await.unwrap();
+ let relocated = prefix
+ .events
+ .iter()
+ .find(|e| e.event.id == opening.id)
+ .unwrap();
+ assert_ne!(relocated.sequence, source_sequence);
+ assert_eq!(relocated.event, opening);
+ let input = destination
+ .editable_message("revision", "editable-run", "editable-message")
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(input.content.text, "original editable image");
+ let attachments = input.content.attachments.as_ref().unwrap();
+ assert_eq!(attachments.len(), 1);
+ let StorageRef::SessionFile {
+ session_id,
+ relative_path,
+ } = &attachments[0].storage_ref
+ else {
+ panic!("editable input lost its retained upload");
+ };
+ assert_eq!(session_id, "revision");
+ assert_eq!(
+ destination
+ .read_artifact_chunk("revision", relative_path, 0, 32768)
+ .await
+ .unwrap()
+ .unwrap()
+ .bytes,
+ GIF
+ );
+ assert!(
+ destination
+ .root_message("revision", "editable-message")
+ .await
+ .unwrap()
+ .is_none()
+ );
+ assert!(
+ destination
+ .get_session::<Value>("images")
+ .await
+ .unwrap()
+ .is_none()
+ );
+ assert!(matches!(
+ destination
+ .read_artifact_chunk("images", "editable-only", 0, 32768)
+ .await,
+ Err(maka_event_log::StoreError::SessionNotFound)
+ ));
+ let (through, rows) = self::rows(&destination, "revision", &i).await;
+ assert_eq!(
+ destination
+ .tool_image("revision", through, rows["borrowed"], 0, 32768)
+ .await
+ .unwrap()
+ .bytes,
+ GIF
+ );
+ let turns = destination
+ .navigation_turns("revision", through, None, 32)
+ .await
+ .unwrap();
+ assert_eq!(turns.contributions.len(), 1);
+ assert_eq!(turns.contributions[0].turn_id, "copy-run");
+ if !reopened {
+ destination.close().await.unwrap();
+ destination = EventLog::open(&path).await.unwrap();
+ }
+ }
+ destination.close().await.unwrap();
source.close().await.unwrap();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org