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 20:57:23 UTC
(maka) branch rust-preview updated: perf(agent): check session availability without decoding snapshots
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 efe03a63f7 perf(agent): check session availability without decoding snapshots
efe03a63f7 is described below
commit efe03a63f76163c36e1b658cf48c724ead372296
Author: Wang <m4...@gmail.com>
AuthorDate: Fri Oct 23 04:57:12 2026 +0800
perf(agent): check session availability without decoding snapshots
---
platform/event-log/src/sessions.rs | 23 ++++
platform/event-log/tests/session_copy/mod.rs | 19 +++
platform/event-log/tests/sessions.rs | 101 ++++++++++++++++
plugins/agent/native/src/owner.rs | 15 +++
plugins/agent/native/src/owner/tests.rs | 3 +
.../native/src/owner/tests/session_availability.rs | 129 +++++++++++++++++++++
plugins/agent/native/src/service.rs | 34 +++---
7 files changed, 307 insertions(+), 17 deletions(-)
diff --git a/platform/event-log/src/sessions.rs b/platform/event-log/src/sessions.rs
index 8aa1aaa133..963cffbf90 100644
--- a/platform/event-log/src/sessions.rs
+++ b/platform/event-log/src/sessions.rs
@@ -221,6 +221,29 @@ impl EventLog {
.await
}
+ /// Whether ordinary reads may see this Session. Archiving preserves visibility;
+ /// removing hides it before physical cleanup. This is not caller authorization
+ /// or admission for new work, and does not decode the Session's metadata.
+ pub async fn session_is_available(&self, id: &str) -> Result<bool, StoreError> {
+ self.validate_root()?;
+ validate_id(id)?;
+ let id = id.to_owned();
+ self.connection
+ .read(move |connection| {
+ Box::pin(async move {
+ Ok(sqlx::query_scalar(
+ "SELECT EXISTS(SELECT 1 FROM session_control WHERE id = ?1)
+ AND NOT EXISTS(SELECT 1 FROM session_retirements
+ WHERE session_id = ?1 AND remove_session != 0)",
+ )
+ .bind(id)
+ .fetch_one(connection)
+ .await?)
+ })
+ })
+ .await
+ }
+
pub async fn get_session<T: DeserializeOwned + Send + 'static>(
&self,
id: &str,
diff --git a/platform/event-log/tests/session_copy/mod.rs b/platform/event-log/tests/session_copy/mod.rs
index 13b119624f..06c1e7f9bf 100644
--- a/platform/event-log/tests/session_copy/mod.rs
+++ b/platform/event-log/tests/session_copy/mod.rs
@@ -350,6 +350,19 @@ async fn family_removal_freezes_one_atomic_plan_and_does_not_rearchive_restored_
else {
panic!("unchanged revision");
};
+ for session in &plan.remove {
+ assert!(!log.session_is_available(session).await.unwrap());
+ }
+ for session in &plan.archive {
+ assert!(log.session_is_available(session).await.unwrap());
+ assert_eq!(
+ log.session_retirement(session).await.unwrap(),
+ Some(SessionRetirement::Archiving)
+ );
+ }
+ for session in ["draft", "branch"] {
+ assert!(log.session_is_available(session).await.unwrap());
+ }
assert_eq!(plan.archived_subtask_count, 3);
assert!(plan.archive.contains(&"late".into()));
assert!(log.get_session::<Value>("root").await.unwrap().is_none());
@@ -381,6 +394,12 @@ async fn family_removal_freezes_one_atomic_plan_and_does_not_rearchive_restored_
SessionRetirement::Archived
);
}
+ for session in &plan.remove {
+ assert!(!log.session_is_available(session).await.unwrap());
+ }
+ for session in &plan.archive {
+ assert!(log.session_is_available(session).await.unwrap());
+ }
let receipt = log
.session_copy_receipt("child-revision")
.await
diff --git a/platform/event-log/tests/sessions.rs b/platform/event-log/tests/sessions.rs
index 1c528aacc4..30113688e1 100644
--- a/platform/event-log/tests/sessions.rs
+++ b/platform/event-log/tests/sessions.rs
@@ -628,3 +628,104 @@ async fn lifecycle_obeys_canonical_invocation_seal_and_validates_bounds() {
);
reopened.close().await.unwrap();
}
+
+#[tokio::test]
+async fn availability_tracks_visibility_without_admitting_work_or_decoding_metadata() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("events.sqlite");
+ let log = EventLog::open(&path).await.unwrap();
+ assert!(!log.session_is_available("missing").await.unwrap());
+ for invalid in ["", "a/b", "a.b", "a b"] {
+ assert!(matches!(
+ log.session_is_available(invalid).await,
+ Err(StoreError::InvalidTransition(_))
+ ));
+ }
+ let session = log
+ .create_session("visible", "create", &json!({}), 1)
+ .await
+ .unwrap();
+ assert!(log.session_is_available("visible").await.unwrap());
+ let archived = log
+ .set_session_archived::<Value>("visible", true, 2)
+ .await
+ .unwrap();
+ assert!(archived.archived);
+ assert!(log.session_is_available("visible").await.unwrap());
+ assert!(log.get_session::<Value>("visible").await.unwrap().is_some());
+ let active = log
+ .set_session_archived::<Value>("visible", false, 3)
+ .await
+ .unwrap();
+ assert!(active.revision > session.revision);
+ // Visibility is independent of unrelated metadata decoding. Full readers
+ // must still reject malformed configuration instead of manufacturing a record.
+ let faults = rusqlite::Connection::open(&path).unwrap();
+ faults
+ .execute(
+ "UPDATE session_control SET configuration='invalid json' WHERE id='visible'",
+ [],
+ )
+ .unwrap();
+ assert!(log.session_is_available("visible").await.unwrap());
+ assert!(matches!(
+ log.get_session::<Value>("visible").await,
+ Err(StoreError::Json(_))
+ ));
+ faults
+ .execute(
+ "UPDATE session_control SET configuration='{}' WHERE id='visible'",
+ [],
+ )
+ .unwrap();
+ drop(faults);
+ assert_eq!(
+ log.begin_session_removal("visible", active.revision)
+ .await
+ .unwrap(),
+ maka_event_log::sessions::SessionRemovalResult::Accepted(
+ maka_event_log::sessions::SessionRetirement::Removing
+ )
+ );
+ assert!(!log.session_is_available("visible").await.unwrap());
+ assert!(log.get_session::<Value>("visible").await.unwrap().is_none());
+ assert!(
+ log.retiring_session::<Value>("visible")
+ .await
+ .unwrap()
+ .is_some()
+ );
+ assert_eq!(
+ log.finish_session_retirement("visible").await.unwrap(),
+ maka_event_log::sessions::SessionRetirement::Removed
+ );
+ assert!(!log.session_is_available("visible").await.unwrap());
+ log.close().await.unwrap();
+ let reopened = EventLog::open(&path).await.unwrap();
+ assert!(!reopened.session_is_available("visible").await.unwrap());
+ reopened.close().await.unwrap();
+}
+
+#[tokio::test]
+async fn availability_rejects_a_root_that_lost_its_authority() {
+ use maka_event_log::root::{RootNamespaces, RootOwner};
+ let directory = tempfile::tempdir().unwrap();
+ let namespaces = RootNamespaces {
+ ownership: directory.path().join("owners"),
+ control: directory.path().join("control"),
+ };
+ let root = directory.path().join("root");
+ let owner = std::sync::Arc::new(RootOwner::create(&root, &namespaces).unwrap());
+ let log = EventLog::for_root(owner).await.unwrap();
+ log.create_session("visible", "create", &json!({}), 1)
+ .await
+ .unwrap();
+ assert!(log.session_is_available("visible").await.unwrap());
+ std::fs::remove_dir_all(&namespaces.control).unwrap();
+ assert!(matches!(
+ log.session_is_available("visible").await,
+ Err(StoreError::Io(_))
+ ));
+ assert!(log.get_session::<Value>("visible").await.is_err());
+ log.close().await.unwrap();
+}
diff --git a/plugins/agent/native/src/owner.rs b/plugins/agent/native/src/owner.rs
index 53d5d7085b..9bc9bdb02f 100644
--- a/plugins/agent/native/src/owner.rs
+++ b/plugins/agent/native/src/owner.rs
@@ -494,6 +494,21 @@ impl Owner {
pub fn owns(&self, id: &str) -> bool {
owns_session(&self.prefix, id)
}
+ pub async fn require_session(&self, id: &str) -> Result<(), String> {
+ if !self.owns(id) {
+ return Err("agent_session_unavailable".into());
+ }
+ if self
+ .store
+ .session_is_available(id)
+ .await
+ .map_err(|_| "agent_store_unavailable".to_owned())?
+ {
+ Ok(())
+ } else {
+ Err("agent_session_unavailable".into())
+ }
+ }
pub async fn session(&self, id: &str) -> Result<SessionRecord<SessionConfiguration>, String> {
if !self.owns(id) {
return Err("agent_session_unavailable".into());
diff --git a/plugins/agent/native/src/owner/tests.rs b/plugins/agent/native/src/owner/tests.rs
index ed06630586..349e409a7a 100644
--- a/plugins/agent/native/src/owner/tests.rs
+++ b/plugins/agent/native/src/owner/tests.rs
@@ -73,6 +73,9 @@ mod interactions;
#[path = "tests/catalog.rs"]
mod catalog;
+#[path = "tests/session_availability.rs"]
+mod session_availability;
+
#[path = "tests/history_search.rs"]
mod history_search;
diff --git a/plugins/agent/native/src/owner/tests/session_availability.rs b/plugins/agent/native/src/owner/tests/session_availability.rs
new file mode 100644
index 0000000000..20812c1cf3
--- /dev/null
+++ b/plugins/agent/native/src/owner/tests/session_availability.rs
@@ -0,0 +1,129 @@
+/*
+ * 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::*;
+
+#[tokio::test]
+async fn session_read_guards_reject_foreign_missing_and_logically_removed_sessions() {
+ tokio::time::timeout(Duration::from_secs(30), async {
+ let fixture = Fixture::new("http://127.0.0.1:9/v1", None).await;
+ let session = fixture.create("Availability").await;
+ fixture.owner.require_session(&session).await.unwrap();
+ let record = fixture.owner.session(&session).await.unwrap();
+ fixture.store.create_session("foreign_session", "foreign", &record.configuration, now()).await.unwrap();
+ let missing = format!("{}{}", fixture.owner.prefix, Uuid::new_v4().simple());
+ fixture.store.begin_session_removal(&session, record.revision).await.unwrap();
+ assert!(fixture.store.retiring_session::<SessionConfiguration>(&session).await.unwrap().is_some());
+ let row = json!({"sourceSequence":1,"itemIndex":0});
+ let requests = [
+ ("observe", json!({})),
+ ("history", json!({"position":{"kind":"start"}})),
+ ("historyLandmarks", json!({"through":0})),
+ ("historyPrompt", json!({"through":1,"position":row})),
+ ("search", json!({"through":0,"query":"needle","caseSensitive":false,"wholeWord":false,"includeInternal":false})),
+ ("source", json!({"through":0,"source":{"kind":"row","position":row},"field":"body","position":{"kind":"start"}})),
+ ("toolImage", json!({"through":0,"position":row,"offset":0})),
+ ("attachmentImage", json!({"source":{"kind":"uploaded","id":"unknown"},"offset":0})),
+ ];
+ for id in ["foreign_session", missing.as_str(), session.as_str()] {
+ assert_eq!(fixture.owner.require_session(id).await.unwrap_err(), "agent_session_unavailable");
+ for (method, input) in &requests {
+ let mut input = input.clone();
+ input["sessionId"] = json!(id);
+ let error = fixture.query(method, input).await.unwrap_err();
+ assert!(error.contains("agent_session_unavailable"), "{method} {id}: {error}");
+ }
+ let mut watch = fixture.app.client().subscribe(fixture.binding(2).await, "watch".into(),
+ json!({"sessionId":id,"after":0}), scope(&fixture.issuer, true),
+ Attachments::new().into()).await.unwrap();
+ assert!(watch.recv().await.is_none());
+ let error = watch.completion().await.result.unwrap_err();
+ assert!(error.contains("agent_session_unavailable"), "watch {id}: {error}");
+ }
+ fixture.app.shutdown().await.unwrap();
+ fixture.config.shutdown().await.unwrap();
+ }).await.unwrap();
+}
+
+#[tokio::test]
+async fn history_visibility_does_not_decode_configuration_but_source_still_requires_it() {
+ tokio::time::timeout(Duration::from_secs(30), async {
+ let fixture = Fixture::new("http://127.0.0.1:9/v1", None).await;
+ let session = fixture.create("History without configuration").await;
+ let history_input = json!({"sessionId":session,"position":{"kind":"start"}});
+ let expected = fixture
+ .query("history", history_input.clone())
+ .await
+ .unwrap();
+ assert_eq!(expected["kind"], "ready");
+ assert_eq!(expected["chunks"], json!([]));
+ let connection =
+ rusqlite::Connection::open(fixture.root.join(maka_event_log::root::ROOT_DATABASE))
+ .unwrap();
+ connection
+ .execute(
+ "UPDATE session_control SET configuration='invalid json' WHERE id=?",
+ [&session],
+ )
+ .unwrap();
+ drop(connection);
+ fixture.owner.require_session(&session).await.unwrap();
+ assert_eq!(
+ fixture.owner.session(&session).await.err().unwrap(),
+ "agent_store_unavailable"
+ );
+ assert_eq!(
+ fixture
+ .query("history", history_input.clone())
+ .await
+ .unwrap(),
+ expected
+ );
+ let error = fixture
+ .query(
+ "source",
+ json!({"sessionId":session,"through":0,
+ "source":{"kind":"row","position":{"sourceSequence":1,"itemIndex":0}},
+ "field":"body","position":{"kind":"start"}}),
+ )
+ .await
+ .unwrap_err();
+ assert!(error.contains("agent_store_unavailable"), "{error}");
+ let cancelled = scope(&fixture.issuer, true);
+ cancelled.cancellation.cancel();
+ if let Ok(read) = fixture
+ .app
+ .client()
+ .start(
+ fixture.binding(2).await,
+ "history".into(),
+ history_input,
+ cancelled,
+ Attachments::new().into(),
+ )
+ .await
+ {
+ assert!(read.result().await.is_err());
+ }
+ fixture.app.shutdown().await.unwrap();
+ fixture.config.shutdown().await.unwrap();
+ })
+ .await
+ .unwrap();
+}
diff --git a/plugins/agent/native/src/service.rs b/plugins/agent/native/src/service.rs
index f92f5ca5a5..118a0bb7b8 100644
--- a/plugins/agent/native/src/service.rs
+++ b/plugins/agent/native/src/service.rs
@@ -312,7 +312,7 @@ impl Handler for Agent {
"observe" => {
use maka_event_log::turns::InvocationState;
let input: SessionRead = decode(input)?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
let observed = owner
.store
.observe_session::<SessionConfiguration>(&input.session_id)
@@ -408,7 +408,7 @@ impl Handler for Agent {
})?;
maka_plugin_core::json::exact_numbers(&result)
.map_err(|_| "agent_observation_unavailable")?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if context.scope().cancellation.is_cancelled() {
return Err("agent_observation_cancelled".into());
}
@@ -418,7 +418,7 @@ impl Handler for Agent {
use maka_agent_ports::session::history::Page;
use maka_agent_ports::session::history::Read;
let input: History = decode(input)?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
let high_water = *owner.store.subscribe_commits().borrow();
let through = input.through.unwrap_or(high_water);
if through > high_water {
@@ -490,7 +490,7 @@ impl Handler for Agent {
.collect(),
}
};
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if context.scope().cancellation.is_cancelled() {
return Err("agent_observation_cancelled".into());
}
@@ -501,7 +501,7 @@ impl Handler for Agent {
}
"historyLandmarks" => {
let input: HistoryLandmarksRead = decode(input)?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if input.through > *owner.store.subscribe_commits().borrow() {
return Err("agent_cursor_invalid".into());
}
@@ -522,7 +522,7 @@ impl Handler for Agent {
})
.collect(),
};
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if context.scope().cancellation.is_cancelled() {
return Err("agent_observation_cancelled".into());
}
@@ -533,7 +533,7 @@ impl Handler for Agent {
}
"historyPrompt" => {
let input: HistoryPromptRead = decode(input)?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if input.through > *owner.store.subscribe_commits().borrow() {
return Err("agent_cursor_invalid".into());
}
@@ -542,7 +542,7 @@ impl Handler for Agent {
.history_prompt(&input.session_id, input.through, input.position)
.await
.map_err(|_| "agent_history_unavailable")?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if context.scope().cancellation.is_cancelled() {
return Err("agent_observation_cancelled".into());
}
@@ -550,7 +550,7 @@ impl Handler for Agent {
}
"search" => {
let input: HistorySearch = decode(input)?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if input.through > *owner.store.subscribe_commits().borrow()
|| input.after.is_some_and(|after| {
@@ -591,7 +591,7 @@ impl Handler for Agent {
next_after: batch.next_after,
}
};
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if context.scope().cancellation.is_cancelled() {
return Err("agent_observation_cancelled".into());
}
@@ -642,7 +642,7 @@ impl Handler for Agent {
file_path,
}
};
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if context.scope().cancellation.is_cancelled() {
return Err("agent_observation_cancelled".into());
}
@@ -653,7 +653,7 @@ impl Handler for Agent {
}
"toolImage" => {
let input: ToolImageRead = decode(input)?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if input.through > *owner.store.subscribe_commits().borrow() {
return Err("agent_cursor_invalid".into());
}
@@ -679,7 +679,7 @@ impl Handler for Agent {
data: STANDARD.encode(image.bytes),
}
};
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if context.scope().cancellation.is_cancelled() {
return Err("agent_observation_cancelled".into());
}
@@ -693,7 +693,7 @@ impl Handler for Agent {
}
"attachmentImage" => {
let input: AttachmentImageRead = decode(input)?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
let image = match &input.source {
AttachmentImageSource::Uploaded { id } => {
owner
@@ -723,7 +723,7 @@ impl Handler for Agent {
}
}
.map_err(|_| "agent_image_unavailable")?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
if context.scope().cancellation.is_cancelled() {
return Err("agent_observation_cancelled".into());
}
@@ -821,7 +821,7 @@ impl Handler for Agent {
}
"watch" => {
let input: Watch = decode(input)?;
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
let mut commits = owner.store.subscribe_commits();
let mut after = input.after;
let mut queue_revision = None;
@@ -852,7 +852,7 @@ impl Handler for Agent {
.emit(encode(DeliveryPage::Interactions { pending })?)
.await?;
}
- owner.session(&input.session_id).await?;
+ owner.require_session(&input.session_id).await?;
let through = *commits.borrow();
if after > through {
return Err("agent_cursor_invalid".into());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org