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/09/23 00:32:09 UTC
(maka) branch feat/runtime-host-rust updated: feat(session): read copied history through owned resources
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch feat/runtime-host-rust
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/feat/runtime-host-rust by this push:
new d05950ec18 feat(session): read copied history through owned resources
d05950ec18 is described below
commit d05950ec1873ee85fbc690abf80a6208b6c72a4f
Author: Wang <m4...@gmail.com>
AuthorDate: Wed Sep 23 08:31:56 2026 +0800
feat(session): read copied history through owned resources
Preserve original execution identities while projecting inherited model input, transcript navigation and archive reads through the destination history ownership.
Generated-by: Codex
---
crates/agent/src/history.rs | 19 ++-
crates/agent/src/history/images.rs | 22 ++--
crates/agent/src/history/projection.rs | 13 +-
crates/agent/src/history/references.rs | 22 ++--
crates/agent/src/history/resources.rs | 127 +++++++++++++++++++
crates/agent/src/history/user.rs | 3 +-
crates/agent/tests/integration/image_projection.rs | 41 +++++-
crates/event-log/src/archive/accepted.rs | 137 ++++++++++++++++++---
crates/event-log/src/artifacts/history.rs | 50 ++++++++
crates/event-log/src/transcript.rs | 42 ++++++-
.../event-log/src/transcript/navigation/queries.rs | 19 ++-
crates/event-log/tests/archive.rs | 32 +++++
crates/event-log/tests/session_copy.rs | 62 ++++++++++
13 files changed, 529 insertions(+), 60 deletions(-)
diff --git a/crates/agent/src/history.rs b/crates/agent/src/history.rs
index 4cfe885ada..65f7f4ef91 100644
--- a/crates/agent/src/history.rs
+++ b/crates/agent/src/history.rs
@@ -54,6 +54,7 @@ impl<'a> From<&'a maka_event_log::context::ContextEvent> for EventRef<'a> {
}
}
mod references;
+mod resources;
pub(super) use images::materialize;
pub(super) use images::materialize_replay;
use projection::build;
@@ -65,6 +66,16 @@ pub fn operation_id(step_id: &str, call_id: &str) -> String {
/// Only committed semantic facts produce provider input. No UI transcript or
/// previous isolate state is read, including after process restart.
pub fn project(prefix: &LogPrefix, session: &str) -> Result<Vec<Message>, RunError> {
+ if !matches!(prefix.scope, maka_runtime::event::LogScope::Root)
+ && prefix
+ .events
+ .iter()
+ .any(|e| e.event.invocation.session_id != session)
+ {
+ return Err(RunError::ReconciliationRequired(
+ "inherited history requires a checked effective context source".into(),
+ ));
+ }
if prefix.events.iter().any(|stored| {
matches!(
stored.event.fact,
@@ -76,8 +87,12 @@ pub fn project(prefix: &LogPrefix, session: &str) -> Result<Vec<Message>, RunErr
));
}
build(
- prefix.events.iter().map(EventRef::Canonical),
- session,
+ prefix
+ .events
+ .iter()
+ .filter(|e| e.event.invocation.session_id == session)
+ .map(EventRef::Canonical),
+ &resources::Resources::native(session),
&mut Vec::new(),
false,
None,
diff --git a/crates/agent/src/history/images.rs b/crates/agent/src/history/images.rs
index 6a42c4b81c..2b81c8d168 100644
--- a/crates/agent/src/history/images.rs
+++ b/crates/agent/src/history/images.rs
@@ -75,16 +75,13 @@ pub(crate) async fn materialize_replay(
replay: Option<super::Replay<'_>>,
) -> Result<Vec<Message>, RunError> {
let mut targets = Vec::new();
- let mut messages = super::build(
- anchor
- .into_iter()
- .map(super::EventRef::Canonical)
- .chain(events.iter().map(super::EventRef::from)),
- session,
- &mut targets,
- vision,
- replay,
- )?;
+ let selected = anchor
+ .into_iter()
+ .map(super::EventRef::Canonical)
+ .chain(events.iter().map(super::EventRef::from));
+ let resources =
+ super::resources::Resources::load(log, session, selected.clone(), cancellation).await?;
+ let mut messages = super::build(selected, &resources, &mut targets, vision, replay)?;
let mut remaining = IMAGE_BUDGET;
let mut omitted = BTreeMap::<usize, usize>::new();
for target in targets {
@@ -95,7 +92,10 @@ pub(crate) async fn materialize_replay(
Target::User { image, .. } => (&image.storage_ref, &image.mime_type),
Target::Tool { image, .. } => (&image.reference, &image.mime_type),
};
- let read = read(log, session, reference, remaining).await?;
+ let read = match resources.resolve(reference) {
+ Some(reference) => read(log, session, reference, remaining).await?,
+ None => ImageRead::Unavailable("session_mismatch"),
+ };
if cancellation.is_cancelled() {
return Err(RunError::Cancelled);
}
diff --git a/crates/agent/src/history/projection.rs b/crates/agent/src/history/projection.rs
index ecfff4f7fa..51c6ed0b20 100644
--- a/crates/agent/src/history/projection.rs
+++ b/crates/agent/src/history/projection.rs
@@ -29,7 +29,7 @@ use std::collections::{HashMap, HashSet};
pub(super) fn build<'a>(
events: impl Iterator<Item = EventRef<'a>> + Clone,
- session: &str,
+ resources: &super::resources::Resources<'_>,
images: &mut Vec<images::Target<'a>>,
vision: bool,
replay: Option<super::Replay<'a>>,
@@ -50,9 +50,6 @@ pub(super) fn build<'a>(
.collect();
let mut purposes = HashMap::new();
for stored in events.clone().filter_map(EventRef::canonical) {
- if stored.event.invocation.session_id != session {
- continue;
- }
if let Fact::ModelRequested {
step_id, purpose, ..
} = &stored.event.fact
@@ -85,9 +82,6 @@ pub(super) fn build<'a>(
let stored = match event {
EventRef::Canonical(stored) => stored,
EventRef::Archived(archived) => {
- if archived.invocation.session_id != session {
- continue;
- }
let (id, name) = calls.remove(&archived.operation_id).ok_or_else(|| {
RunError::ReconciliationRequired("archived result lacks provider call".into())
})?;
@@ -112,9 +106,6 @@ pub(super) fn build<'a>(
continue;
}
};
- if stored.event.invocation.session_id != session {
- continue;
- }
if compact_invocations.contains(&stored.event.invocation.invocation_id) {
continue;
}
@@ -130,6 +121,7 @@ pub(super) fn build<'a>(
};
messages.push(user::project(
content,
+ resources,
false,
messages.len(),
images,
@@ -142,6 +134,7 @@ pub(super) fn build<'a>(
.map_err(|reason| RunError::ReconciliationRequired(reason.into()))?;
messages.push(user::project(
&message.content,
+ resources,
true,
messages.len(),
images,
diff --git a/crates/agent/src/history/references.rs b/crates/agent/src/history/references.rs
index 10e1da82d6..500d9ba714 100644
--- a/crates/agent/src/history/references.rs
+++ b/crates/agent/src/history/references.rs
@@ -24,7 +24,10 @@ use serde_json::json;
/// The log retains references structurally. Only provider input folds them into
/// text; UI chips and invocation fingerprints never use this derived string.
-pub(super) fn text(content: &MessageInput) -> Result<String, RunError> {
+pub(super) fn text(
+ content: &MessageInput,
+ resources: &super::resources::Resources<'_>,
+) -> Result<String, RunError> {
let mut text = content.text.clone();
if let Some(quotes) = &content.quotes
&& !quotes.is_empty()
@@ -65,7 +68,10 @@ pub(super) fn text(content: &MessageInput) -> Result<String, RunError> {
if index != 0 {
text.push('\n');
}
- text.push_str(&attachment_text(attachment));
+ text.push_str(&attachment_text(
+ attachment,
+ resources.resolve(&attachment.storage_ref),
+ ));
}
}
if let Some(references) = &content.directory_references
@@ -88,29 +94,29 @@ pub(super) fn text(content: &MessageInput) -> Result<String, RunError> {
Ok(text)
}
-fn attachment_text(attachment: &AttachmentRef) -> String {
+fn attachment_text(attachment: &AttachmentRef, reference: Option<&StorageRef>) -> String {
let mut text = String::from("<attachment>\n");
- if let Some(resource) = attachment.storage_ref.resource_ref() {
+ if let Some(resource) = reference.and_then(StorageRef::resource_ref) {
text.push_str(&format!("Read argument: {}\n", json!({"path":resource})));
if attachment.kind == AttachmentKind::Image {
text.push_str(&format!("Markdown image source: {}\n", json!(resource)));
}
text.push_str("This is a Session resource, not a workspace file. Use the path above; never use the display name as a path.\n");
} else {
- match &attachment.storage_ref {
- StorageRef::WorkspaceFile { relative_path } => {
+ match reference {
+ Some(StorageRef::WorkspaceFile { relative_path }) => {
text.push_str(&format!(
"Read argument: {}\n",
json!({"path":relative_path})
));
}
- StorageRef::ExternalFile { absolute_path } => {
+ Some(StorageRef::ExternalFile { absolute_path }) => {
text.push_str(&format!(
"Read argument: {}\n",
json!({"path":absolute_path})
));
}
- StorageRef::SessionFile { .. } | StorageRef::SessionContext { .. } => {
+ Some(StorageRef::SessionFile { .. } | StorageRef::SessionContext { .. }) | None => {
text.push_str("The attachment content is unavailable to Read.\n")
}
}
diff --git a/crates/agent/src/history/resources.rs b/crates/agent/src/history/resources.rs
new file mode 100644
index 0000000000..aec2350c15
--- /dev/null
+++ b/crates/agent/src/history/resources.rs
@@ -0,0 +1,127 @@
+/*
+ * 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::EventRef;
+use crate::RunError;
+use maka_event_log::EventLog;
+use maka_runtime::{
+ attachment::StorageRef,
+ event::{Fact, InvocationInput, ToolOutcome},
+ tool_output::{DurableToolProjection, ProjectionPart},
+};
+use std::collections::HashMap;
+use tokio_util::sync::CancellationToken;
+
+/// Request-local rendering aliases. Canonical references and evidence stay intact.
+pub(super) struct Resources<'a> {
+ session: &'a str,
+ inherited: HashMap<(&'a str, &'a str), Option<StorageRef>>,
+}
+
+impl<'a> Resources<'a> {
+ pub fn native(session: &'a str) -> Self {
+ Self {
+ session,
+ inherited: HashMap::new(),
+ }
+ }
+
+ pub async fn load(
+ log: &EventLog,
+ session: &'a str,
+ events: impl Iterator<Item = EventRef<'a>>,
+ cancellation: &CancellationToken,
+ ) -> Result<Self, RunError> {
+ let mut resources = Self::native(session);
+ for event in events.filter_map(EventRef::canonical) {
+ let mut references = Vec::new();
+ match &event.event.fact {
+ Fact::InvocationOpened {
+ input: InvocationInput::Message { content, .. },
+ ..
+ } => {
+ references.extend(content.attachments.iter().flatten().map(|a| &a.storage_ref));
+ }
+ Fact::MessageSteered { message } => {
+ references.extend(
+ message
+ .content
+ .attachments
+ .iter()
+ .flatten()
+ .map(|a| &a.storage_ref),
+ );
+ }
+ Fact::ToolSettled {
+ outcome:
+ ToolOutcome::Succeeded {
+ model_projection: DurableToolProjection::Content { parts },
+ ..
+ },
+ ..
+ } => {
+ references.extend(parts.iter().filter_map(|part| match part {
+ ProjectionPart::Artifact { image } => Some(&image.reference),
+ ProjectionPart::Text { .. } => None,
+ }));
+ }
+ _ => {}
+ }
+ for reference in references {
+ let StorageRef::SessionFile {
+ session_id,
+ relative_path,
+ } = reference
+ else {
+ continue;
+ };
+ if session_id == session
+ || resources
+ .inherited
+ .contains_key(&(session_id.as_str(), relative_path.as_str()))
+ {
+ continue;
+ }
+ if cancellation.is_cancelled() {
+ return Err(RunError::Cancelled);
+ }
+ let resolved = log
+ .resolve_history_artifact(session, session_id, relative_path)
+ .await?;
+ resources
+ .inherited
+ .insert((session_id, relative_path), resolved);
+ }
+ }
+ Ok(resources)
+ }
+
+ pub fn resolve<'b>(&'b self, reference: &'b StorageRef) -> Option<&'b StorageRef> {
+ match reference {
+ StorageRef::SessionFile {
+ session_id,
+ relative_path,
+ } if session_id != self.session => self
+ .inherited
+ .get(&(session_id.as_str(), relative_path.as_str()))
+ .and_then(Option::as_ref),
+ _ => Some(reference),
+ }
+ }
+}
diff --git a/crates/agent/src/history/user.rs b/crates/agent/src/history/user.rs
index 9c72535b06..d855af3532 100644
--- a/crates/agent/src/history/user.rs
+++ b/crates/agent/src/history/user.rs
@@ -26,6 +26,7 @@ use maka_runtime::{attachment::AttachmentKind, input::MessageInput};
/// is never deduplicated or mistaken for a steering directive.
pub(super) fn project<'a>(
content: &'a MessageInput,
+ resources: &super::resources::Resources<'_>,
steering: bool,
index: usize,
images: &mut Vec<images::Target<'a>>,
@@ -44,7 +45,7 @@ pub(super) fn project<'a>(
}),
);
}
- let text = references::text(content)?;
+ let text = references::text(content, resources)?;
let text = if steering {
format!(
"The user sent a message while you were working:\n<user_query>\n{text}\n</user_query>"
diff --git a/crates/agent/tests/integration/image_projection.rs b/crates/agent/tests/integration/image_projection.rs
index 0de0738a86..9de9dbe5a4 100644
--- a/crates/agent/tests/integration/image_projection.rs
+++ b/crates/agent/tests/integration/image_projection.rs
@@ -47,7 +47,9 @@ async fn reopened_user_and_typed_tool_images_share_actual_byte_budget_with_curre
artifact(&log, "invalid", FOUR_MIB + 1, false).await;
artifact(&log, "overflow", 8, true).await;
let prior = identity("historical");
- let message = serde_json::from_value(json!({"text":"historical", "attachments":[attachment("historical")]})).unwrap();
+ let mut historical = attachment("historical");
+ historical["bytes"] = json!(FOUR_MIB);
+ let message = serde_json::from_value(json!({"text":"historical", "attachments":[historical]})).unwrap();
let facts = [
Fact::InvocationOpened { input: InvocationInput::Message { source_messages: Vec::new(), content: message, request_fingerprint: None }, configuration: None },
Fact::ModelRequested { effective_source_digest: None, purpose: maka_runtime::context::ModelPurpose::Main, context: None, checkpoint_event_id: None, step_id: "prior-step".into(), model_id: "test".into(),
@@ -62,6 +64,11 @@ async fn reopened_user_and_typed_tool_images_share_actual_byte_budget_with_curre
log.append(&EventWrite::tool_success("prior-result".into(), std::time::SystemTime::now(), prior.clone(), "prior-step:read".into(),
ToolOutput::Image(ImageOutput { mime_type: "image/png".into(), reference: serde_json::from_value(attachment("tool")["ref"].clone()).unwrap() }).into()).unwrap().0).await.unwrap();
log.append(&EventWrite::plain(RuntimeEvent::new(prior, Fact::InvocationEnded { outcome: InvocationOutcome::Completed })).unwrap()).await.unwrap();
+ let revision = log.get_session::<Value>("session").await.unwrap().unwrap().revision;
+ assert!(matches!(log.copy_session(maka_event_log::sessions::SessionCopy {
+ source_session_id: "session".into(), target_session_id: "branch".into(),
+ expected_source_revision: revision, cut: maka_event_log::context::HistoryCut::End,
+ }, &json!({}), 2).await.unwrap(), maka_event_log::sessions::SessionCopyResult::Committed(_)));
log.close().await.unwrap();
let log = Arc::new(EventLog::open(&path).await.unwrap());
@@ -106,6 +113,38 @@ async fn reopened_user_and_typed_tool_images_share_actual_byte_budget_with_curre
let canonical = serde_json::to_string(&prefix).unwrap();
assert!(!canonical.contains("data:image"), "projection must not persist image bytes");
assert!(canonical.len() < 32 * 1024);
+ for id in ["historical", "tool"] {
+ assert_eq!(log.delete_user_artifact("session", id).await.unwrap(),
+ maka_event_log::artifacts::ArtifactDeletion::Deleted);
+ }
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let base = format!("http://{}/v1", listener.local_addr().unwrap());
+ let copied_request = tokio::spawn(async move {
+ let (mut socket, _) = listener.accept().await.unwrap();
+ let body = request(&mut socket).await;
+ respond(&mut socket).await;
+ body
+ });
+ let mut copied = input(&base, json!([]));
+ copied.invocation = identity("branch-current");
+ copied.invocation.session_id = "branch".into();
+ engine.run(copied, CancellationToken::new()).await.unwrap();
+ let copied_request = copied_request.await.unwrap();
+ let mut inherited_images = Vec::new();
+ collect_images(&copied_request["messages"], &mut inherited_images);
+ assert_eq!(inherited_images.len(), 2, "copy retains both user and typed tool images after source deletion");
+ let reference = log.resolve_history_artifact("branch", "session", "historical").await.unwrap().unwrap();
+ assert!(copied_request["messages"].to_string().contains(&reference.resource_ref().unwrap()),
+ "Read must receive the destination-owned resource URI");
+ let tool_reference = log.resolve_history_artifact("branch", "session", "tool").await.unwrap().unwrap();
+ let maka_runtime::attachment::StorageRef::SessionFile { relative_path, .. } = tool_reference else { panic!("owned tool image") };
+ assert!(log.read_tool_result("branch", "prior-result").await.unwrap().unwrap().serialized_result.contains(&relative_path),
+ "tool-result Read renders the copied image's locator, not the source artifact ID");
+ let original = log.prefix(100, 128 * 1024).await.unwrap();
+ let source_opening = original.events.iter().find(|e| e.event.invocation.invocation_id == "invocation-historical"
+ && matches!(e.event.fact, Fact::InvocationOpened { .. })).unwrap();
+ assert!(!serde_json::to_string(&source_opening.event).unwrap().contains("history-"),
+ "rendering aliases cannot rewrite canonical input");
engine.drain().await;
}).await.expect("image projection must make bounded progress");
}
diff --git a/crates/event-log/src/archive/accepted.rs b/crates/event-log/src/archive/accepted.rs
index 910ccadee3..cd3ccbd42f 100644
--- a/crates/event-log/src/archive/accepted.rs
+++ b/crates/event-log/src/archive/accepted.rs
@@ -209,6 +209,71 @@ pub(crate) async fn archived(
}))
}
+enum ArchiveCut {
+ Current,
+ Pinned(Option<u64>),
+}
+
+impl ArchiveCut {
+ async fn resolve(
+ &self,
+ connection: &mut SqliteConnection,
+ target: &target::Target,
+ ) -> Result<Option<ArchivedPlaceholder>, StoreError> {
+ let before = match self {
+ Self::Current => i64::MAX as u64,
+ Self::Pinned(None) => return Ok(None),
+ Self::Pinned(Some(sequence)) => sequence + 1,
+ };
+ let found = find(
+ connection,
+ &target.invocation.session_id,
+ &target.event_id,
+ before,
+ )
+ .await?;
+ if let Self::Pinned(expected) = self
+ && found.as_ref().map(|(sequence, _)| *sequence) != *expected
+ {
+ return Err(ArchiveError::Corrupt.into());
+ }
+ Ok(found.map(|(_, placeholder)| placeholder))
+ }
+}
+
+/// Read ownership is separate from the original execution's append authority.
+/// A copied target sees only the archive identity accepted when it was copied.
+async fn owned_target(
+ connection: &mut SqliteConnection,
+ session: &str,
+ event_id: &str,
+) -> Result<Option<(target::Target, ArchiveCut)>, StoreError> {
+ let owner: Option<(String, bool, Option<i64>)> = sqlx::query_as(
+ "SELECT json_extract(e.event_json,'$.invocation.session_id'),
+ h.sequence IS NOT NULL, h.archive_sequence
+ FROM runtime_events e LEFT JOIN session_history_members h
+ ON h.sequence=e.sequence AND h.session_id=?1
+ WHERE e.event_id=?2 AND e.kind='tool_settled'
+ AND (json_extract(e.event_json,'$.invocation.session_id')=?1 OR h.sequence IS NOT NULL)",
+ )
+ .bind(session)
+ .bind(event_id)
+ .fetch_optional(&mut *connection)
+ .await?;
+ let Some((source, inherited, archive_sequence)) = owner else {
+ return Ok(None);
+ };
+ let cut = if inherited {
+ ArchiveCut::Pinned(archive_sequence.map(sequence_number).transpose()?)
+ } else {
+ ArchiveCut::Current
+ };
+ let target = target::read(connection, &source, event_id)
+ .await?
+ .ok_or(ArchiveError::Corrupt)?;
+ Ok(Some((target, cut)))
+}
+
impl EventLog {
/// Read the committed model projection with or without a later archive replacement.
pub async fn read_tool_result(
@@ -223,18 +288,48 @@ impl EventLog {
.run(move |connection| {
Box::pin(async move {
let mut tx = connection.begin().await?;
- let Some(target) = target::read(&mut tx, &session, &event_id).await? else {
+ let Some((mut target, cut)) =
+ owned_target(&mut tx, &session, &event_id).await?
+ else {
return Ok(None);
};
target
.projection
- .validate(&session)
+ .validate(&target.invocation.session_id)
+ .map_err(|_| ArchiveError::Corrupt)?;
+ if let Some(placeholder) = cut.resolve(&mut tx, &target).await? {
+ validate(&target, &placeholder)?;
+ }
+ // Verify original evidence before rendering typed media locators.
+ // Raw archive reads below retain the original digest and bytes.
+ if target.invocation.session_id != session
+ && let maka_runtime::tool_output::DurableToolProjection::Content { parts } =
+ &mut target.projection
+ {
+ for part in parts {
+ if let maka_runtime::tool_output::ProjectionPart::Artifact { image } =
+ part
+ {
+ let maka_runtime::attachment::StorageRef::SessionFile {
+ session_id,
+ relative_path,
+ } = &image.reference
+ else {
+ return Err(ArchiveError::Corrupt.into());
+ };
+ image.reference = crate::artifacts::history::resolve(
+ &mut tx,
+ &session,
+ session_id,
+ relative_path,
+ )
+ .await?
+ .ok_or(ArchiveError::Corrupt)?;
+ }
+ }
+ }
+ let bytes = maka_runtime::archive::encode_projection(&target.projection)
.map_err(|_| ArchiveError::Corrupt)?;
- let bytes = match find(&mut tx, &session, &event_id, i64::MAX as u64).await? {
- Some((_, placeholder)) => validate(&target, &placeholder)?,
- None => maka_runtime::archive::encode_projection(&target.projection)
- .map_err(|_| ArchiveError::Corrupt)?,
- };
let serialized_result =
String::from_utf8(bytes).map_err(|_| ArchiveError::Corrupt)?;
tx.commit().await?;
@@ -276,16 +371,22 @@ impl EventLog {
self.validate_root()?;
crate::sessions::validate_id(session)?;
let (session, event_id) = (session.to_owned(), event_id.to_owned());
- self.connection.run(move |connection| Box::pin(async move {
- let mut tx = connection.begin().await?;
- let exists: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM runtime_events WHERE event_id=? AND json_extract(event_json,'$.invocation.session_id')=?)")
- .bind(&event_id).bind(&session).fetch_one(&mut *tx).await?;
- if !exists { return Ok(None); }
- let Some((_, placeholder)) = find(&mut tx, &session, &event_id, i64::MAX as u64).await? else { return Ok(None); };
- let target = target::read(&mut tx, &session, &event_id).await?.ok_or(ArchiveError::Corrupt)?;
- let body = validate(&target, &placeholder)?;
- tx.commit().await?;
- Ok(Some((placeholder.identity, body)))
- })).await
+ self.connection
+ .run(move |connection| {
+ Box::pin(async move {
+ let mut tx = connection.begin().await?;
+ let Some((target, cut)) = owned_target(&mut tx, &session, &event_id).await?
+ else {
+ return Ok(None);
+ };
+ let Some(placeholder) = cut.resolve(&mut tx, &target).await? else {
+ return Ok(None);
+ };
+ let body = validate(&target, &placeholder)?;
+ tx.commit().await?;
+ Ok(Some((placeholder.identity, body)))
+ })
+ })
+ .await
}
}
diff --git a/crates/event-log/src/artifacts/history.rs b/crates/event-log/src/artifacts/history.rs
index d0924a29d6..c7ede0f7dc 100644
--- a/crates/event-log/src/artifacts/history.rs
+++ b/crates/event-log/src/artifacts/history.rs
@@ -26,6 +26,56 @@ use maka_runtime::{
};
use sqlx::SqliteConnection;
+impl crate::EventLog {
+ /// Resolve original provenance only through this Session's immutable copy
+ /// ownership. Knowing a source Session or Artifact ID grants no access.
+ pub async fn resolve_history_artifact(
+ &self,
+ session: &str,
+ source_session: &str,
+ source_artifact: &str,
+ ) -> Result<Option<StorageRef>, StoreError> {
+ self.validate_root()?;
+ for id in [session, source_session, source_artifact] {
+ crate::sessions::validate_id(id)?;
+ }
+ let (session, source_session, source_artifact) = (
+ session.to_owned(),
+ source_session.to_owned(),
+ source_artifact.to_owned(),
+ );
+ self.connection
+ .run(move |connection| {
+ Box::pin(async move {
+ resolve(connection, &session, &source_session, &source_artifact).await
+ })
+ })
+ .await
+ }
+}
+
+pub(crate) async fn resolve(
+ connection: &mut SqliteConnection,
+ session: &str,
+ source_session: &str,
+ source_artifact: &str,
+) -> Result<Option<StorageRef>, StoreError> {
+ let id: Option<String> = sqlx::query_scalar(
+ "SELECT h.artifact_id FROM session_history_artifacts h
+ JOIN artifacts a ON a.session_id=h.session_id AND a.id=h.artifact_id
+ WHERE h.session_id=? AND h.source_session_id=? AND h.source_artifact_id=?",
+ )
+ .bind(session)
+ .bind(source_session)
+ .bind(source_artifact)
+ .fetch_optional(connection)
+ .await?;
+ Ok(id.map(|relative_path| StorageRef::SessionFile {
+ session_id: session.into(),
+ relative_path,
+ }))
+}
+
/// Only typed canonical references convey resource ownership. Never interpret
/// plugin JSON or strings containing an attachment URI as an access grant.
pub(crate) async fn retain(
diff --git a/crates/event-log/src/transcript.rs b/crates/event-log/src/transcript.rs
index e74654dc15..9c494b9f21 100644
--- a/crates/event-log/src/transcript.rs
+++ b/crates/event-log/src/transcript.rs
@@ -74,8 +74,8 @@ impl EventLog {
// and budget-checked separately, one boundary at a time.
let boundaries = {
let records = sqlx::query(
- "SELECT sequence, invocation_id, operation_id FROM runtime_events
- WHERE json_extract(event_json, '$.invocation.session_id') = ?1
+ "SELECT sequence, invocation_id, operation_id FROM session_history_events runtime_events
+ WHERE owner_session_id = ?1
AND sequence > ?2 AND sequence <= ?3
AND NOT EXISTS (SELECT 1 FROM runtime_events opening
WHERE opening.invocation_id = runtime_events.invocation_id
@@ -148,7 +148,10 @@ impl EventLog {
} else {
None
};
- for row in view.push_with_tool_output(&fact, resolved.as_ref())? {
+ for mut row in view.push_with_tool_output(&fact, resolved.as_ref())? {
+ if fact.event.invocation.session_id != session {
+ owned_references(&mut tx, &session, &mut row.message).await?;
+ }
persist(&mut tx, &session, row).await?;
}
}
@@ -171,6 +174,39 @@ impl EventLog {
}
}
+async fn owned_references(
+ tx: &mut SqliteConnection,
+ session: &str,
+ message: &mut maka_presentation::Message,
+) -> Result<(), StoreError> {
+ use maka_presentation::{Content, ToolContent};
+ use maka_runtime::attachment::StorageRef;
+ let references: Vec<_> = match &mut message.content {
+ Content::User { attachments, .. } => attachments
+ .iter_mut()
+ .flatten()
+ .map(|a| &mut a.storage_ref)
+ .collect(),
+ Content::ToolResult {
+ content: ToolContent::Image { reference, .. },
+ ..
+ } => vec![reference],
+ _ => Vec::new(),
+ };
+ for reference in references {
+ if let StorageRef::SessionFile {
+ session_id,
+ relative_path,
+ } = reference
+ {
+ *reference = crate::artifacts::history::resolve(tx, session, session_id, relative_path)
+ .await?
+ .ok_or(StoreError::TranscriptConflict)?;
+ }
+ }
+ Ok(())
+}
+
async fn persist(tx: &mut SqliteConnection, session: &str, row: Row) -> Result<(), StoreError> {
let bytes = serde_json::to_vec(&row.message)?;
let sequence =
diff --git a/crates/event-log/src/transcript/navigation/queries.rs b/crates/event-log/src/transcript/navigation/queries.rs
index 54a747c39c..d82a97b520 100644
--- a/crates/event-log/src/transcript/navigation/queries.rs
+++ b/crates/event-log/src/transcript/navigation/queries.rs
@@ -18,10 +18,17 @@
*/
pub(super) const FENCE: &str = "
-SELECT MAX(event.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
-AND json_extract(opening.event_json, '$.fact.input.kind') IN ('message', 'continuation', 'handoff')";
+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
+ AND json_extract(opening.event_json, '$.fact.input.kind') IN ('message', 'continuation', 'handoff')
+ UNION ALL
+ SELECT MAX(h.sequence) FROM session_history_members h
+ JOIN runtime_events event ON event.sequence=h.sequence
+ JOIN runtime_events opening ON opening.invocation_id = event.invocation_id AND opening.kind = 'invocation_opened'
+ WHERE h.session_id = ?1
+ AND json_extract(opening.event_json, '$.fact.input.kind') IN ('message', 'continuation', 'handoff'))";
pub(super) const ROWS: &str = "
SELECT row.sequence, row.turn_id,
@@ -39,8 +46,8 @@ 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 runtime_events opening
- WHERE json_extract(opening.event_json, '$.invocation.session_id') = ?1
+ 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')
), candidates AS (
diff --git a/crates/event-log/tests/archive.rs b/crates/event-log/tests/archive.rs
index e59e5214a1..9d604b665a 100644
--- a/crates/event-log/tests/archive.rs
+++ b/crates/event-log/tests/archive.rs
@@ -253,6 +253,32 @@ async fn archive_atomic_retry_reopen_scope_and_source_integrity() {
.read_model_context("archived-copy", None, 100, 64 * 1024)
.await
.unwrap();
+ assert_eq!(
+ log.read_archive("archived-copy", &placeholder.identity)
+ .await
+ .unwrap()
+ .unwrap(),
+ expected,
+ "copy reads its accepted archive without borrowing source Session access"
+ );
+ assert!(
+ log.read_archive("unpruned-copy", &placeholder.identity)
+ .await
+ .unwrap()
+ .is_none(),
+ "a later source archive does not become an accepted archive in an older copy"
+ );
+ for session in ["archived-copy", "unpruned-copy"] {
+ assert_eq!(
+ log.read_tool_result(session, &target.event().id)
+ .await
+ .unwrap()
+ .unwrap()
+ .serialized_result
+ .as_bytes(),
+ expected
+ );
+ }
assert!(
adopted
.tail
@@ -402,6 +428,12 @@ async fn archive_atomic_retry_reopen_scope_and_source_integrity() {
.await
.is_err()
);
+ assert!(
+ log.read_archive("archived-copy", &placeholder.identity)
+ .await
+ .is_err(),
+ "history membership never bypasses original evidence integrity"
+ );
assert!(
log.read_archive("session", &placeholder.identity)
.await
diff --git a/crates/event-log/tests/session_copy.rs b/crates/event-log/tests/session_copy.rs
index 6cf55f2664..5d82d1ef80 100644
--- a/crates/event-log/tests/session_copy.rs
+++ b/crates/event-log/tests/session_copy.rs
@@ -265,6 +265,68 @@ async fn copies_own_history_and_files_without_replaying_execution_across_retries
.bytes,
b"hello"
);
+ let fence = log.navigation_fence(session).await.unwrap().unwrap();
+ assert_eq!(
+ fence,
+ maka_presentation::watermark(root.high_water).unwrap()
+ );
+ assert!(
+ log.prepare_transcript(session, root.high_water, 32)
+ .await
+ .unwrap()
+ );
+ let turns = log.navigation_turns(session, fence, 0, 32).await.unwrap();
+ assert_eq!(turns.contributions.len(), 1);
+ assert_eq!(turns.contributions[0].turn_id, "first");
+ assert_eq!(
+ log.navigation_landmarks(session, fence, 8, None)
+ .await
+ .unwrap()
+ .len(),
+ 1
+ );
+ let history = log
+ .history_text(session, root.high_water, None)
+ .await
+ .unwrap();
+ let maka_plugins::session::history::Page::Ready { chunks, next, .. } = &history else {
+ panic!("prepared history");
+ };
+ assert!(next.is_none());
+ assert_eq!(chunks.len(), 1);
+ assert_eq!(chunks[0].text, "first\ninput.txt");
+ assert_eq!(
+ chunks[0].attachments[0].storage_ref,
+ StorageRef::SessionFile {
+ session_id: session.into(),
+ relative_path: files[0].id.clone(),
+ }
+ );
+ let inspect = rusqlite::Connection::open(&path).unwrap();
+ for table in ["transcript_text", "transcript_rows", "transcript_progress"] {
+ inspect
+ .execute(
+ &format!("DELETE FROM {table} WHERE session_id = ?"),
+ [session],
+ )
+ .unwrap();
+ }
+ drop(inspect);
+ assert!(
+ log.prepare_transcript(session, root.high_water, 32)
+ .await
+ .unwrap()
+ );
+ assert_eq!(
+ serde_json::to_value(
+ log.history_text(session, root.high_water, None)
+ .await
+ .unwrap()
+ )
+ .unwrap(),
+ serde_json::to_value(history).unwrap(),
+ "rebuilding cannot depend on source metadata or cache state"
+ );
}
assert_eq!(
log.prefix(100, 65536).await.unwrap().events.len(),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org