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 18:04:20 UTC
(maka) branch feat/runtime-host-rust updated: test(host): verify attachment consumption through native client
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 c42431ca9e test(host): verify attachment consumption through native client
new 1ab4cb6af7 merge: sync configurable native Jev plugin
c42431ca9e is described below
commit c42431ca9e0a7adc0f43ec292c67d990ea3ba1f0
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Thu Sep 24 02:01:07 2026 +0800
test(host): verify attachment consumption through native client
Register typed single-turn admission and preserve native image, large-result, exact-replay and restart coverage.
Generated-by: Codex
---
crates/client/src/operations.rs | 8 +-
crates/client/src/turn.rs | 22 ++
crates/client/tests/integration/turn.rs | 40 +--
.../tests/integration/attachment_consumption.rs | 72 ++----
.../integration/attachment_consumption/workflow.rs | 234 +++++++++++++++++
.../tests/integration/large_outputs.rs | 69 ++---
.../tests/integration/large_outputs/workflow.rs | 266 +++++++++++++++++++
crates/runtime-host/tests/integration/main.rs | 2 +
.../tests/integration/support/attachment_client.rs | 286 +++++++++++++++++++++
tests/fixtures/client-artifact-upload.mjs | 39 ---
tests/fixtures/client-attachment-model.mjs | 120 ---------
tests/fixtures/client-attachment-workflow.mjs | 181 -------------
tests/fixtures/client-entry.mjs | 16 +-
tests/fixtures/client-large-output-fixture.mjs | 115 ---------
tests/fixtures/client-large-output-workflow.mjs | 281 --------------------
tests/fixtures/client.mjs | 15 +-
16 files changed, 885 insertions(+), 881 deletions(-)
diff --git a/crates/client/src/operations.rs b/crates/client/src/operations.rs
index c8cfa271c9..414588b18e 100644
--- a/crates/client/src/operations.rs
+++ b/crates/client/src/operations.rs
@@ -103,6 +103,10 @@ impl OperationRegistry for Operations {
)?)
.map_err(|error| maka_protocol::ProtocolError::invalid(error.to_string()));
}
+ if operation == Operation::TurnStart {
+ return serde_json::to_value(maka_protocol::turn::decode_turn_start_input(value)?)
+ .map_err(|error| maka_protocol::ProtocolError::invalid(error.to_string()));
+ }
if operation == Operation::TurnQuery {
maka_protocol::turn::decode_turn_query_input(value)?;
return Ok(value.clone());
@@ -184,7 +188,7 @@ impl OperationRegistry for Operations {
maka_protocol::turn::decode_turn_snapshot(value)?;
return Ok(value.clone());
}
- if operation == Operation::TurnBatchStart {
+ if matches!(operation, Operation::TurnStart | Operation::TurnBatchStart) {
maka_protocol::turn::decode_turn_start_result(value)?;
return Ok(value.clone());
}
@@ -258,7 +262,7 @@ impl OperationRegistry for Operations {
if operation == Operation::TurnStop {
return Some(maka_protocol::turn::STOP_ERRORS);
}
- if operation == Operation::TurnBatchStart {
+ if matches!(operation, Operation::TurnStart | Operation::TurnBatchStart) {
return Some(maka_protocol::turn::START_ERRORS);
}
if operation == Operation::TurnQuery {
diff --git a/crates/client/src/turn.rs b/crates/client/src/turn.rs
index ed0c85a857..4dc5012d13 100644
--- a/crates/client/src/turn.rs
+++ b/crates/client/src/turn.rs
@@ -21,6 +21,28 @@ use crate::{Client, ClientError, RequestFailure};
use maka_protocol::{Operation, turn::*};
impl Client {
+ /// Start once with the caller's stable identity; unknown outcomes are not retried.
+ pub async fn start_turn(
+ &self,
+ input: TurnStartInput,
+ ) -> Result<TurnStartResult, RequestFailure> {
+ let output = self
+ .request(
+ Operation::TurnStart,
+ serde_json::to_value(&input).expect("wire input"),
+ )
+ .await?;
+ if let Ok(output) = decode_turn_start_result(&output)
+ && assert_start_output_for_input(&input, &output).is_ok()
+ {
+ return Ok(output);
+ }
+ self.disconnect();
+ Err(RequestFailure::Unknown(ClientError::Protocol(
+ "Start receipt does not match the requested Turn".into(),
+ )))
+ }
+
/// Submit a frozen batch once. Unknown outcomes require an explicit query;
/// this method never retries or allocates another Turn identity.
pub async fn start_turn_batch(
diff --git a/crates/client/tests/integration/turn.rs b/crates/client/tests/integration/turn.rs
index 71bca11f2d..25a54c5b61 100644
--- a/crates/client/tests/integration/turn.rs
+++ b/crates/client/tests/integration/turn.rs
@@ -24,41 +24,47 @@ use serde_json::json;
use std::time::Duration;
#[tokio::test]
-async fn batch_and_query_bind_receipts_to_the_exact_turn_without_automatic_replay() {
- for query in [false, true] {
+async fn starts_and_query_bind_receipts_to_the_exact_turn_without_automatic_replay() {
+ for operation in ["turn.start", "turn.batch.start", "turn.query"] {
for changed in [None, Some("sessionId"), Some("turnId")] {
let (client, _notices, mut reader, mut writer) =
pair_with(maka_client::Operations).await;
let input = json!({"sessionId":"session","turnId":"turn","messages":[
{"content":{"text":"first"}}, {"content":{"text":"second"},"inputSelections":{"example":["chosen"]}}
]});
+ let input = if operation == "turn.start" {
+ json!({"sessionId":"session","turnId":"turn","content":{"text":"first"}})
+ } else {
+ input
+ };
let task = tokio::spawn({
let client = client.clone();
- let input = decode_turn_batch_start_input(&input).unwrap();
+ let input = input.clone();
async move {
- if query {
+ if operation == "turn.query" {
client
.query_turn(TurnQueryInput {
- session_id: input.session_id,
- turn_id: input.turn_id,
+ session_id: "session".into(),
+ turn_id: "turn".into(),
})
.await
.map(|_| ())
+ } else if operation == "turn.start" {
+ client
+ .start_turn(decode_turn_start_input(&input).unwrap())
+ .await
+ .map(|_| ())
} else {
- client.start_turn_batch(input).await.map(|_| ())
+ client
+ .start_turn_batch(decode_turn_batch_start_input(&input).unwrap())
+ .await
+ .map(|_| ())
}
}
});
let frame = reader.read().await.unwrap().unwrap();
- assert_eq!(
- frame["operation"],
- if query {
- "turn.query"
- } else {
- "turn.batch.start"
- }
- );
- if !query {
+ assert_eq!(frame["operation"], operation);
+ if operation != "turn.query" {
assert_eq!(frame["input"], input);
}
let mut turn =
@@ -66,7 +72,7 @@ async fn batch_and_query_bind_receipts_to_the_exact_turn_without_automatic_repla
if let Some(field) = changed {
turn[field] = json!("other");
}
- let result = if query {
+ let result = if operation == "turn.query" {
turn
} else {
json!({"kind":"started","turn":turn,"preparation":[]})
diff --git a/crates/runtime-host/tests/integration/attachment_consumption.rs b/crates/runtime-host/tests/integration/attachment_consumption.rs
index 8ad82c57d2..93ceb6b50b 100644
--- a/crates/runtime-host/tests/integration/attachment_consumption.rs
+++ b/crates/runtime-host/tests/integration/attachment_consumption.rs
@@ -18,6 +18,7 @@
*/
#![cfg(unix)]
+use super::support::attachment_client::NativeHost;
use maka_event_log::{
EventLog,
root::{ROOT_DATABASE, RootNamespaces, RootOwner},
@@ -27,12 +28,11 @@ use maka_runtime::{
input::InvocationInput,
tool_output::ToolOutput,
};
-use maka_runtime_host::server::{Host, local::LocalListener};
-use std::{os::unix::fs::PermissionsExt, path::Path, process::Command, time::Duration};
-use tokio_util::sync::CancellationToken;
+use std::{os::unix::fs::PermissionsExt, time::Duration};
+mod workflow;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn original_client_consumes_uploads_with_selected_vision_and_reopens_exact_facts() {
+async fn native_client_consumes_uploads_with_selected_vision_and_reopens_exact_facts() {
let directory = tempfile::Builder::new()
.prefix("maka-attachments-")
.permissions(std::fs::Permissions::from_mode(0o700))
@@ -44,60 +44,22 @@ async fn original_client_consumes_uploads_with_selected_vision_and_reopens_exact
};
let root = directory.path().join("root");
let owner = RootOwner::create(&root, &ns).unwrap();
- let root_id = owner.root_id().to_owned();
drop(owner);
let mut original = None;
+ let mut saved = Vec::new();
for reopened in [false, true] {
- let host = Host::open(RootOwner::open(&root, &ns).unwrap())
- .await
- .unwrap();
- let socket = directory.path().join("h.sock");
- let cancellation = CancellationToken::new();
- let server = tokio::spawn(
- LocalListener::bind(&socket)
- .unwrap()
- .serve(host, cancellation.clone()),
- );
- let probe = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/client.mjs");
- let workspace = directory.path().to_owned();
- let expected_id = root_id.clone();
- let client = tokio::task::spawn_blocking(move || {
- let mut command = Command::new("node");
- command
- .arg(probe)
- .arg("--socket")
- .arg(socket)
- .args(["--root-id", &expected_id])
- .arg("--attachment-workspace")
- .arg(workspace);
- if reopened {
- command.arg("--reopened");
- }
- command.output().unwrap()
- });
- let output = tokio::time::timeout(Duration::from_secs(30), client)
- .await
- .unwrap()
- .unwrap();
- cancellation.cancel();
- tokio::time::timeout(Duration::from_secs(5), server)
- .await
- .unwrap()
- .unwrap()
- .unwrap();
- assert!(
- output.status.success(),
- "stdout: {}\nstderr: {}",
- String::from_utf8_lossy(&output.stdout),
- String::from_utf8_lossy(&output.stderr)
- );
- assert!(
- String::from_utf8_lossy(&output.stdout).contains(if reopened {
- "attachment-consumption-reopened"
- } else {
- "attachment-consumption-passed"
- })
- );
+ let native = NativeHost::open(
+ RootOwner::open(&root, &ns).unwrap(),
+ &directory.path().join("h.sock"),
+ )
+ .await;
+ tokio::time::timeout(
+ Duration::from_secs(30),
+ workflow::verify(&native.client, directory.path(), reopened, &mut saved),
+ )
+ .await
+ .unwrap();
+ native.close().await;
let log = EventLog::open(&root.join(ROOT_DATABASE)).await.unwrap();
let prefix = log.prefix(200, 2 * 1024 * 1024).await.unwrap();
let mut opened = 0;
diff --git a/crates/runtime-host/tests/integration/attachment_consumption/workflow.rs b/crates/runtime-host/tests/integration/attachment_consumption/workflow.rs
new file mode 100644
index 0000000000..624f377e90
--- /dev/null
+++ b/crates/runtime-host/tests/integration/attachment_consumption/workflow.rs
@@ -0,0 +1,234 @@
+/*
+ * 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::support::attachment_client as support;
+use base64::{Engine as _, engine::general_purpose::STANDARD};
+use maka_client::Client;
+use maka_protocol::{
+ Operation, OperationErrorCode as Code,
+ artifact::{ArtifactDeleteInput, ArtifactQueryInput, ArtifactQueryResult},
+};
+use maka_runtime::artifact::upload_artifact_id;
+use serde_json::{Value, json};
+use std::path::Path;
+
+const TEXT: &str = "uploaded text 😀, outside workspace authority";
+const PNG: &str =
+ "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aWQAAAABJRU5ErkJggg==";
+pub(super) struct Saved {
+ turn: Value,
+ terminal: Value,
+ rows: Vec<Value>,
+}
+pub(super) async fn verify(
+ client: &Client,
+ workspace: &Path,
+ reopened: bool,
+ saved: &mut Vec<Saved>,
+) {
+ if reopened {
+ for item in saved {
+ assert_eq!(
+ support::rows(client, item.turn["sessionId"].as_str().unwrap()).await,
+ item.rows
+ );
+ assert_eq!(
+ client
+ .request(Operation::TurnStart, item.turn.clone())
+ .await
+ .unwrap()["turn"],
+ item.terminal
+ );
+ for attachment in item.turn["content"]["attachments"].as_array().unwrap() {
+ let ArtifactQueryResult::Artifact { artifact, .. } = client
+ .query_artifact(ArtifactQueryInput::Get {
+ session_id: item.turn["sessionId"].as_str().unwrap().into(),
+ artifact_id: attachment["ref"]["relativePath"].as_str().unwrap().into(),
+ })
+ .await
+ .unwrap()
+ else {
+ panic!()
+ };
+ assert!(artifact.is_none());
+ }
+ }
+ return;
+ }
+ let model = support::Model::start(6, |index, input| {
+ let vision = index >= 3;
+ let step = index % 3 + 1;
+ assert_eq!(input["model"], if vision { "gpt-4o" } else { "text-model" });
+ let session = if vision {
+ "vision-session"
+ } else {
+ "text-session"
+ };
+ let ids = [
+ upload_artifact_id(session, "text"),
+ upload_artifact_id(session, "image"),
+ ];
+ let messages = input["messages"].as_array().unwrap();
+ let serialized = serde_json::to_string(messages).unwrap();
+ for id in &ids {
+ assert!(serialized.contains(&format!("maka://runtime/attachments/{id}")));
+ }
+ if vision {
+ assert_eq!(
+ serialized
+ .matches(&format!("data:image/png;base64,{PNG}"))
+ .count(),
+ if step == 3 { 2 } else { 1 }
+ );
+ } else {
+ assert!(!serialized.contains(PNG));
+ assert!(!serialized.contains("image_url"));
+ let user = messages.iter().find(|m| m["role"] == "user").unwrap()["content"]
+ .as_str()
+ .unwrap();
+ assert!(user.contains("consume uploaded resources"));
+ assert!(user.contains("note.txt") && user.contains("pixel.png"));
+ }
+ let tools: Vec<_> = messages.iter().filter(|m| m["role"] == "tool").collect();
+ if step == 2 {
+ assert_eq!(
+ serde_json::from_str::<Value>(tools.last().unwrap()["content"].as_str().unwrap())
+ .unwrap(),
+ json!({"content":TEXT,"offset":0,"returnedLines":1,"totalLines":1,"next":null})
+ );
+ }
+ if step == 3 && vision {
+ assert!(
+ tools
+ .iter()
+ .all(|m| !m["content"].to_string().contains(PNG))
+ );
+ let tool_index = messages.iter().rposition(|m| m["role"] == "tool").unwrap();
+ assert!(messages[tool_index + 1..].iter().any(|m| {
+ m["role"] == "user"
+ && m["content"]
+ .as_array()
+ .is_some_and(|parts| parts.iter().any(|p| p["type"] == "image_url"))
+ }));
+ }
+ if step < 3 {
+ support::read(&format!("read-{step}"), &ids[step - 1])
+ } else {
+ json!({"content":"attachment complete"})
+ }
+ })
+ .await;
+ let connection = support::configure(
+ client,
+ &model.url,
+ json!({
+ "text-model":{"vision":false},"gpt-4o":{"vision":true,"codeMode":false}
+ }),
+ )
+ .await;
+ for (session, model_id) in [("text-session", "text-model"), ("vision-session", "gpt-4o")] {
+ support::session(client, &connection, workspace, session, model_id).await;
+ let attachments = vec![
+ support::upload(
+ client,
+ session,
+ "text",
+ "note.txt",
+ "text/plain",
+ TEXT.as_bytes(),
+ )
+ .await,
+ support::upload(
+ client,
+ session,
+ "image",
+ "pixel.png",
+ "image/png",
+ &STANDARD.decode(PNG).unwrap(),
+ )
+ .await,
+ ];
+ let attachments = serde_json::to_value(attachments).unwrap();
+ let turn = json!({"sessionId":session,"turnId":"consume","maxSteps":4,
+ "content":{"text":"consume uploaded resources","attachments":attachments}});
+ for (index, field, value) in [
+ (0, "name", json!("forged.txt")),
+ (0, "bytes", json!(TEXT.len() + 1)),
+ (0, "mimeType", json!("application/json")),
+ (1, "kind", json!("other")),
+ (
+ 0,
+ "ref",
+ json!({"kind":"session_file","sessionId":"other-session","relativePath":attachments[0]["ref"]["relativePath"]}),
+ ),
+ ] {
+ let mut corrupt = attachments[index].clone();
+ corrupt[field] = value;
+ support::rejected(client.request(Operation::TurnStart,json!({
+ "sessionId":session,"turnId":"rejected","maxSteps":4,"content":{"text":"consume uploaded resources","attachments":[corrupt]}
+ })).await,Code::OperationConflict);
+ support::rejected(
+ client
+ .request(
+ Operation::TurnQuery,
+ json!({"sessionId":session,"turnId":"rejected"}),
+ )
+ .await,
+ Code::NotFound,
+ );
+ }
+ let terminal = support::completed(client, &turn).await;
+ let rows = support::rows(client, session).await;
+ assert_eq!(
+ rows.iter().find(|r| r["type"] == "user").unwrap()["attachments"],
+ attachments
+ );
+ let results: Vec<_> = rows.iter().filter(|r| r["type"] == "tool_result").collect();
+ assert_eq!(results.len(), 2);
+ assert!(results.iter().all(|r| r["isError"] != true));
+ assert_eq!(results[0]["content"], json!({"kind":"text","text":TEXT}));
+ assert_eq!(
+ results[1]["content"],
+ json!({"kind":"image","mimeType":"image/png","ref":attachments[1]["ref"]})
+ );
+ assert_eq!(results[1]["content"]["ref"]["sessionId"], session);
+ for attachment in attachments.as_array().unwrap() {
+ client
+ .delete_artifact(ArtifactDeleteInput {
+ session_id: session.into(),
+ artifact_id: attachment["ref"]["relativePath"].as_str().unwrap().into(),
+ })
+ .await
+ .unwrap();
+ }
+ assert_eq!(
+ client
+ .request(Operation::TurnStart, turn.clone())
+ .await
+ .unwrap()["turn"],
+ terminal
+ );
+ saved.push(Saved {
+ turn,
+ terminal,
+ rows,
+ });
+ }
+ model.finish().await;
+}
diff --git a/crates/runtime-host/tests/integration/large_outputs.rs b/crates/runtime-host/tests/integration/large_outputs.rs
index 7be8cc33bf..b08f297476 100644
--- a/crates/runtime-host/tests/integration/large_outputs.rs
+++ b/crates/runtime-host/tests/integration/large_outputs.rs
@@ -19,6 +19,7 @@
#![cfg(unix)]
+use super::support::attachment_client::NativeHost;
use maka_event_log::{
EventLog,
root::{RootNamespaces, RootOwner},
@@ -27,9 +28,8 @@ use maka_runtime::{
event::{Fact, ToolOutcome},
tool_output::{DurableToolProjection, ToolOutput},
};
-use maka_runtime_host::server::{Host, local::LocalListener};
-use std::{os::unix::fs::PermissionsExt, path::Path, process::Command, time::Duration};
-use tokio_util::sync::CancellationToken;
+use std::{os::unix::fs::PermissionsExt, time::Duration};
+mod workflow;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn large_attachment_results_preserve_raw_without_poisoning_model_history() {
@@ -46,57 +46,23 @@ async fn large_attachment_results_preserve_raw_without_poisoning_model_history()
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let owner = RootOwner::create(&root, &ns).unwrap();
- let root_id = owner.root_id().to_owned();
drop(owner);
let mut original = None;
+ let mut saved = Vec::new();
+ let model = workflow::model().await;
for reopened in [false, true] {
- let host = Host::open(RootOwner::open(&root, &ns).unwrap())
- .await
- .unwrap();
- let socket = directory.path().join("h.sock");
- let listener = LocalListener::bind(&socket).unwrap();
- let cancellation = CancellationToken::new();
- let server = tokio::spawn(listener.serve(host, cancellation.clone()));
- let probe = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/client.mjs");
- let client_workspace = workspace.clone();
- let expected_id = root_id.clone();
- let client = tokio::task::spawn_blocking(move || {
- let mut command = Command::new("node");
- command
- .arg(probe)
- .arg("--socket")
- .arg(socket)
- .args(["--root-id", &expected_id])
- .arg("--large-output-workspace")
- .arg(client_workspace);
- if reopened {
- command.arg("--reopened");
- }
- command.output().unwrap()
- });
- let output = tokio::time::timeout(Duration::from_secs(150), client)
- .await
- .unwrap()
- .unwrap();
- cancellation.cancel();
- tokio::time::timeout(Duration::from_secs(10), server)
- .await
- .unwrap()
- .unwrap()
- .unwrap();
- assert!(
- output.status.success(),
- "stdout: {}\nstderr: {}",
- String::from_utf8_lossy(&output.stdout),
- String::from_utf8_lossy(&output.stderr)
- );
- assert!(
- String::from_utf8_lossy(&output.stdout).contains(if reopened {
- "large-output-reopened"
- } else {
- "large-output-passed"
- })
- );
+ let mut native = NativeHost::open(
+ RootOwner::open(&root, &ns).unwrap(),
+ &directory.path().join("h.sock"),
+ )
+ .await;
+ tokio::time::timeout(
+ Duration::from_secs(150),
+ workflow::verify(&mut native, &workspace, &model.url, reopened, &mut saved),
+ )
+ .await
+ .unwrap();
+ native.close().await;
let log = EventLog::open(&root.join(maka_event_log::root::ROOT_DATABASE))
.await
@@ -180,4 +146,5 @@ async fn large_attachment_results_preserve_raw_without_poisoning_model_history()
}
log.close().await.unwrap();
}
+ model.finish().await;
}
diff --git a/crates/runtime-host/tests/integration/large_outputs/workflow.rs b/crates/runtime-host/tests/integration/large_outputs/workflow.rs
new file mode 100644
index 0000000000..4f8f7c98eb
--- /dev/null
+++ b/crates/runtime-host/tests/integration/large_outputs/workflow.rs
@@ -0,0 +1,266 @@
+/*
+ * 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::support::attachment_client::{self as support, NativeHost};
+use maka_client::{Client, Notification};
+use maka_protocol::{
+ Operation, OperationErrorCode as Code,
+ subscription::{SubscriptionOpenInput, TranscriptPolicy},
+ transcript::{SessionTranscriptPageDirection, SessionTranscriptPageInput},
+};
+use maka_runtime::artifact::{content_digest, upload_artifact_id};
+use serde_json::{Value, json};
+use std::{path::Path, time::Duration};
+
+const SIZE: usize = 10 * 1024 * 1024;
+pub(super) struct Saved {
+ turn: Value,
+ terminal: Value,
+ digest: String,
+ row_id: Option<Value>,
+}
+pub(super) async fn model() -> support::Model {
+ support::Model::start(8, |index, input| {
+ let case = if [0, 1, 4, 6].contains(&index) {
+ "ascii"
+ } else {
+ "control"
+ };
+ assert_eq!(input["model"], format!("{case}-model"));
+ let tools: Vec<_> = input["messages"]
+ .as_array()
+ .unwrap()
+ .iter()
+ .filter(|m| m["role"] == "tool")
+ .collect();
+ if index == 0 || index == 2 {
+ assert!(tools.is_empty());
+ support::read(
+ "read-large",
+ &upload_artifact_id(&format!("large-output-{case}"), "large-text"),
+ )
+ } else {
+ assert_eq!(tools.len(), 1);
+ assert_eq!(tools[0]["tool_call_id"], "read-large");
+ let content = tools[0]["content"].as_str().unwrap();
+ assert!(content.encode_utf16().count() <= 7500);
+ let page: Value = serde_json::from_str(content).unwrap();
+ assert!(!page["content"].as_str().unwrap().is_empty());
+ assert_eq!(page["partialLine"], true);
+ assert_eq!(page["returnedLines"], 0);
+ assert_eq!(page["totalLines"], 1);
+ assert!(
+ page["next"]["path"]
+ .as_str()
+ .unwrap()
+ .starts_with("maka://read/")
+ );
+ json!({"content":"large output complete"})
+ }
+ })
+ .await
+}
+pub(super) async fn verify(
+ native: &mut NativeHost,
+ workspace: &Path,
+ url: &str,
+ reopened: bool,
+ saved: &mut Vec<Saved>,
+) {
+ let client = &native.client;
+ let mut live = None;
+ if !reopened {
+ let connection = support::configure(
+ client,
+ url,
+ json!({
+ "ascii-model":{"vision":false},"control-model":{"vision":false}
+ }),
+ )
+ .await;
+ for (case, byte) in [("ascii", 65), ("control", 1)] {
+ let session = format!("large-output-{case}");
+ support::session(
+ client,
+ &connection,
+ workspace,
+ &session,
+ &format!("{case}-model"),
+ )
+ .await;
+ let bytes = vec![byte; SIZE];
+ let attachment = support::upload(
+ client,
+ &session,
+ "large-text",
+ &format!("{case}.txt"),
+ "text/plain",
+ &bytes,
+ )
+ .await;
+ let turn = json!({"sessionId":session,"turnId":"read-large","maxSteps":3,
+ "content":{"text":"Read the uploaded text","attachments":[attachment]}});
+ if case == "control" {
+ let open = client
+ .open_subscription(SubscriptionOpenInput {
+ session_id: session.clone(),
+ transcript: TranscriptPolicy::Tail { max_bytes: 2 },
+ })
+ .await
+ .unwrap();
+ client
+ .ready_subscription(&open.subscription_id)
+ .await
+ .unwrap();
+ live = Some(open.subscription_id);
+ }
+ let terminal = support::completed(client, &turn).await;
+ if case == "control" {
+ capacity(client, &session, live.as_deref()).await;
+ let basis = client
+ .request(
+ Operation::SessionCatalogQuery,
+ json!({"kind":"get","sessionId":session}),
+ )
+ .await
+ .unwrap();
+ let changed = client.request(Operation::SessionMetadataUpdate,json!({
+ "sessionId":session,"expectedRevision":basis["session"]["revision"],"patch":{"name":"Capacity remains observable"}
+ })).await.unwrap();
+ assert_eq!(changed["kind"], "committed");
+ tokio::time::timeout(Duration::from_secs(5), async {
+ loop {
+ let notice = native.notices.recv().await.unwrap();
+ if let Notification::Observation(frame) = notice
+ && let maka_protocol::subscription::ObservationFrame::Projection(frame) = *frame
+ {
+ let maka_protocol::subscription::SessionProjectionFrame::SessionProjection {
+ subscription_id, snapshot, ..
+ } = *frame;
+ if Some(subscription_id.as_str()) == live.as_deref()
+ && Some(snapshot.session.metadata_revision) == changed["session"]["revision"].as_u64() {
+ break;
+ }
+ }
+ }
+ })
+ .await
+ .unwrap();
+ }
+ saved.push(Saved {
+ turn,
+ terminal,
+ digest: content_digest(&bytes),
+ row_id: None,
+ });
+ }
+ }
+ for item in saved {
+ let session = item.turn["sessionId"].as_str().unwrap();
+ assert_eq!(
+ client
+ .request(
+ Operation::TurnQuery,
+ json!({"sessionId":session,"turnId":"read-large"})
+ )
+ .await
+ .unwrap(),
+ item.terminal
+ );
+ assert_eq!(
+ client
+ .request(Operation::TurnStart, item.turn.clone())
+ .await
+ .unwrap()["turn"],
+ item.terminal
+ );
+ if session.ends_with("ascii") {
+ let rows = support::rows(client, session).await;
+ let results: Vec<_> = rows.iter().filter(|r| r["type"] == "tool_result").collect();
+ assert_eq!(results.len(), 1, "no replayed effect");
+ let row = results[0];
+ assert_eq!(row["isError"], false);
+ assert_eq!(row["content"]["kind"], "text");
+ let text = row["content"]["text"].as_str().unwrap();
+ assert_eq!(text.len(), SIZE);
+ assert_eq!(content_digest(text.as_bytes()), item.digest);
+ if let Some(id) = &item.row_id {
+ assert_eq!(id, &row["id"]);
+ } else {
+ item.row_id = Some(row["id"].clone());
+ }
+ } else {
+ capacity(client, session, live.as_deref()).await;
+ }
+ support::completed(
+ client,
+ &json!({"sessionId":session,
+ "turnId":if reopened {"after-reopen"} else {"after-capacity"},
+ "content":{"text":"Confirm completion"},"maxSteps":1
+ }),
+ )
+ .await;
+ if session.ends_with("control") {
+ capacity(client, session, live.as_deref()).await;
+ }
+ assert_eq!(
+ client
+ .request(Operation::HostStatus, json!({}))
+ .await
+ .unwrap()["state"],
+ "ready"
+ );
+ }
+ if let Some(id) = live {
+ client.close_subscription(&id).await.unwrap();
+ }
+}
+async fn capacity(client: &Client, session: &str, live: Option<&str>) {
+ if let Some(id) = live {
+ support::rejected(
+ client
+ .transcript_page(SessionTranscriptPageInput {
+ subscription_id: id.into(),
+ direction: SessionTranscriptPageDirection::Older,
+ through_sequence: None,
+ cursor: None,
+ anchor_sequence: None,
+ max_bytes: 48 * 1024,
+ })
+ .await,
+ Code::OperationUnavailable,
+ );
+ }
+ support::rejected(
+ client
+ .open_subscription(SubscriptionOpenInput {
+ session_id: session.into(),
+ transcript: TranscriptPolicy::Tail { max_bytes: 2 },
+ })
+ .await,
+ Code::OperationUnavailable,
+ );
+ assert_eq!(
+ client
+ .request(Operation::HostStatus, json!({}))
+ .await
+ .unwrap()["state"],
+ "ready"
+ );
+}
diff --git a/crates/runtime-host/tests/integration/main.rs b/crates/runtime-host/tests/integration/main.rs
index 9550c761dd..79b9de90c9 100644
--- a/crates/runtime-host/tests/integration/main.rs
+++ b/crates/runtime-host/tests/integration/main.rs
@@ -96,6 +96,8 @@ mod workhub_public;
mod workspace_images;
mod support {
+ #[cfg(unix)]
+ pub(crate) mod attachment_client;
pub(crate) mod client_probe;
#[cfg(unix)]
pub(crate) mod execution_drain;
diff --git a/crates/runtime-host/tests/integration/support/attachment_client.rs b/crates/runtime-host/tests/integration/support/attachment_client.rs
new file mode 100644
index 0000000000..1ac987b14a
--- /dev/null
+++ b/crates/runtime-host/tests/integration/support/attachment_client.rs
@@ -0,0 +1,286 @@
+/*
+ * 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 base64::{Engine as _, engine::general_purpose::STANDARD};
+use maka_client::{Client, ClientError, RequestFailure};
+use maka_event_log::root::RootOwner;
+use maka_protocol::{
+ Operation, OperationErrorCode,
+ artifact::{ArtifactIngestInput as Ingest, ArtifactIngestResult},
+ subscription::{SubscriptionOpenInput, TranscriptPolicy},
+ transcript::{SessionTranscriptPageDirection, SessionTranscriptPageInput},
+ turn::AttachmentRef,
+};
+use maka_runtime_host::server::{Host, local::LocalListener};
+use serde_json::{Value, json};
+use std::{path::Path, sync::Arc, time::Duration};
+use tokio_util::sync::CancellationToken;
+
+pub struct NativeHost {
+ pub client: Client,
+ pub notices: tokio::sync::mpsc::Receiver<maka_client::Notification>,
+ cancel: CancellationToken,
+ server: tokio::task::JoinHandle<Result<(), maka_runtime_host::server::HostError>>,
+}
+impl NativeHost {
+ pub async fn open(owner: RootOwner, socket: &Path) -> Self {
+ let root = owner.root_id().to_owned();
+ let host = Host::open(owner).await.unwrap();
+ let (peer, hello) =
+ super::peer::Peer::handshake(host.clone(), "attachments-bootstrap").await;
+ peer.close().await;
+ let cancel = CancellationToken::new();
+ let server = tokio::spawn(
+ LocalListener::bind(socket)
+ .unwrap()
+ .serve(host, cancel.clone()),
+ );
+ let (client, notices) = Client::connect(
+ maka_client::local::open_stream(socket).await.unwrap(),
+ &root,
+ hello["hostEpoch"].as_str().unwrap(),
+ maka_client::Operations,
+ )
+ .await
+ .unwrap();
+ Self {
+ client,
+ notices,
+ cancel,
+ server,
+ }
+ }
+ pub async fn close(mut self) {
+ self.client.disconnect();
+ self.cancel.cancel();
+ tokio::time::timeout(Duration::from_secs(10), &mut self.server)
+ .await
+ .unwrap()
+ .unwrap()
+ .unwrap();
+ }
+}
+impl Drop for NativeHost {
+ fn drop(&mut self) {
+ self.client.disconnect();
+ self.cancel.cancel();
+ }
+}
+
+pub async fn configure(client: &Client, url: &str, models: Value) -> Value {
+ let created = client.request(Operation::ConnectionCatalogCreate, json!({
+ "expectedCatalogRevision":0,"connection":{"slug":"attachments","name":"Attachment fixture",
+ "providerType":"openai","baseUrl":url,"enabled":true,
+ "enabledModelIds":models.as_object().unwrap().keys().collect::<Vec<_>>(),"modelOverrides":models}
+ })).await.unwrap();
+ assert_eq!(created["kind"], "committed");
+ let basis = &created["connection"];
+ let credential = client.request(Operation::CredentialVaultSet, json!({
+ "locator":{"scope":"connection","connectionId":basis["connectionId"],"kind":"api_key"},
+ "expected":null,"expectedConnection":{"connectionId":basis["connectionId"],"revision":basis["revision"],
+ "slug":"attachments","providerType":"openai","effectiveBaseUrl":url},"secret":"attachment-fixture"
+ })).await.unwrap();
+ assert_eq!(credential["kind"], "committed");
+ basis["connectionId"].clone()
+}
+pub async fn session(client: &Client, connection: &Value, workspace: &Path, id: &str, model: &str) {
+ client.create_session(maka_protocol::session::decode_session_create_input(&json!({
+ "sessionId":id,"workspace":{"kind":"host_path","path":workspace},
+ "modelTarget":{"kind":"explicit","connectionId":connection,"connectionSlug":"attachments","model":model},
+ "sandboxMode":"read-only","mode":"bot"
+ })).unwrap()).await.unwrap();
+}
+pub async fn upload(
+ client: &Client,
+ session: &str,
+ id: &str,
+ name: &str,
+ mime: &str,
+ bytes: &[u8],
+) -> AttachmentRef {
+ client
+ .ingest_artifact(Ingest::Begin {
+ session_id: session.into(),
+ upload_id: id.into(),
+ name: name.into(),
+ mime_type: mime.into(),
+ total_bytes: bytes.len() as u64,
+ content_sha256: maka_runtime::artifact::content_digest(bytes),
+ })
+ .await
+ .unwrap();
+ for (index, part) in bytes.chunks(48 * 1024).enumerate() {
+ client
+ .ingest_artifact(Ingest::Chunk {
+ session_id: session.into(),
+ upload_id: id.into(),
+ offset: (index * 48 * 1024) as u64,
+ chunk_base64: STANDARD.encode(part),
+ })
+ .await
+ .unwrap();
+ }
+ let ArtifactIngestResult::Committed { attachment, .. } = client
+ .ingest_artifact(Ingest::Commit {
+ session_id: session.into(),
+ upload_id: id.into(),
+ })
+ .await
+ .unwrap()
+ else {
+ panic!("expected committed upload");
+ };
+ attachment
+}
+pub async fn completed(client: &Client, turn: &Value) -> Value {
+ client
+ .request(Operation::TurnStart, turn.clone())
+ .await
+ .unwrap();
+ tokio::time::timeout(Duration::from_secs(45), async {
+ loop {
+ let snapshot = client
+ .request(
+ Operation::TurnQuery,
+ json!({
+ "sessionId":turn["sessionId"],"turnId":turn["turnId"]
+ }),
+ )
+ .await
+ .unwrap();
+ match snapshot["status"].as_str().unwrap() {
+ "completed" => break snapshot,
+ "failed" | "cancelled" => panic!("unexpected terminal turn: {snapshot}"),
+ _ => tokio::time::sleep(Duration::from_millis(10)).await,
+ }
+ }
+ })
+ .await
+ .unwrap()
+}
+pub fn rejected<T: std::fmt::Debug>(result: Result<T, RequestFailure>, code: OperationErrorCode) {
+ assert!(
+ matches!(&result, Err(RequestFailure::Rejected(ClientError::Rejected(error))) if error.code==code),
+ "{result:?}"
+ );
+}
+pub async fn rows(client: &Client, session: &str) -> Vec<Value> {
+ let open = client
+ .open_subscription(SubscriptionOpenInput {
+ session_id: session.into(),
+ transcript: TranscriptPolicy::Tail { max_bytes: 2 },
+ })
+ .await
+ .unwrap();
+ let mut page = open.transcript.unwrap().durable;
+ assert!(page.next_cursor.is_some(), "exercise fragmented bootstrap");
+ let mut rows = Vec::new();
+ loop {
+ let batch = client
+ .complete_transcript_page(&open.subscription_id, page)
+ .await
+ .unwrap();
+ rows.extend(batch.rows.into_iter().map(|row| (row.sequence, row.value)));
+ let Some(cursor) = batch.next_cursor else {
+ break;
+ };
+ page = client
+ .transcript_page(SessionTranscriptPageInput {
+ subscription_id: open.subscription_id.clone(),
+ direction: SessionTranscriptPageDirection::Older,
+ through_sequence: batch.through_sequence,
+ cursor: Some(cursor),
+ anchor_sequence: None,
+ max_bytes: 48 * 1024,
+ })
+ .await
+ .unwrap();
+ }
+ client
+ .close_subscription(&open.subscription_id)
+ .await
+ .unwrap();
+ rows.sort_by_key(|(sequence, _)| *sequence);
+ assert!(rows.windows(2).all(|pair| pair[0].0 < pair[1].0));
+ rows.into_iter().map(|(_, row)| row).collect()
+}
+
+/// A real HTTP/SSE provider; assertions run against serialized model requests.
+pub struct Model {
+ pub url: String,
+ server: tokio::task::JoinHandle<()>,
+}
+impl Model {
+ pub async fn start(
+ count: usize,
+ response: impl Fn(usize, &Value) -> Value + Send + Sync + 'static,
+ ) -> Self {
+ use http_body_util::{BodyExt, Full, Limited};
+ use hyper::{
+ Request, Response,
+ body::{Bytes, Incoming},
+ server::conn::http1,
+ service::service_fn,
+ };
+ use hyper_util::rt::TokioIo;
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let url = format!("http://{}/v1", listener.local_addr().unwrap());
+ let response = Arc::new(response);
+ let server = tokio::spawn(async move {
+ for index in 0..count {
+ let (stream, _) = listener.accept().await.unwrap();
+ let response = response.clone();
+ http1::Builder::new().serve_connection(TokioIo::new(stream), service_fn(move |request: Request<Incoming>| {
+ let response = response.clone();
+ async move {
+ assert_eq!(request.uri().path(), "/v1/chat/completions");
+ assert_eq!(request.headers()["authorization"], "Bearer attachment-fixture");
+ let bytes = Limited::new(request.into_body(),128*1024).collect().await.unwrap().to_bytes();
+ let input: Value = serde_json::from_slice(&bytes).unwrap();
+ assert_eq!(input["stream"],true);
+ let delta = response(index,&input);
+ let finish = if delta.get("tool_calls").is_some() { "tool_calls" } else { "stop" };
+ let event = json!({"id":"attachment-fixture","object":"chat.completion.chunk","model":input["model"],
+ "choices":[{"index":0,"delta":delta,"finish_reason":finish}]});
+ Ok::<_,std::convert::Infallible>(Response::builder().status(200)
+ .header("content-type","text/event-stream").header("connection","close")
+ .body(Full::new(Bytes::from(format!("data: {event}\n\ndata: [DONE]\n\n")))).unwrap())
+ }
+ })).await.unwrap();
+ }
+ });
+ Self { url, server }
+ }
+ pub async fn finish(mut self) {
+ tokio::time::timeout(Duration::from_secs(10), &mut self.server)
+ .await
+ .unwrap()
+ .unwrap();
+ }
+}
+impl Drop for Model {
+ fn drop(&mut self) {
+ self.server.abort();
+ }
+}
+pub fn read(id: &str, artifact: &str) -> Value {
+ json!({"tool_calls":[{"index":0,"id":id,"type":"function","function":{
+ "name":"Read","arguments":json!({"path":format!("maka://runtime/attachments/{artifact}")}).to_string()
+ }}]})
+}
diff --git a/tests/fixtures/client-artifact-upload.mjs b/tests/fixtures/client-artifact-upload.mjs
deleted file mode 100644
index 34030819f6..0000000000
--- a/tests/fixtures/client-artifact-upload.mjs
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.
- */
-
-import { createHash } from 'node:crypto';
-
-export async function upload(request, sessionId, uploadId, bytes, name, mimeType) {
- const command = { sessionId, uploadId };
- await request('artifact.ingest', {
- ...command,
- kind: 'begin',
- name,
- mimeType,
- totalBytes: bytes.length,
- contentSha256: 'sha256:' + createHash('sha256').update(bytes).digest('hex'),
- });
- await request('artifact.ingest', {
- ...command,
- kind: 'chunk',
- offset: 0,
- chunkBase64: bytes.toString('base64'),
- });
- return (await request('artifact.ingest', { ...command, kind: 'commit' })).attachment;
-}
diff --git a/tests/fixtures/client-attachment-model.mjs b/tests/fixtures/client-attachment-model.mjs
index 1fc0e87718..7ca9c12b16 100644
--- a/tests/fixtures/client-attachment-model.mjs
+++ b/tests/fixtures/client-attachment-model.mjs
@@ -17,127 +17,7 @@
* under the License.
*/
-import assert from 'node:assert/strict';
-import { createServer } from 'node:http';
-import { once } from 'node:events';
-import { formatTextWithInlineRefs } from '../../packages/runtime/src/model-history.ts';
-
-export const text = 'uploaded text 😀, outside workspace authority';
export const png = Buffer.from(
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aWQAAAABJRU5ErkJggg==',
'base64',
);
-export async function attachmentModel() {
- const requests = [];
- const scenarios = new Map();
- let failure;
- const server = createServer(async (request, response) => {
- try {
- assert.equal(request.url, '/v1/chat/completions');
- let body = '';
- for await (const chunk of request) body += chunk;
- const input = JSON.parse(body);
- requests.push(input);
- const scenario = scenarios.get(input.model);
- assert(scenario);
- const { content, vision } = scenario;
- const step = ++scenario.steps;
- assert(step <= 3);
- const serialized = JSON.stringify(input.messages);
- for (const attachment of content.attachments)
- assert(serialized.includes('maka://runtime/attachments/' + attachment.ref.relativePath));
- if (vision) {
- assert.equal(
- serialized.split('data:image/png;base64,' + png.toString('base64')).length - 1,
- step === 3 ? 2 : 1,
- 'vision model receives both uploaded user and Read images as actual image parts',
- );
- } else {
- assert(!serialized.includes(png.toString('base64')));
- assert(!serialized.includes('image_url'));
- assert.equal(
- input.messages.find((m) => m.role === 'user').content,
- formatTextWithInlineRefs(content),
- );
- }
- if (step === 2) {
- const result = input.messages.findLast((m) => m.role === 'tool');
- assert.deepEqual(JSON.parse(result.content), {
- content: text,
- offset: 0,
- returnedLines: 1,
- totalLines: 1,
- next: null,
- });
- }
- if (step === 3 && vision) {
- // The Rust projection deliberately corrects the original Chat adapter's
- // JSON-stringification of tool images: bytes must reach image_url parts.
- const toolMessages = input.messages.filter((m) => m.role === 'tool');
- assert(
- toolMessages.every((m) => !JSON.stringify(m.content).includes(png.toString('base64'))),
- );
- const toolIndex = input.messages.findLastIndex((m) => m.role === 'tool');
- assert(
- input.messages
- .slice(toolIndex + 1)
- .some((m) => m.role === 'user' && m.content.some((part) => part.type === 'image_url')),
- );
- }
- const delta =
- step < 3
- ? {
- tool_calls: [
- {
- index: 0,
- id: 'read-' + step,
- type: 'function',
- function: {
- name: 'Read',
- arguments: JSON.stringify({
- path:
- 'maka://runtime/attachments/' +
- content.attachments[step - 1].ref.relativePath,
- }),
- },
- },
- ],
- }
- : { content: 'attachment complete' };
- const frame = (delta, finish_reason) =>
- 'data: ' +
- JSON.stringify({
- id: 'attachment-fixture',
- object: 'chat.completion.chunk',
- created: 1,
- model: input.model,
- choices: [{ index: 0, delta, finish_reason }],
- }) +
- '\n\n';
- response.writeHead(200, { 'Content-Type': 'text/event-stream', Connection: 'close' });
- response.end(
- frame(delta, null) + frame({}, step < 3 ? 'tool_calls' : 'stop') + 'data: [DONE]\n\n',
- );
- } catch (error) {
- failure = error;
- response.destroy(error);
- }
- });
- server.listen(0, '127.0.0.1');
- await once(server, 'listening');
- return {
- baseUrl: 'http://127.0.0.1:' + server.address().port + '/v1',
- scenarios,
- verify() {
- if (failure) throw failure;
- assert.equal(requests.length, 6);
- },
- check() {
- if (failure) throw failure;
- },
- async close() {
- server.closeAllConnections();
- await new Promise((r) => server.close(r));
- },
- };
-}
diff --git a/tests/fixtures/client-attachment-workflow.mjs b/tests/fixtures/client-attachment-workflow.mjs
deleted file mode 100644
index 52fd166c2c..0000000000
--- a/tests/fixtures/client-attachment-workflow.mjs
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * 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.
- */
-
-import assert from 'node:assert/strict';
-import { upload } from './client-artifact-upload.mjs';
-import { readFile, writeFile } from 'node:fs/promises';
-import { join } from 'node:path';
-import { setTimeout as delay } from 'node:timers/promises';
-import { decodeStoredMessage } from '../../packages/core/src/session.ts';
-import { projectDesktopToolResultContent } from '../../apps/desktop/src/shared/desktop-session-projection.ts';
-import { watchSession } from './client-subscription.mjs';
-import { attachmentModel, png, text } from './client-attachment-model.mjs';
-
-async function rows(connection, sessionId) {
- const observer = await watchSession(connection, sessionId, { kind: 'tail', maxBytes: 2 });
- try {
- return await observer.subscription.loadTranscript(decodeStoredMessage);
- } finally {
- await observer.close();
- }
-}
-export async function verifyConsumption(connection, workspace, reopened) {
- const request = (operation, input) => connection.request(operation, input, 3000);
- const snapshot = join(workspace, 'attachments.json');
- if (reopened) {
- const saved = JSON.parse(await readFile(snapshot, 'utf8'));
- for (const item of saved) {
- assert.deepEqual(await rows(connection, item.turn.sessionId), item.rows);
- assert.deepEqual((await request('turn.start', item.turn)).turn, item.terminal);
- for (const attachment of item.turn.content.attachments)
- assert.equal(
- (
- await request('artifact.query', {
- kind: 'get',
- sessionId: item.turn.sessionId,
- artifactId: attachment.ref.relativePath,
- })
- ).artifact,
- null,
- );
- }
- return;
- }
- const model = await attachmentModel();
- try {
- const created = await request('connection.catalog.create', {
- expectedCatalogRevision: 0,
- connection: {
- slug: 'attachments',
- name: 'Attachments',
- providerType: 'openai',
- baseUrl: model.baseUrl,
- enabled: true,
- enabledModelIds: ['text-model', 'gpt-4o'],
- modelOverrides: {
- 'text-model': { vision: false },
- 'gpt-4o': { vision: true, codeMode: false },
- },
- },
- });
- assert.equal(created.kind, 'committed');
- const basis = created.connection;
- await request('credential.vault.set', {
- locator: { scope: 'connection', connectionId: basis.connectionId, kind: 'api_key' },
- expected: null,
- expectedConnection: {
- ...basis,
- slug: 'attachments',
- providerType: 'openai',
- effectiveBaseUrl: model.baseUrl,
- },
- secret: 'fixture',
- });
- const saved = [];
- for (const vision of [false, true]) {
- const sessionId = vision ? 'vision-session' : 'text-session';
- const modelId = vision ? 'gpt-4o' : 'text-model';
- await request('session.create', {
- sessionId,
- workspace: { kind: 'host_path', path: workspace },
- modelTarget: {
- kind: 'explicit',
- connectionId: basis.connectionId,
- connectionSlug: 'attachments',
- model: modelId,
- },
- sandboxMode: 'read-only',
- mode: 'bot',
- });
- const attachments = [
- await upload(request, sessionId, 'text', Buffer.from(text), 'note.txt', 'text/plain'),
- await upload(request, sessionId, 'image', png, 'pixel.png', 'image/png'),
- ];
- const turn = {
- sessionId,
- turnId: 'consume',
- maxSteps: 4,
- content: { text: 'consume uploaded resources', attachments },
- };
- model.scenarios.set(modelId, { vision, content: turn.content, steps: 0 });
- for (const corrupt of [
- { ...attachments[0], name: 'forged.txt' },
- { ...attachments[0], bytes: attachments[0].bytes + 1 },
- { ...attachments[0], mimeType: 'application/json' },
- { ...attachments[1], kind: 'other' },
- { ...attachments[0], ref: { ...attachments[0].ref, sessionId: 'other-session' } },
- ]) {
- await assert.rejects(
- request('turn.start', {
- ...turn,
- turnId: 'rejected',
- content: { ...turn.content, attachments: [corrupt] },
- }),
- (error) => error.code === 'operation_conflict',
- );
- await assert.rejects(
- request('turn.query', { sessionId, turnId: 'rejected' }),
- (error) => error.code === 'not_found',
- );
- }
- await request('turn.start', turn);
- let terminal;
- const deadline = Date.now() + 6000;
- while (Date.now() < deadline) {
- model.check();
- terminal = await request('turn.query', { sessionId, turnId: turn.turnId });
- if (['completed', 'failed', 'cancelled'].includes(terminal.status)) break;
- await delay(10);
- }
- assert.equal(terminal.status, 'completed');
- const stored = await rows(connection, sessionId);
- assert.deepEqual(stored.find((r) => r.type === 'user').attachments, attachments);
- const results = stored.filter((r) => r.type === 'tool_result');
- assert.equal(results.length, 2);
- assert(results.every((r) => !r.isError));
- assert.deepEqual(results[0].content, { kind: 'text', text });
- assert.deepEqual(results[1].content, {
- kind: 'image',
- mimeType: 'image/png',
- ref: attachments[1].ref,
- });
- const desktopImage = projectDesktopToolResultContent(
- { hostId: connection.rootId },
- results[1].content,
- );
- assert.deepEqual(desktopImage.ref, {
- ...attachments[1].ref,
- sessionId: JSON.stringify([connection.rootId, sessionId]),
- });
- assert.equal(
- results[1].content.ref.sessionId,
- sessionId,
- 'Desktop scoping never mutates canonical Session reference',
- );
- for (const attachment of attachments)
- await request('artifact.delete', { sessionId, artifactId: attachment.ref.relativePath });
- assert.deepEqual((await request('turn.start', turn)).turn, terminal);
- saved.push({ turn, terminal, rows: stored });
- }
- model.verify();
- await writeFile(snapshot, JSON.stringify(saved));
- } finally {
- await model.close();
- }
-}
diff --git a/tests/fixtures/client-entry.mjs b/tests/fixtures/client-entry.mjs
index 91d08c7347..4a00e284ae 100644
--- a/tests/fixtures/client-entry.mjs
+++ b/tests/fixtures/client-entry.mjs
@@ -60,10 +60,8 @@ import { verifyPruning } from './client-pruning-workflow.mjs';
import { verifyContextCompact } from './client-context-compact-workflow.mjs';
import { verifyResume } from './client-resume-workflow.mjs';
import { verifyWorkspaceImage } from './client-workspace-image-workflow.mjs';
-import { verifyLargeOutput } from './client-large-output-workflow.mjs';
import { verifyCompatibleChat } from './client-compatible-workflow.mjs';
import { verifyRelayOptions } from './client-relay-workflow.mjs';
-import { verifyConsumption } from './client-attachment-workflow.mjs';
const workflows = {
'resume-workspace': { verify: verifyResume, marker: 'resume' },
@@ -71,10 +69,8 @@ const workflows = {
'pruning-workspace': { verify: verifyPruning, marker: 'pruning' },
'context-compact-workspace': { verify: verifyContextCompact, marker: 'context-compact' },
'workspace-image-workspace': { verify: verifyWorkspaceImage, marker: 'workspace-image' },
- 'large-output-workspace': { verify: verifyLargeOutput, marker: 'large-output' },
'compatible-workspace': { verify: verifyCompatibleChat, marker: 'compatible-options' },
'relay-workspace': { verify: verifyRelayOptions, marker: 'relay-options' },
- 'attachment-workspace': { verify: verifyConsumption, marker: 'attachment-consumption' },
};
async function main() {
@@ -162,13 +158,11 @@ async function main() {
? 140000
: values['message-queue-workspace']
? 45000 // Includes 1,030 serial durable commands; each request still has its own deadline.
- : values['large-output-workspace']
- ? 120000
- : values['shell-workspace']
- ? 30000
- : selected.length
- ? 15000
- : 10000,
+ : values['shell-workspace']
+ ? 30000
+ : selected.length
+ ? 15000
+ : 10000,
);
try {
({ connection, transport } = await openClient(values, input));
diff --git a/tests/fixtures/client-large-output-fixture.mjs b/tests/fixtures/client-large-output-fixture.mjs
deleted file mode 100644
index c3c7d74448..0000000000
--- a/tests/fixtures/client-large-output-fixture.mjs
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * 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.
- */
-
-import assert from 'node:assert/strict';
-import { once } from 'node:events';
-import { createServer } from 'node:http';
-
-export async function largeOutputFixture(port = 0) {
- const scenarios = new Map();
- let failure,
- requests = 0;
- const server = createServer(async (request, response) => {
- try {
- assert.equal(request.url, '/v1/chat/completions');
- assert.equal(request.headers.authorization, 'Bearer large-output-fixture');
- const chunks = [];
- let bytes = 0;
- for await (const chunk of request) {
- bytes += chunk.length;
- assert(bytes <= 128 * 1024, 'raw output must never enter the next model request');
- chunks.push(chunk);
- }
- const input = JSON.parse(Buffer.concat(chunks).toString());
- assert.equal(input.stream, true);
- const scenario = scenarios.get(input.model);
- assert(scenario);
- requests++;
- const step = ++scenario.steps;
- assert(step <= scenario.expectedSteps);
- const tools = input.messages.filter((message) => message.role === 'tool');
- if (step > 1 || scenario.followup) {
- assert.equal(tools.length, 1);
- assert.equal(tools[0].tool_call_id, 'read-large');
- assert(tools[0].content.length <= 7500);
- const page = JSON.parse(tools[0].content);
- assert(page.content.length > 0);
- assert.equal(page.partialLine, true);
- assert.equal(page.returnedLines, 0);
- assert.equal(page.totalLines, 1);
- assert(page.next.path.startsWith('maka://read/'));
- }
- const read = !scenario.followup && step === 1;
- if (read) assert.equal(tools.length, 0);
- const delta = read
- ? {
- tool_calls: [
- {
- index: 0,
- id: 'read-large',
- type: 'function',
- function: {
- name: 'Read',
- arguments: JSON.stringify({
- path: 'maka://runtime/attachments/' + scenario.artifactId,
- }),
- },
- },
- ],
- }
- : { content: 'large output complete' };
- const frame = (delta, finish_reason) =>
- 'data: ' +
- JSON.stringify({
- id: 'large-output-' + requests,
- object: 'chat.completion.chunk',
- created: 1,
- model: input.model,
- choices: [{ index: 0, delta, finish_reason }],
- }) +
- '\n\n';
- response.writeHead(200, { 'Content-Type': 'text/event-stream', Connection: 'close' });
- response.end(
- frame(delta, null) + frame({}, read ? 'tool_calls' : 'stop') + 'data: [DONE]\n\n',
- );
- } catch (error) {
- failure = error;
- response.destroy(error);
- }
- });
- server.listen(port, '127.0.0.1');
- await once(server, 'listening');
- return {
- baseUrl: 'http://127.0.0.1:' + server.address().port + '/v1',
- scenarios,
- check() {
- if (failure) throw failure;
- },
- verify() {
- this.check();
- assert.equal(scenarios.size, 2);
- for (const scenario of scenarios.values())
- assert.equal(scenario.steps, scenario.expectedSteps);
- },
- async close() {
- server.closeAllConnections();
- await new Promise((resolve) => server.close(resolve));
- },
- };
-}
diff --git a/tests/fixtures/client-large-output-workflow.mjs b/tests/fixtures/client-large-output-workflow.mjs
deleted file mode 100644
index 9ba6df15d8..0000000000
--- a/tests/fixtures/client-large-output-workflow.mjs
+++ /dev/null
@@ -1,281 +0,0 @@
-/*
- * 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.
- */
-
-import assert from 'node:assert/strict';
-import { createHash } from 'node:crypto';
-import { readFile, writeFile } from 'node:fs/promises';
-import { join } from 'node:path';
-import { setTimeout as delay } from 'node:timers/promises';
-import { decodeStoredMessage } from '../../packages/core/src/session.ts';
-import { projectDesktopToolResultContent } from '../../apps/desktop/src/shared/desktop-session-projection.ts';
-import { watchSession } from './client-subscription.mjs';
-import { largeOutputFixture } from './client-large-output-fixture.mjs';
-
-const SIZE = 10 * 1024 * 1024;
-const CLIENT_LIMIT = 16 * 1024 * 1024;
-const digest = (bytes) => 'sha256:' + createHash('sha256').update(bytes).digest('hex');
-
-async function upload(request, sessionId, caseName) {
- const bytes = Buffer.alloc(SIZE, caseName === 'ascii' ? 65 : 1);
- const command = { sessionId, uploadId: 'large-text' };
- await request('artifact.ingest', {
- ...command,
- kind: 'begin',
- name: caseName + '.txt',
- mimeType: 'text/plain',
- totalBytes: SIZE,
- contentSha256: digest(bytes),
- });
- for (let offset = 0; offset < bytes.length; offset += 48 * 1024) {
- await request('artifact.ingest', {
- ...command,
- kind: 'chunk',
- offset,
- chunkBase64: bytes.subarray(offset, offset + 48 * 1024).toString('base64'),
- });
- }
- return {
- attachment: (await request('artifact.ingest', { ...command, kind: 'commit' })).attachment,
- digest: digest(bytes),
- };
-}
-
-async function completed(request, turn, model) {
- await request('turn.start', turn);
- const deadline = Date.now() + 45000;
- let terminal;
- do {
- model.check();
- terminal = await request('turn.query', { sessionId: turn.sessionId, turnId: turn.turnId });
- if (['completed', 'failed', 'cancelled'].includes(terminal.status)) break;
- await delay(20);
- } while (Date.now() < deadline);
- model.check();
- assert.equal(terminal.status, 'completed');
- return terminal;
-}
-
-async function verifyAscii(connection, item) {
- const observer = await watchSession(connection, item.sessionId, { kind: 'tail', maxBytes: 2 });
- try {
- const subscription = observer.subscription;
- let page = subscription.transcriptBootstrap.durable;
- assert(page.nextCursor, 'two-byte bootstrap must exercise continuation');
- const messages = [];
- let assemblyBytes = 0,
- peakBytes = 0;
- const account = (delta) => {
- assemblyBytes += delta;
- peakBytes = Math.max(peakBytes, assemblyBytes);
- assert(assemblyBytes >= 0 && assemblyBytes <= 64 * 1024 * 1024);
- };
- for (;;) {
- const decoded = await subscription.decodeTranscriptPage(
- page,
- decodeStoredMessage,
- CLIENT_LIMIT,
- account,
- );
- messages.push(...decoded.messages);
- if (decoded.nextCursor === null) break;
- page = await subscription.loadTranscriptPage({
- direction: 'older',
- throughSequence: page.throughSequence,
- cursor: decoded.nextCursor,
- anchorSequence: null,
- maxBytes: 48 * 1024,
- });
- }
- assert.equal(assemblyBytes, 0);
- assert(peakBytes >= SIZE, 'full raw row passes through the original client assembler');
- const results = messages
- .map((entry) => entry.message)
- .filter((row) => row.type === 'tool_result');
- assert.equal(results.length, 1, 'no replayed tool effect');
- const row = results[0];
- assert.equal(row.isError, false);
- assert.equal(row.content.kind, 'text');
- assert.equal(Buffer.byteLength(row.content.text), SIZE);
- assert.equal(digest(row.content.text), item.digest);
- const displayed = projectDesktopToolResultContent({ hostId: connection.rootId }, row.content);
- assert.equal(displayed.kind, 'text');
- assert.equal(digest(displayed.text), item.digest);
- return row.id;
- } finally {
- await observer.close();
- }
-}
-
-async function verifyControlCapacity(connection, sessionId, live) {
- await assert.rejects(
- async () => {
- if (live) {
- await live.subscription.loadTranscriptPage({
- direction: 'older',
- throughSequence: null,
- cursor: null,
- anchorSequence: null,
- maxBytes: 48 * 1024,
- });
- } else {
- const opened = await watchSession(connection, sessionId, { kind: 'tail', maxBytes: 2 });
- await opened.close();
- }
- },
- (error) => error.code === 'operation_unavailable',
- 'oversized transcript is explicitly unavailable without closing its connection',
- );
- assert.equal((await connection.status(3000)).state, 'ready');
-}
-
-export async function verifyLargeOutput(connection, workspace, reopened) {
- const request = (operation, input) => connection.request(operation, input, 10000);
- const path = join(workspace, 'large-output.json');
- const saved = reopened ? JSON.parse(await readFile(path, 'utf8')) : { sessions: [] };
- const model = await largeOutputFixture(reopened ? Number(new URL(saved.baseUrl).port) : 0);
- let liveControl;
- try {
- if (!reopened) {
- const created = await request('connection.catalog.create', {
- expectedCatalogRevision: 0,
- connection: {
- slug: 'large-output',
- name: 'Large output fixture',
- providerType: 'openai',
- baseUrl: model.baseUrl,
- enabled: true,
- enabledModelIds: ['ascii-model', 'control-model'],
- modelOverrides: {
- 'ascii-model': { vision: false },
- 'control-model': { vision: false },
- },
- },
- });
- assert.equal(created.kind, 'committed');
- const basis = created.connection;
- assert.equal(
- (
- await request('credential.vault.set', {
- locator: { scope: 'connection', connectionId: basis.connectionId, kind: 'api_key' },
- expected: null,
- expectedConnection: {
- ...basis,
- slug: 'large-output',
- providerType: 'openai',
- effectiveBaseUrl: model.baseUrl,
- },
- secret: 'large-output-fixture',
- })
- ).kind,
- 'committed',
- );
- saved.baseUrl = model.baseUrl;
- for (const caseName of ['ascii', 'control']) {
- const sessionId = 'large-output-' + caseName;
- await request('session.create', {
- sessionId,
- workspace: { kind: 'host_path', path: workspace },
- mode: 'bot',
- sandboxMode: 'read-only',
- modelTarget: {
- kind: 'explicit',
- connectionId: basis.connectionId,
- connectionSlug: 'large-output',
- model: caseName + '-model',
- },
- });
- const uploaded = await upload(request, sessionId, caseName);
- const turn = {
- sessionId,
- turnId: 'read-large',
- maxSteps: 3,
- content: { text: 'Read the uploaded text', attachments: [uploaded.attachment] },
- };
- model.scenarios.set(caseName + '-model', {
- artifactId: uploaded.attachment.ref.relativePath,
- steps: 0,
- expectedSteps: 3,
- });
- if (caseName === 'control')
- liveControl = await watchSession(connection, sessionId, { kind: 'tail', maxBytes: 2 });
- const terminal = await completed(request, turn, model);
- if (caseName === 'control') {
- await liveControl.terminal(terminal);
- await verifyControlCapacity(connection, sessionId, liveControl);
- const session = (await request('session.catalog.query', { kind: 'get', sessionId }))
- .session;
- const changed = await request('session.metadata.update', {
- sessionId,
- expectedRevision: session.revision,
- patch: { name: 'Capacity remains observable' },
- });
- assert.equal(changed.kind, 'committed');
- await liveControl.waitFor(
- (frame) =>
- frame.kind === 'subscription.session_projection' &&
- frame.snapshot.session.metadataRevision === changed.session.revision,
- );
- }
- saved.sessions.push({ caseName, sessionId, turn, terminal, digest: uploaded.digest });
- }
- }
- for (const item of saved.sessions) {
- if (reopened)
- model.scenarios.set(item.caseName + '-model', {
- steps: 0,
- expectedSteps: 1,
- followup: true,
- });
- const terminal = await request('turn.query', {
- sessionId: item.sessionId,
- turnId: item.turn.turnId,
- });
- assert.deepEqual(terminal, item.terminal);
- assert.deepEqual((await request('turn.start', item.turn)).turn, item.terminal);
- if (item.caseName === 'ascii') {
- const rowId = await verifyAscii(connection, item);
- if (reopened) assert.equal(rowId, item.rowId);
- else item.rowId = rowId;
- } else await verifyControlCapacity(connection, item.sessionId, liveControl);
- // A subsequent provider turn proves both host usability and compact model history.
- const followup = {
- sessionId: item.sessionId,
- turnId: reopened ? 'after-reopen' : 'after-capacity',
- content: { text: 'Confirm completion' },
- maxSteps: 1,
- };
- const scenario = model.scenarios.get(item.caseName + '-model');
- scenario.followup = true;
- const followupTerminal = await completed(request, followup, model);
- if (item.caseName === 'control' && liveControl) {
- await liveControl.terminal(followupTerminal);
- await verifyControlCapacity(connection, item.sessionId, liveControl);
- }
- assert.equal((await connection.status(3000)).state, 'ready');
- }
- model.verify();
- if (!reopened) await writeFile(path, JSON.stringify(saved));
- } finally {
- try {
- await liveControl?.close();
- } finally {
- await model.close();
- }
- }
-}
diff --git a/tests/fixtures/client.mjs b/tests/fixtures/client.mjs
index 73bbece0bc..24ed121c2f 100644
--- a/tests/fixtures/client.mjs
+++ b/tests/fixtures/client.mjs
@@ -56,15 +56,12 @@ await withSourceBundle(fileURLToPath(new URL(entry, import.meta.url)), (bundle,
? 150000
: process.argv.includes('--message-queue-workspace')
? 55000
- : process.argv.includes('--large-output-workspace')
- ? 120000
- : process.argv.includes('--sandbox-workspace') ||
- process.argv.includes('--shell-workspace') ||
- process.argv.includes('--onboarding-workspace') ||
- (process.argv.includes('--native-managed') &&
- process.argv.includes('--management'))
- ? 45000
- : 15000,
+ : process.argv.includes('--sandbox-workspace') ||
+ process.argv.includes('--shell-workspace') ||
+ process.argv.includes('--onboarding-workspace') ||
+ (process.argv.includes('--native-managed') && process.argv.includes('--management'))
+ ? 45000
+ : 15000,
});
if (child.error) throw child.error;
if (child.signal) throw new Error(`Original-client subprocess terminated by ${child.signal}`);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org