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 11:59:56 UTC
(maka) branch feat/runtime-host-rust updated: feat(client): expose native session branches and revisions
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 ec16f7b542 feat(client): expose native session branches and revisions
ec16f7b542 is described below
commit ec16f7b5420a8f74792bac1e1418797848306423
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Wed Sep 23 19:59:34 2026 +0800
feat(client): expose native session branches and revisions
Preserve copy intent and source boundaries, validate receipt identity, and exercise revision lifecycle through the native Client against a real Host.
Generated-by: Codex
---
crates/client/src/session.rs | 2 +
crates/client/src/session/copy.rs | 168 +++++++++++++++++++++
crates/client/tests/integration/session.rs | 2 +
crates/client/tests/integration/session/copy.rs | 96 ++++++++++++
.../tests/integration/session_history.rs | 158 ++++++++++---------
5 files changed, 356 insertions(+), 70 deletions(-)
diff --git a/crates/client/src/session.rs b/crates/client/src/session.rs
index ac834d33cd..3e216e4abe 100644
--- a/crates/client/src/session.rs
+++ b/crates/client/src/session.rs
@@ -20,6 +20,8 @@
use crate::{Client, ClientError, RequestFailure};
use maka_protocol::{Operation, ProtocolError, session::*};
+mod copy;
+
impl Client {
pub async fn session_catalog(
&self,
diff --git a/crates/client/src/session/copy.rs b/crates/client/src/session/copy.rs
new file mode 100644
index 0000000000..6661de4c47
--- /dev/null
+++ b/crates/client/src/session/copy.rs
@@ -0,0 +1,168 @@
+/*
+ * 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 crate::{Client, ClientError, RequestFailure};
+use maka_protocol::{
+ Operation, ProtocolError,
+ session::{copy, sources},
+};
+use serde_json::{Value, json};
+
+impl Client {
+ pub async fn copy_session(&self, input: copy::Input) -> Result<copy::Output, RequestFailure> {
+ let (operation, value) = wire(&input)
+ .map_err(|e| RequestFailure::NotDispatched(ClientError::Protocol(e.to_string())))?;
+ let output = self.request(operation, value).await?;
+ copy::decode_output(&input, &output).map_err(|e| self.invalid_session(e))
+ }
+
+ /// Read an exact copy's receipt without creating/adopting a target. An absent
+ /// receipt does not prove that an earlier request was never admitted.
+ pub async fn query_session_copy(
+ &self,
+ expected: copy::Input,
+ ) -> Result<copy::QueryResult, RequestFailure> {
+ let output = self
+ .request(
+ Operation::SessionCopyQuery,
+ json!({"targetSessionId":expected.target_session_id}),
+ )
+ .await?;
+ let output = copy::decode_query_result(&output).map_err(|e| self.invalid_session(e))?;
+ if output
+ .receipt
+ .as_ref()
+ .is_some_and(|receipt| receipt.request != expected)
+ {
+ return Err(self.invalid_session(ProtocolError::invalid(
+ "Session copy receipt changed its request",
+ )));
+ }
+ Ok(output)
+ }
+
+ pub async fn abandon_session_revision(
+ &self,
+ input: copy::AbandonInput,
+ ) -> Result<copy::AbandonOutput, RequestFailure> {
+ let output = self
+ .request(
+ Operation::SessionRevisionAbandon,
+ serde_json::to_value(&input).expect("wire input"),
+ )
+ .await?;
+ copy::decode_abandon_output(&input, &output).map_err(|e| self.invalid_session(e))
+ }
+
+ /// Return original inputs, including attachments and selections. A source
+ /// read is not admission validation and must not be flattened into plain text.
+ pub async fn session_turn_sources(
+ &self,
+ input: sources::Input,
+ ) -> Result<sources::Output, RequestFailure> {
+ let output = self
+ .request(
+ Operation::SessionSourcesQuery,
+ serde_json::to_value(&input).expect("wire input"),
+ )
+ .await?;
+ let output = sources::decode_output(&output).map_err(|e| self.invalid_session(e))?;
+ if output.session_id != input.session_id || output.turn_id != input.turn_id {
+ return Err(self.invalid_session(ProtocolError::invalid(
+ "Session sources do not match the selected Turn",
+ )));
+ }
+ Ok(output)
+ }
+}
+
+fn wire(input: ©::Input) -> Result<(Operation, Value), ProtocolError> {
+ let (operation, turn, side) = match &input.purpose {
+ copy::Purpose::Branch {
+ turn_id: Some(turn),
+ side_conversation,
+ } => (
+ Operation::SessionBranchCreate,
+ Some(turn),
+ *side_conversation,
+ ),
+ copy::Purpose::EmptySideConversation => (Operation::SessionBranchCreate, None, true),
+ copy::Purpose::Revision { turn_id } => {
+ (Operation::SessionRevisionCreate, Some(turn_id), false)
+ }
+ // The storage API supports a history-end cut; the public wire API does
+ // not. Never silently convert that into an empty side conversation.
+ copy::Purpose::Branch { turn_id: None, .. } => {
+ return Err(ProtocolError::invalid(
+ "Session branch requires an explicit Turn boundary",
+ ));
+ }
+ };
+ let mut value = json!({
+ "sourceSessionId":input.source_session_id,
+ "targetSessionId":input.target_session_id,
+ "expectedSourceRevision":input.expected_source_revision,
+ });
+ if let Some(turn) = turn {
+ value["sourceTurnId"] = json!(turn);
+ }
+ if side {
+ value["intent"] = json!("side_conversation");
+ }
+ Ok((operation, value))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ #[test]
+ fn wire_preserves_history_cut_and_side_intent_without_coercing_end_into_empty() {
+ let mut input = copy::Input {
+ source_session_id: "source".into(),
+ target_session_id: "target".into(),
+ expected_source_revision: 7,
+ purpose: copy::Purpose::EmptySideConversation,
+ };
+ for purpose in [
+ copy::Purpose::EmptySideConversation,
+ copy::Purpose::Branch {
+ turn_id: Some("turn".into()),
+ side_conversation: false,
+ },
+ copy::Purpose::Branch {
+ turn_id: Some("turn".into()),
+ side_conversation: true,
+ },
+ copy::Purpose::Revision {
+ turn_id: "turn".into(),
+ },
+ ] {
+ input.purpose = purpose;
+ let (op, value) = wire(&input).unwrap();
+ assert_eq!(copy::decode_input(op, &value).unwrap(), input);
+ }
+ for side_conversation in [false, true] {
+ input.purpose = copy::Purpose::Branch {
+ turn_id: None,
+ side_conversation,
+ };
+ assert!(wire(&input).is_err());
+ }
+ }
+}
diff --git a/crates/client/tests/integration/session.rs b/crates/client/tests/integration/session.rs
index 37849161e8..5945d8c96b 100644
--- a/crates/client/tests/integration/session.rs
+++ b/crates/client/tests/integration/session.rs
@@ -23,6 +23,8 @@ use maka_protocol::session::*;
use serde_json::json;
use std::time::Duration;
+mod copy;
+
#[tokio::test]
async fn catalog_rejects_wrong_identity_revision_variant_and_nonadvancing_pages() {
let revision = format!("sha256:{}", "a".repeat(64));
diff --git a/crates/client/tests/integration/session/copy.rs b/crates/client/tests/integration/session/copy.rs
new file mode 100644
index 0000000000..75d567a04c
--- /dev/null
+++ b/crates/client/tests/integration/session/copy.rs
@@ -0,0 +1,96 @@
+/*
+ * 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::super::connection::pair_with;
+use maka_client::{ClientError, RequestFailure};
+use maka_protocol::session::{copy, sources};
+use serde_json::json;
+use std::time::Duration;
+
+#[tokio::test]
+async fn copy_sources_and_abandon_reject_cross_identity_and_changed_copy_intent() {
+ for case in 0..9 {
+ let input = copy::Input {
+ source_session_id: "source".into(),
+ target_session_id: "target".into(),
+ expected_source_revision: 7,
+ purpose: copy::Purpose::Revision {
+ turn_id: "turn".into(),
+ },
+ };
+ let (client, _notices, mut reader, mut writer) = pair_with(maka_client::Operations).await;
+ let request = tokio::spawn({
+ let client = client.clone();
+ let input = input.clone();
+ async move {
+ match case {
+ 0..=4 => client.query_session_copy(input).await.map(|_| ()),
+ 5 => client.copy_session(input).await.map(|_| ()),
+ 6 => client
+ .abandon_session_revision(copy::AbandonInput {
+ target_session_id: "target".into(),
+ })
+ .await
+ .map(|_| ()),
+ _ => client
+ .session_turn_sources(sources::Input {
+ session_id: "source".into(),
+ turn_id: "turn".into(),
+ })
+ .await
+ .map(|_| ()),
+ }
+ }
+ });
+ let frame = reader.read().await.unwrap().unwrap();
+ let result = match case {
+ 0..=4 => {
+ let mut receipt = json!({"receipt":{"request":input,"state":"preparing"}});
+ match case {
+ 0 => receipt["receipt"]["request"]["sourceSessionId"] = json!("other"),
+ 1 => receipt["receipt"]["request"]["targetSessionId"] = json!("other"),
+ 2 => receipt["receipt"]["request"]["expectedSourceRevision"] = json!(8),
+ 3 => receipt["receipt"]["request"]["purpose"]["turnId"] = json!("other"),
+ _ => {
+ receipt["receipt"]["request"]["purpose"] =
+ json!({"kind":"branch","turnId":"turn","sideConversation":true});
+ receipt["receipt"]["state"] = json!("committed");
+ }
+ }
+ receipt
+ }
+ 5 => json!({"kind":"source_revision_conflict","expectedRevision":8,"actualRevision":9}),
+ 6 => json!({"kind":"abandoned","sessionId":"other"}),
+ _ => {
+ json!({"sessionId":if case == 7 {"source"} else {"other"},"turnId":if case == 7 {"other"} else {"turn"},"messages":[{"messageId":"message","content":{"text":"original input"}}]})
+ }
+ };
+ writer.write(&json!({"requestId":frame["requestId"],"operation":frame["operation"],"ok":true,"result":result})).await.unwrap();
+ assert!(
+ matches!(
+ request.await.unwrap(),
+ Err(RequestFailure::Unknown(ClientError::Protocol(_)))
+ ),
+ "case {case}"
+ );
+ tokio::time::timeout(Duration::from_secs(1), client.closed())
+ .await
+ .unwrap();
+ }
+}
diff --git a/crates/runtime-host/tests/integration/session_history.rs b/crates/runtime-host/tests/integration/session_history.rs
index fb5c0de1e6..ddf6cdfe66 100644
--- a/crates/runtime-host/tests/integration/session_history.rs
+++ b/crates/runtime-host/tests/integration/session_history.rs
@@ -19,7 +19,9 @@
use super::support::{client_probe::ClientFixture, peer::Peer};
use maka_event_log::sessions::SessionCopyResult;
-use maka_protocol::session::{RevisionState, SandboxMode, WorkspaceProjection, WorkspaceTarget};
+use maka_protocol::session::{
+ RevisionState, SandboxMode, WorkspaceProjection, WorkspaceTarget, copy, sources,
+};
use maka_runtime::{
event::{EventWrite, Fact, Invocation, InvocationInput, InvocationOutcome, RuntimeEvent},
input::DeliveredMessage,
@@ -171,7 +173,15 @@ async fn catalog_revisions_preserve_branch_origin_without_inheriting_execution_s
// A catalog-only observer must hear retention even without a canonical
// invocation or a Session/PTY subscription driving its connection loop.
let mut observer = Peer::new(host.clone(), "revision-catalog").await;
- let mut operator = Peer::new(host.clone(), "revision-shell").await;
+ let (mut operator, hello) = Peer::handshake(host.clone(), "revision-shell").await;
+ let (client, _notices) = maka_client::Client::connect(
+ maka_client::local::open_stream(&endpoint).await.unwrap(),
+ hello["rootId"].as_str().unwrap(),
+ hello["hostEpoch"].as_str().unwrap(),
+ maka_runtime_host::server::HostOperations,
+ )
+ .await
+ .unwrap();
let started = operator
.rpc(
"runtime.resource.start",
@@ -207,34 +217,35 @@ async fn catalog_revisions_preserve_branch_origin_without_inheriting_execution_s
"sourceSessionId":"source", "targetSessionId":"native-revision",
"sourceTurnId":"turn", "expectedSourceRevision":source["result"]["session"]["revision"]
});
- let absent = operator
- .rpc(
- "session.copy.query",
- json!({"targetSessionId":"native-revision"}),
- )
- .await;
- assert_eq!(absent["result"], json!({"receipt":null}), "{absent}");
- let copied = operator
- .rpc("session.revision.create", request.clone())
- .await;
- assert_eq!(copied["ok"], true, "{copied}");
- assert_eq!(copied["result"]["kind"], "committed");
- assert_eq!(copied["result"]["session"]["revisionState"], "preparing");
+ let input =
+ copy::decode_input(maka_protocol::Operation::SessionRevisionCreate, &request).unwrap();
+ assert!(
+ client
+ .query_session_copy(input.clone())
+ .await
+ .unwrap()
+ .receipt
+ .is_none()
+ );
+ let copied = client.copy_session(input.clone()).await.unwrap();
+ assert!(
+ matches!(&copied, copy::Output::Committed { session } if session.revision_state == Some(RevisionState::Preparing))
+ );
for session in ["source", "branch", "native-revision"] {
- let sources = operator
- .rpc(
- "session.sources.query",
- json!({"sessionId":session, "turnId":"turn"}),
- )
- .await;
+ let sources = client
+ .session_turn_sources(sources::Input {
+ session_id: session.into(),
+ turn_id: "turn".into(),
+ })
+ .await
+ .unwrap();
assert_eq!(
- sources["result"],
+ serde_json::to_value(sources).unwrap(),
json!({
"sessionId":session, "turnId":"turn", "messages":[{
"messageId":"message", "content":{"text":"unprepared source"}
}]
- }),
- "{sources}"
+ })
);
}
let renamed = operator
@@ -247,26 +258,18 @@ async fn catalog_revisions_preserve_branch_origin_without_inheriting_execution_s
)
.await;
assert_eq!(renamed["result"]["kind"], "committed", "{renamed}");
- let receipt = operator
- .rpc(
- "session.copy.query",
- json!({"targetSessionId":"native-revision"}),
- )
- .await;
+ let receipt = client
+ .query_session_copy(input.clone())
+ .await
+ .unwrap()
+ .receipt
+ .unwrap();
assert_eq!(
- receipt["result"],
- json!({"receipt": {
- "request": {
- "sourceSessionId":"source", "targetSessionId":"native-revision",
- "expectedSourceRevision":request["expectedSourceRevision"],
- "purpose":{"kind":"revision", "turnId":"turn"}
- }, "state":"preparing"
- }}),
- "{receipt}"
+ receipt.request, input,
+ "receipt preserves its original source revision after a concurrent edit"
);
- let replayed = operator
- .rpc("session.revision.create", request.clone())
- .await;
+ assert_eq!(receipt.state, maka_runtime::session::CopyState::Preparing);
+ let replayed = client.copy_session(input.clone()).await.unwrap();
assert_eq!(
replayed, copied,
"a lost reply must not recapture changed source settings"
@@ -280,37 +283,52 @@ async fn catalog_revisions_preserve_branch_origin_without_inheriting_execution_s
let conflict = operator.rpc("session.revision.create", stale).await;
assert_eq!(conflict["result"]["kind"], "source_revision_conflict");
for _ in 0..2 {
- let abandoned = operator
- .rpc(
- "session.revision.abandon",
- json!({"targetSessionId":"native-revision"}),
- )
- .await;
- assert_eq!(abandoned["result"]["kind"], "abandoned", "{abandoned}");
+ let abandoned = client
+ .abandon_session_revision(copy::AbandonInput {
+ target_session_id: "native-revision".into(),
+ })
+ .await
+ .unwrap();
+ assert_eq!(
+ abandoned,
+ copy::AbandonOutput::Abandoned {
+ session_id: "native-revision".into()
+ }
+ );
}
let abandoned_retry = operator.rpc("session.revision.create", request).await;
assert_eq!(abandoned_retry["error"]["code"], "not_found");
- let tombstone = operator
- .rpc(
- "session.copy.query",
- json!({"targetSessionId":"native-revision"}),
- )
- .await;
- let mut expected = receipt["result"].clone();
- expected["receipt"]["state"] = json!("abandoned");
- assert_eq!(tombstone["result"], expected, "{tombstone}");
- let retained = operator
- .rpc(
- "session.revision.abandon",
- json!({"targetSessionId":"revision"}),
- )
- .await;
- assert_eq!(retained["result"]["kind"], "retained", "{retained}");
- let side = operator.rpc("session.branch.create", json!({
- "sourceSessionId":"source", "targetSessionId":"empty-side",
- "expectedSourceRevision":renamed["result"]["session"]["revision"], "intent":"side_conversation"
- })).await;
- assert_eq!(side["result"]["kind"], "committed", "{side}");
+ let tombstone = client
+ .query_session_copy(input)
+ .await
+ .unwrap()
+ .receipt
+ .unwrap();
+ assert_eq!(tombstone.request, receipt.request);
+ assert_eq!(tombstone.state, maka_runtime::session::CopyState::Abandoned);
+ let retained = client
+ .abandon_session_revision(copy::AbandonInput {
+ target_session_id: "revision".into(),
+ })
+ .await
+ .unwrap();
+ assert_eq!(
+ retained,
+ copy::AbandonOutput::Retained {
+ session_id: "revision".into()
+ }
+ );
+ let side = client
+ .copy_session(copy::Input {
+ source_session_id: "source".into(),
+ target_session_id: "empty-side".into(),
+ expected_source_revision: renamed["result"]["session"]["revision"].as_u64().unwrap(),
+ purpose: copy::Purpose::EmptySideConversation,
+ })
+ .await
+ .unwrap();
+ assert!(matches!(side, copy::Output::Committed { .. }));
+ client.disconnect();
observer.close().await;
operator.close().await;
stop.cancel();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org