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/17 21:34:30 UTC
(maka) branch rust-preview updated: fix(agent): preserve complete admitted image payloads
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 8a224fe3b5 fix(agent): preserve complete admitted image payloads
8a224fe3b5 is described below
commit 8a224fe3b54510b331de114715d4ec4fb39e1fb4
Author: Wang <m4...@gmail.com>
AuthorDate: Sun Oct 18 05:34:24 2026 +0800
fix(agent): preserve complete admitted image payloads
Read authorized image artifacts under the storage admission limit and use
V8's default heap policy for the trusted provider SDK. Keep separate guest
limits and existing runtime failure and cleanup handling.
Cover full-size projection, copied image authority, oversize storage
rejection, and original uploaded PNG delivery. Observe canonical failures
while the image fixture awaits HTTP so diagnostics survive cleanup.
Generated-by: Codex
---
platform/event-log/tests/artifacts.rs | 56 ++++++++++++++++++++++
platform/js-runtime/README.md | 2 +
platform/js-runtime/README.zh-CN.md | 2 +
platform/js-runtime/src/trusted/engine.rs | 3 --
plugins/agent/engine/README.md | 2 +
plugins/agent/engine/README.zh-CN.md | 2 +
plugins/agent/engine/src/history/images.rs | 14 ++----
.../engine/tests/integration/image_projection.rs | 18 ++++---
.../tests/integration/support/image_projection.rs | 1 -
.../native/src/owner/tests/scale/image_upload.rs | 27 ++++++++++-
10 files changed, 101 insertions(+), 26 deletions(-)
diff --git a/platform/event-log/tests/artifacts.rs b/platform/event-log/tests/artifacts.rs
index a1bbeaa619..025b2dbc5a 100644
--- a/platform/event-log/tests/artifacts.rs
+++ b/platform/event-log/tests/artifacts.rs
@@ -40,6 +40,62 @@ fn artifact(id: &str, bytes: &[u8]) -> Artifact {
}
}
+#[tokio::test]
+async fn oversized_artifact_rejection_preserves_catalog_and_writer() {
+ let temp = tempfile::tempdir().unwrap();
+ let log = EventLog::open(&temp.path().join("runtime.sqlite"))
+ .await
+ .unwrap();
+ log.create_session("session", "create", &json!({}), 1)
+ .await
+ .unwrap();
+ let before = log.list_artifacts("session", 0, 128).await.unwrap();
+ let bytes = vec![0x5a; 50 * 1024 * 1024 + 1];
+ let record = artifact("oversized", &bytes);
+ assert!(matches!(
+ log.commit_artifact(record, bytes).await,
+ Err(StoreError::InvalidTransition(message))
+ if message == "Artifact payload size mismatch or limit exceeded"
+ ));
+ let after = log.list_artifacts("session", 0, 128).await.unwrap();
+ assert_eq!(after.revision, before.revision);
+ assert_eq!(after.records, before.records);
+ assert_eq!(after.total, 0);
+ assert!(
+ log.get_artifact("session", "oversized")
+ .await
+ .unwrap()
+ .record
+ .is_none()
+ );
+ assert!(
+ log.read_artifact_chunk("session", "oversized", 0, 1)
+ .await
+ .unwrap()
+ .is_none()
+ );
+
+ let accepted = artifact("accepted", b"writer remains usable");
+ assert_eq!(
+ log.commit_artifact(accepted.clone(), b"writer remains usable".to_vec())
+ .await
+ .unwrap(),
+ accepted
+ );
+ let committed = log.list_artifacts("session", 0, 128).await.unwrap();
+ assert_ne!(committed.revision, before.revision);
+ assert_eq!(committed.records, [accepted]);
+ assert_eq!(
+ log.read_artifact_chunk("session", "accepted", 0, 64)
+ .await
+ .unwrap()
+ .unwrap()
+ .bytes,
+ b"writer remains usable"
+ );
+ log.close().await.unwrap();
+}
+
#[tokio::test]
async fn immutable_uploads_are_scoped_bounded_and_reopen_with_atomic_catalog_changes() {
let temp = tempfile::tempdir().unwrap();
diff --git a/platform/js-runtime/README.md b/platform/js-runtime/README.md
index e879796363..04be1c3ba3 100644
--- a/platform/js-runtime/README.md
+++ b/platform/js-runtime/README.md
@@ -23,4 +23,6 @@
Provides the V8 execution machinery shared by plugin guests and persistent JavaScript REPLs. `plugin` implements bounded guest execution; `repl` and `CodeExecutor` support cells that retain variables between calls. Embeddings supply callable capabilities and own their resources. A JavaScript promise finishing does not prove that the native work it started has settled.
+The trusted provider SDK isolate uses V8’s default heap policy and queues requests with byte reservations held through native cleanup. Heap-limit notifications mark the shared runtime failed; native out-of-memory failures can still terminate the process. Generated code and plugin guests keep their separately configured limits.
+
[Source](src/lib.rs) · [Cargo manifest](Cargo.toml)
diff --git a/platform/js-runtime/README.zh-CN.md b/platform/js-runtime/README.zh-CN.md
index b8d572f3d0..836e5cb906 100644
--- a/platform/js-runtime/README.zh-CN.md
+++ b/platform/js-runtime/README.zh-CN.md
@@ -23,4 +23,6 @@
提供插件 guest 和持久 JavaScript REPL 共用的 V8 执行机制。`plugin` 负责受限的 guest 执行，`repl` 与 `CodeExecutor` 支持跨调用保留变量的代码单元。嵌入方提供可调用能力并管理其资源；JavaScript Promise 完成不代表它启动的原生工作已完成清理。
+可信 provider SDK isolate 使用 V8 默认堆策略，通过字节预留安排请求，并将预留持有至原生资源清理完成。堆上限回调会标记共享运行时失败；原生内存耗尽仍可能终止进程。生成代码与插件 guest 仍使用各自配置的限制。
+
[源码](src/lib.rs) · [Cargo 配置](Cargo.toml)
diff --git a/platform/js-runtime/src/trusted/engine.rs b/platform/js-runtime/src/trusted/engine.rs
index 4f14b08c04..6720048bb9 100644
--- a/platform/js-runtime/src/trusted/engine.rs
+++ b/platform/js-runtime/src/trusted/engine.rs
@@ -74,9 +74,6 @@ pub(super) fn create(health: &Arc<Health>) -> Result<JsRuntime> {
deno_crypto::deno_crypto::init(None),
super::ops::maka_trusted::init(),
],
- create_params: Some(
- deno_core::v8::CreateParams::default().heap_limits(0, 256 * 1024 * 1024),
- ),
..Default::default()
})
.map_err(|error| TrustedError::Failed(error.to_string()))?;
diff --git a/plugins/agent/engine/README.md b/plugins/agent/engine/README.md
index 0f24ad3709..32caaba6af 100644
--- a/plugins/agent/engine/README.md
+++ b/plugins/agent/engine/README.md
@@ -25,4 +25,6 @@ Coordinates Agent runs, model steps, tools, handoffs and recovery. `Engine` cons
Resuming a run preserves its original input and settled history. New model steps use the current authorized tool state.
+Vision history projection reads each authorized image in full under the existing artifact storage limit. Copied sessions resolve retained images through destination-owned references.
+
[Source](src/lib.rs) · [Cargo manifest](Cargo.toml)
diff --git a/plugins/agent/engine/README.zh-CN.md b/plugins/agent/engine/README.zh-CN.md
index c9bf8b344e..71d8a9beff 100644
--- a/plugins/agent/engine/README.zh-CN.md
+++ b/plugins/agent/engine/README.zh-CN.md
@@ -25,4 +25,6 @@
续跑保留原运行的输入和已完成历史。新的模型步骤使用当前已授权的工具状态。
+视觉模型的历史投影按既有 artifact 存储上限读取每张已授权图片的完整内容。复制会话通过目标会话持有的引用解析保留的图片。
+
[源码](src/lib.rs) · [Cargo 配置](Cargo.toml)
diff --git a/plugins/agent/engine/src/history/images.rs b/plugins/agent/engine/src/history/images.rs
index 2759b7fe7f..9ab9ec024e 100644
--- a/plugins/agent/engine/src/history/images.rs
+++ b/plugins/agent/engine/src/history/images.rs
@@ -19,6 +19,7 @@
use base64::{Engine as _, engine::general_purpose::STANDARD};
use maka_agent_facts::attachment::AttachmentRef;
+use maka_agent_facts::attachment::MAX_ATTACHMENT_BYTES;
use maka_agent_facts::attachment::StorageRef;
use maka_agent_facts::attachment::sniff_binary_mime;
use maka_agent_facts::event::StoredEvent;
@@ -32,9 +33,6 @@ use crate::RunError;
pub(super) const NO_VISION: &str =
"Image was read, but the selected model does not support image input.";
-// Bound each artifact read, not the accumulated media in a model request.
-// Historical images must not consume the allowance for a newly captured image.
-const MEDIA_READ_LIMIT: usize = 12 * 1024 * 1024;
/// Targets borrow canonical evidence and are emitted alongside their messages.
/// Arbitrary JSON tool results cannot authorize a storage read.
@@ -179,10 +177,9 @@ async fn read(
if maka_agent_facts::interaction::entity_id(relative_path).is_err() {
return Ok(ImageRead::Unavailable("not_found"));
}
- // Signatures fit within 1024 bytes. Read one bounded snapshot, enough to
- // distinguish an invalid image from a valid image exceeding the per-artifact read limit.
+ // Storage admission bounds each artifact; project its complete accepted payload.
let Some(chunk) = log
- .read_artifact_chunk(session, relative_path, 0, MEDIA_READ_LIMIT + 1)
+ .read_artifact_chunk(session, relative_path, 0, MAX_ATTACHMENT_BYTES as usize)
.await?
else {
return Ok(ImageRead::Unavailable("not_found"));
@@ -190,10 +187,5 @@ async fn read(
if !audio && sniff_binary_mime(&chunk.bytes).is_none() {
return Ok(ImageRead::Unavailable("unsupported_mime"));
}
- if chunk.total_bytes > MEDIA_READ_LIMIT as u64 {
- return Ok(ImageRead::Unavailable(
- "media exceeds the 12 MiB per-artifact read limit",
- ));
- }
Ok(ImageRead::Bytes(chunk.bytes))
}
diff --git a/plugins/agent/engine/tests/integration/image_projection.rs b/plugins/agent/engine/tests/integration/image_projection.rs
index 7671c9c0ee..ecce7d2eeb 100644
--- a/plugins/agent/engine/tests/integration/image_projection.rs
+++ b/plugins/agent/engine/tests/integration/image_projection.rs
@@ -39,7 +39,7 @@ use tokio_util::sync::CancellationToken;
const FOUR_MIB: usize = 4 * 1024 * 1024;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn reopened_images_do_not_starve_current_images_and_each_artifact_read_is_bounded() {
+async fn reopened_images_preserve_complete_accepted_artifacts_and_copied_authority() {
tokio::time::timeout(Duration::from_secs(60), async {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("events.sqlite");
@@ -48,7 +48,7 @@ async fn reopened_images_do_not_starve_current_images_and_each_artifact_read_is_
for id in ["historical", "tool", "fits"] { artifact(&log, id, FOUR_MIB, true).await; }
artifact(&log, "invalid", FOUR_MIB + 1, false).await;
artifact(&log, "overflow", 8, true).await;
- artifact(&log, "too-large", 3 * FOUR_MIB + 1, true).await;
+ artifact(&log, "accepted-50-mib", 50 * 1024 * 1024, true).await;
let prior = identity("historical");
let mut historical = attachment("historical");
historical["bytes"] = json!(FOUR_MIB);
@@ -94,26 +94,24 @@ async fn reopened_images_do_not_starve_current_images_and_each_artifact_read_is_
workspace["ref"] = json!({"kind":"workspace_file", "relativePath":"historical"});
let engine = engine(log.clone());
engine.run(input(&base, json!([attachment("invalid"), attachment("missing"), foreign, workspace,
- attachment("fits"), attachment("overflow"), attachment("too-large")])), CancellationToken::new()).await.unwrap();
+ attachment("fits"), attachment("overflow"), attachment("accepted-50-mib")])), CancellationToken::new()).await.unwrap();
let body = server.await.unwrap();
let mut urls = Vec::new();
collect_images(&body["messages"], &mut urls);
- assert_eq!(urls.len(), 4, "prior user + typed Read + both current images; history must not crowd out the newest image");
- let mut sizes = Vec::new();
- for url in urls {
+ assert_eq!(urls.len(), 5, "prior user + typed Read + all three current images must reach the provider");
+ for (url, size) in urls.into_iter().zip([FOUR_MIB, FOUR_MIB, FOUR_MIB, 8, 50 * 1024 * 1024]) {
let data = url.strip_prefix("data:image/png;base64,").unwrap();
let bytes = STANDARD.decode(data).unwrap();
- sizes.push(bytes.len());
+ assert!(bytes.len() == size, "provider image length differs from the accepted artifact");
assert!(bytes.starts_with(b"\x89PNG\r\n\x1a\n"));
+ assert!(bytes[8..].iter().all(|byte| *byte == 0), "provider image bytes differ from the accepted artifact");
}
- assert_eq!(sizes, [FOUR_MIB, FOUR_MIB, FOUR_MIB, 8]);
let text = body["messages"].to_string();
assert!(text.contains("\"detail\":\"high\""), "tool image detail reaches the provider after reopening");
- for name in ["invalid", "missing", "historical", "workspace", "too-large"] {
+ for name in ["invalid", "missing", "historical", "workspace"] {
assert!(text.contains(&format!("Image attachment \\\"{name}\\\" could not be loaded")), "missing error for {name}");
}
assert!(text.contains("unsupported_mime"));
- assert!(text.contains("per-artifact read limit"));
assert!(!text.contains("per-request image budget"));
assert!(text.contains("session_mismatch"));
assert!(text.contains("unsupported_ref_kind"));
diff --git a/plugins/agent/engine/tests/integration/support/image_projection.rs b/plugins/agent/engine/tests/integration/support/image_projection.rs
index 9380355cdf..5af2f81e89 100644
--- a/plugins/agent/engine/tests/integration/support/image_projection.rs
+++ b/plugins/agent/engine/tests/integration/support/image_projection.rs
@@ -136,7 +136,6 @@ pub async fn request(socket: &mut TcpStream) -> Value {
.then(|| value.trim().parse().unwrap())
})
.unwrap();
- assert!(length < 20 * 1024 * 1024);
let received = bytes.len();
bytes.resize(boundary + length, 0);
socket.read_exact(&mut bytes[received..]).await.unwrap();
diff --git a/plugins/agent/native/src/owner/tests/scale/image_upload.rs b/plugins/agent/native/src/owner/tests/scale/image_upload.rs
index 1fb56b54f5..801ee959d1 100644
--- a/plugins/agent/native/src/owner/tests/scale/image_upload.rs
+++ b/plugins/agent/native/src/owner/tests/scale/image_upload.rs
@@ -443,7 +443,27 @@ async fn run(count: usize, side: u32) {
invocation = Some(sent["invocationId"].as_str().unwrap().to_owned());
metrics["admission_ms"] = json!(turn.elapsed().as_secs_f64()*1000.0);
stage.set("model_http");
- receive.await.unwrap().unwrap_or_else(|panic| std::panic::resume_unwind(panic));
+ let mut commits = f.store.subscribe_commits();
+ let terminal_failure = async {
+ loop {
+ commits.borrow_and_update();
+ for stored in canonical(&f, session).await.unwrap() {
+ if stored.event.invocation.invocation_id != invocation.as_deref().unwrap() {
+ continue;
+ }
+ if let Fact::InvocationEnded { outcome } = stored.event.fact {
+ assert!(matches!(outcome, InvocationOutcome::Completed),
+ "image invocation ended before HTTP completion: {outcome:?}");
+ }
+ }
+ commits.changed().await.unwrap();
+ }
+ };
+ tokio::select! {
+ biased;
+ result = receive => result.unwrap().unwrap_or_else(|panic| std::panic::resume_unwind(panic)),
+ _ = terminal_failure => unreachable!("terminal failure observer only returns by panic"),
+ }
stage.set("terminal");
assert!(f.owner.workers.close());
f.owner.workers.wait().await;
@@ -584,6 +604,11 @@ async fn uploaded_pngs_reach_vision_http_and_complete_history() {
run(2, 64).await;
}
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn uploaded_large_png_reaches_vision_http_and_complete_history() {
+ run(1, 2048).await;
+}
+
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "explicit real upload/vision count and encoded-size probe, one case per process"]
async fn real_image_upload_vision_scale() {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org