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/14 16:09:35 UTC
(maka) 03/20: feat(runtime-host): transfer WorkHub attachments atomically
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
commit 0cb2be88e6b0e4f7157761bd63e4800c8a4ab3ff
Author: Wang <m4...@gmail.com>
AuthorDate: Mon Sep 14 17:18:50 2026 +0800
feat(runtime-host): transfer WorkHub attachments atomically
Generated-by: OpenAI Codex
---
crates/event-log/src/artifacts.rs | 2 +
crates/event-log/src/artifacts/copy.rs | 113 ++++++++++++
crates/event-log/src/workhub.rs | 33 +++-
crates/event-log/tests/workhub_attachments.rs | 199 +++++++++++++++++++++
.../src/execution/prepare/attachments.rs | 90 +---------
crates/runtime/src/artifact.rs | 80 +++++++++
crates/runtime/src/workhub.rs | 43 ++++-
docs/rust-runtime.md | 4 +-
docs/rust-runtime.zh-CN.md | 2 +-
tests/fixtures/client-artifact-upload.mjs | 39 ++++
tests/fixtures/client-attachment-workflow.mjs | 20 +--
tests/fixtures/client-workhub-answer.mjs | 33 ++--
tests/fixtures/client-workhub-delegation.mjs | 54 ++++--
13 files changed, 550 insertions(+), 162 deletions(-)
diff --git a/crates/event-log/src/artifacts.rs b/crates/event-log/src/artifacts.rs
index 2e84bee4c0..04c0af25d7 100644
--- a/crates/event-log/src/artifacts.rs
+++ b/crates/event-log/src/artifacts.rs
@@ -17,7 +17,9 @@
* under the License.
*/
+mod copy;
mod read;
+pub(crate) use copy::copy_in_transaction;
mod records;
use crate::{EventLog, StoreError};
use maka_runtime::{
diff --git a/crates/event-log/src/artifacts/copy.rs b/crates/event-log/src/artifacts/copy.rs
new file mode 100644
index 0000000000..583dea342a
--- /dev/null
+++ b/crates/event-log/src/artifacts/copy.rs
@@ -0,0 +1,113 @@
+/*
+ * 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::records::{advance, invalid, read_record, require_session};
+use crate::StoreError;
+use maka_runtime::{
+ artifact::ArtifactSource,
+ attachment::{AttachmentRef, MAX_ATTACHMENT_BYTES, StorageRef},
+};
+use sqlx::SqliteConnection;
+
+/// Copy an explicitly authorized attachment without materializing its BLOB in the Host.
+/// The caller commits this together with the message that owns the destination.
+pub(crate) async fn copy_in_transaction(
+ tx: &mut SqliteConnection,
+ source: &AttachmentRef,
+ destination: &AttachmentRef,
+ turn: &str,
+ created_at: u64,
+) -> Result<(), StoreError> {
+ let StorageRef::SessionFile {
+ session_id: source_session,
+ relative_path: source_id,
+ } = &source.storage_ref
+ else {
+ return Err(invalid("Attachment source is not a Session Artifact"));
+ };
+ let StorageRef::SessionFile {
+ session_id: target_session,
+ relative_path: target_id,
+ } = &destination.storage_ref
+ else {
+ return Err(invalid("Attachment destination is not a Session Artifact"));
+ };
+ let (mut artifact, digest) = read_record(tx, source_session, source_id)
+ .await?
+ .ok_or_else(|| invalid("Delegated attachment source no longer exists"))?;
+ artifact.validate_attachment(source).map_err(invalid)?;
+ if artifact.size_bytes > MAX_ATTACHMENT_BYTES
+ || artifact.source == ArtifactSource::UserUpload
+ && artifact.summary.as_ref() != Some(&digest)
+ {
+ return Err(StoreError::ArtifactConflict);
+ }
+ require_session(tx, target_session).await?;
+ artifact.session_id = target_session.clone();
+ artifact.id = target_id.clone();
+ artifact.turn_id = turn.into();
+ artifact.created_at = created_at;
+ artifact.validate().map_err(invalid)?;
+ artifact.validate_attachment(destination).map_err(invalid)?;
+ let encoded = serde_json::to_string(&artifact)?;
+ if encoded.len() > super::MAX_RECORD_BYTES {
+ return Err(invalid("Copied attachment metadata exceeds limit"));
+ }
+ if let Some((existing, target_digest)) = read_record(tx, target_session, target_id).await? {
+ if existing != artifact || target_digest != digest {
+ return Err(StoreError::ArtifactConflict);
+ }
+ let same: bool = sqlx::query_scalar(
+ "SELECT EXISTS(SELECT 1 FROM artifacts s JOIN artifacts t
+ ON t.session_id = ?3 AND t.id = ?4 WHERE s.session_id = ?1 AND s.id = ?2
+ AND length(s.payload) = ?5 AND s.payload = t.payload)",
+ )
+ .bind(source_session)
+ .bind(source_id)
+ .bind(target_session)
+ .bind(target_id)
+ .bind(artifact.size_bytes as i64)
+ .fetch_one(tx)
+ .await?;
+ return if same {
+ Ok(())
+ } else {
+ Err(StoreError::ArtifactConflict)
+ };
+ }
+ let inserted = sqlx::query(
+ "INSERT INTO artifacts (session_id, id, created_at, record_json, content_sha256, payload)
+ SELECT ?3, ?4, ?5, ?6, content_sha256, payload FROM artifacts
+ WHERE session_id = ?1 AND id = ?2 AND length(payload) = ?7",
+ )
+ .bind(source_session)
+ .bind(source_id)
+ .bind(target_session)
+ .bind(target_id)
+ .bind(created_at as i64)
+ .bind(encoded)
+ .bind(artifact.size_bytes as i64)
+ .execute(&mut *tx)
+ .await?
+ .rows_affected();
+ if inserted != 1 {
+ return Err(StoreError::ArtifactConflict);
+ }
+ advance(tx, target_session).await
+}
diff --git a/crates/event-log/src/workhub.rs b/crates/event-log/src/workhub.rs
index 20b91ee02b..553dfdbd8b 100644
--- a/crates/event-log/src/workhub.rs
+++ b/crates/event-log/src/workhub.rs
@@ -138,18 +138,35 @@ pub(crate) async fn apply(
if crate::shell_runs::unsettled(tx, &delegation.target.session_id).await? {
return Err(StoreError::SessionBusy);
}
+ let admitted_at = event
+ .recorded_at
+ .duration_since(std::time::UNIX_EPOCH)
+ .map_err(|_| invalid("invalid WorkHub admission time"))?
+ .as_millis()
+ .try_into()
+ .map_err(|_| invalid("invalid WorkHub admission time"))?;
+ let message = delegation.message(&content).map_err(invalid)?;
+ for (source, destination) in content
+ .attachments
+ .iter()
+ .flatten()
+ .zip(message.message.content.attachments.iter().flatten())
+ {
+ crate::artifacts::copy_in_transaction(
+ tx,
+ source,
+ destination,
+ &delegation.target.turn_id,
+ admitted_at,
+ )
+ .await?;
+ }
let admission = PendingMessageAdmission {
invocation: delegation.target.clone(),
steering_invocation: None,
- source: delegation.message(&content).map_err(invalid)?,
+ source: message,
required_tools: Default::default(),
- admitted_at: event
- .recorded_at
- .duration_since(std::time::UNIX_EPOCH)
- .map_err(|_| invalid("invalid WorkHub admission time"))?
- .as_millis()
- .try_into()
- .map_err(|_| invalid("invalid WorkHub admission time"))?,
+ admitted_at,
};
crate::message_admissions::insert::insert(
tx,
diff --git a/crates/event-log/tests/workhub_attachments.rs b/crates/event-log/tests/workhub_attachments.rs
new file mode 100644
index 0000000000..49fbb28db2
--- /dev/null
+++ b/crates/event-log/tests/workhub_attachments.rs
@@ -0,0 +1,199 @@
+/*
+ * 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 maka_event_log::EventLog;
+use maka_runtime::{
+ artifact::{Artifact, ArtifactKind, ArtifactSource, content_digest},
+ attachment::{AttachmentKind, AttachmentRef, StorageRef},
+ event::{EventWrite, Fact, Invocation, InvocationInput, RuntimeEvent},
+ input::MessageInput,
+ workhub::{COORDINATION_SESSION_ID, Delegation},
+};
+use serde_json::json;
+
+fn upload(id: &str) -> Artifact {
+ Artifact {
+ id: id.into(),
+ session_id: COORDINATION_SESSION_ID.into(),
+ turn_id: "upload".into(),
+ created_at: 1,
+ name: format!("{id}.txt"),
+ kind: ArtifactKind::File,
+ size_bytes: 65_536,
+ mime_type: Some("text/plain".into()),
+ source: ArtifactSource::UserUpload,
+ summary: Some(content_digest(&vec![b'x'; 65_536])),
+ }
+}
+fn attachment(artifact: &Artifact) -> AttachmentRef {
+ AttachmentRef {
+ kind: AttachmentKind::Other,
+ name: artifact.name.clone(),
+ mime_type: artifact.mime_type.clone().unwrap(),
+ bytes: artifact.size_bytes,
+ storage_ref: StorageRef::SessionFile {
+ session_id: artifact.session_id.clone(),
+ relative_path: artifact.id.clone(),
+ },
+ }
+}
+
+#[tokio::test]
+async fn delegation_copies_attachments_atomically_and_replays_without_the_deleted_source() {
+ let temp = tempfile::tempdir().unwrap();
+ let path = temp.path().join("events.sqlite");
+ let log = EventLog::open(&path).await.unwrap();
+ for session in [COORDINATION_SESSION_ID, "target"] {
+ log.create_session(session, "create", &json!({}), 1)
+ .await
+ .unwrap();
+ }
+ let first = upload("first");
+ let second = upload("second");
+ log.commit_artifact(first.clone(), vec![b'x'; 65_536])
+ .await
+ .unwrap();
+ log.commit_artifact(second.clone(), vec![b'x'; 65_536])
+ .await
+ .unwrap();
+ let coordinator = Invocation {
+ session_id: COORDINATION_SESSION_ID.into(),
+ turn_id: "source-turn".into(),
+ run_id: "source-run".into(),
+ invocation_id: "source-invocation".into(),
+ };
+ let content = MessageInput {
+ attachments: Some(vec![attachment(&first), attachment(&second)]),
+ .."Use both attachments".into()
+ };
+ let opening = EventWrite::plain(RuntimeEvent::new(
+ coordinator.clone(),
+ Fact::InvocationOpened {
+ configuration: None,
+ input: InvocationInput::Message {
+ content: content.clone(),
+ request_fingerprint: None,
+ source_messages: Vec::new(),
+ skill_invocation: None,
+ },
+ },
+ ))
+ .unwrap();
+ log.append(&opening).await.unwrap();
+ log.delete_user_artifact(COORDINATION_SESSION_ID, &second.id)
+ .await
+ .unwrap();
+ let delegation = Delegation {
+ action_id: "copy-action".into(),
+ request_fingerprint: content_digest(b"copy request"),
+ source_message_event_id: opening.event().id.clone(),
+ target_revision: 1,
+ target: Invocation {
+ session_id: "target".into(),
+ turn_id: "target-turn".into(),
+ run_id: "target-run".into(),
+ invocation_id: "target-invocation".into(),
+ },
+ delegation_text: "Read the evidence".into(),
+ };
+ let target_message = delegation.message(&content).unwrap();
+ let action = EventWrite::plain(RuntimeEvent::new(
+ coordinator,
+ Fact::WorkhubDelegated {
+ delegation: Box::new(delegation),
+ },
+ ))
+ .unwrap();
+ let initial = log.list_artifacts("target", 0, 128).await.unwrap().revision;
+ assert!(
+ log.append(&action).await.is_err(),
+ "the second source is missing"
+ );
+ assert_eq!(
+ log.list_artifacts("target", 0, 128).await.unwrap().revision,
+ initial
+ );
+ assert!(
+ log.list_artifacts("target", 0, 128)
+ .await
+ .unwrap()
+ .records
+ .is_empty()
+ );
+ assert!(log.pending_messages("target").await.unwrap().is_empty());
+ log.commit_artifact(second.clone(), vec![b'x'; 65_536])
+ .await
+ .unwrap();
+ let db = rusqlite::Connection::open(&path).unwrap();
+ db.execute_batch("CREATE TRIGGER fail_copy_action BEFORE INSERT ON runtime_events
+ WHEN NEW.kind = 'workhub_delegated' BEGIN SELECT RAISE(ABORT, 'injected action failure'); END;").unwrap();
+ assert!(log.append(&action).await.is_err());
+ assert_eq!(
+ log.list_artifacts("target", 0, 128).await.unwrap().revision,
+ initial
+ );
+ assert!(
+ log.list_artifacts("target", 0, 128)
+ .await
+ .unwrap()
+ .records
+ .is_empty()
+ );
+ assert!(log.pending_messages("target").await.unwrap().is_empty());
+ assert!(log.workhub_action("copy-action").await.unwrap().is_none());
+ db.execute_batch("DROP TRIGGER fail_copy_action;").unwrap();
+ drop(db);
+ let sequence = log.append(&action).await.unwrap();
+ assert_eq!(
+ log.pending_messages("target").await.unwrap()[0].source,
+ target_message
+ );
+ let records = log.list_artifacts("target", 0, 128).await.unwrap().records;
+ assert_eq!(records.len(), 2);
+ for copied in &records {
+ assert_eq!(copied.session_id, "target");
+ assert_eq!(copied.turn_id, "target-turn");
+ assert!(copied.id != first.id && copied.id != second.id);
+ assert_eq!(
+ log.read_artifact_chunk("target", &copied.id, 0, 65_536)
+ .await
+ .unwrap()
+ .unwrap()
+ .bytes,
+ vec![b'x'; 65_536]
+ );
+ }
+ for source in [&first, &second] {
+ log.delete_user_artifact(COORDINATION_SESSION_ID, &source.id)
+ .await
+ .unwrap();
+ }
+ log.close().await.unwrap();
+ let log = EventLog::open(&path).await.unwrap();
+ assert_eq!(log.append(&action).await.unwrap(), sequence);
+ assert_eq!(
+ log.list_artifacts("target", 0, 128).await.unwrap().records,
+ records
+ );
+ assert_eq!(
+ log.pending_messages("target").await.unwrap()[0].source,
+ target_message
+ );
+ log.close().await.unwrap();
+}
diff --git a/crates/runtime-host/src/execution/prepare/attachments.rs b/crates/runtime-host/src/execution/prepare/attachments.rs
index aaa56b8c77..4e015c5b6f 100644
--- a/crates/runtime-host/src/execution/prepare/attachments.rs
+++ b/crates/runtime-host/src/execution/prepare/attachments.rs
@@ -20,8 +20,7 @@
use super::{Code, Result, failure, internal};
use maka_event_log::EventLog;
use maka_runtime::{
- artifact::{Artifact, ArtifactKind},
- attachment::{AttachmentKind, AttachmentRef, StorageRef},
+ attachment::{AttachmentRef, StorageRef},
interaction::entity_id,
};
@@ -61,92 +60,9 @@ pub(super) async fn validate(
.map_err(internal)?
.record
.ok_or_else(|| failure(Code::OperationConflict, "Attachment Artifact was not found"))?;
- validate_descriptor(attachment, &record)
+ record
+ .validate_attachment(attachment)
.map_err(|message| failure(Code::OperationConflict, message))?;
}
Ok(())
}
-
-fn validate_descriptor(
- attachment: &AttachmentRef,
- record: &Artifact,
-) -> std::result::Result<(), &'static str> {
- if record.name != attachment.name
- || record.mime_type.as_deref() != Some(attachment.mime_type.as_str())
- || record.size_bytes != attachment.bytes
- {
- return Err("Attachment metadata does not match its canonical Artifact");
- }
- let kind = AttachmentKind::from_metadata(&attachment.mime_type, &record.name);
- if attachment.kind != kind
- || (kind == AttachmentKind::Image) != (record.kind == ArtifactKind::Image)
- || (kind == AttachmentKind::Pdf) != (record.kind == ArtifactKind::Pdf)
- {
- return Err("Attachment kind does not match its canonical Artifact");
- }
- Ok(())
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use maka_runtime::artifact::ArtifactSource;
-
- #[test]
- fn canonical_metadata_and_bidirectional_modality_are_required_without_source_restriction() {
- let attachment = AttachmentRef {
- kind: AttachmentKind::Image,
- name: "picture.png".into(),
- mime_type: "image/png".into(),
- bytes: 4,
- storage_ref: StorageRef::SessionFile {
- session_id: "session".into(),
- relative_path: "artifact".into(),
- },
- };
- let mut record = Artifact {
- id: "artifact".into(),
- session_id: "session".into(),
- turn_id: "turn".into(),
- created_at: 1,
- name: attachment.name.clone(),
- kind: ArtifactKind::Image,
- size_bytes: attachment.bytes,
- mime_type: Some(attachment.mime_type.clone()),
- source: ArtifactSource::ToolResultProjection,
- summary: None,
- };
- assert!(validate_descriptor(&attachment, &record).is_ok());
- for changed in [
- AttachmentRef {
- name: "other.png".into(),
- ..attachment.clone()
- },
- AttachmentRef {
- mime_type: "image/jpeg".into(),
- ..attachment.clone()
- },
- AttachmentRef {
- bytes: 5,
- ..attachment.clone()
- },
- AttachmentRef {
- kind: AttachmentKind::Other,
- ..attachment.clone()
- },
- ] {
- assert!(validate_descriptor(&changed, &record).is_err());
- }
- record.kind = ArtifactKind::File;
- assert!(validate_descriptor(&attachment, &record).is_err());
- let mut text = attachment;
- text.mime_type = "text/plain".into();
- text.kind = AttachmentKind::Other;
- record.mime_type = Some(text.mime_type.clone());
- assert!(validate_descriptor(&text, &record).is_ok());
- for kind in [ArtifactKind::Image, ArtifactKind::Pdf] {
- record.kind = kind;
- assert!(validate_descriptor(&text, &record).is_err());
- }
- }
-}
diff --git a/crates/runtime/src/artifact.rs b/crates/runtime/src/artifact.rs
index 8df5e0a68d..ac192e2330 100644
--- a/crates/runtime/src/artifact.rs
+++ b/crates/runtime/src/artifact.rs
@@ -75,6 +75,27 @@ pub struct Artifact {
}
impl Artifact {
+ pub fn validate_attachment(
+ &self,
+ attachment: &crate::attachment::AttachmentRef,
+ ) -> Result<(), &'static str> {
+ use crate::attachment::AttachmentKind;
+ if self.name != attachment.name
+ || self.mime_type.as_deref() != Some(attachment.mime_type.as_str())
+ || self.size_bytes != attachment.bytes
+ {
+ return Err("Attachment metadata does not match its canonical Artifact");
+ }
+ let kind = AttachmentKind::from_metadata(&attachment.mime_type, &self.name);
+ if attachment.kind != kind
+ || (kind == AttachmentKind::Image) != (self.kind == ArtifactKind::Image)
+ || (kind == AttachmentKind::Pdf) != (self.kind == ArtifactKind::Pdf)
+ {
+ return Err("Attachment kind does not match its canonical Artifact");
+ }
+ Ok(())
+ }
+
pub fn validate(&self) -> Result<(), &'static str> {
crate::interaction::entity_id(&self.id)?;
crate::interaction::entity_id(&self.session_id)?;
@@ -157,6 +178,65 @@ fn js_whitespace(character: char) -> bool {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::attachment::{AttachmentKind, AttachmentRef, StorageRef};
+
+ #[test]
+ fn canonical_metadata_and_bidirectional_modality_are_required_without_source_restriction() {
+ let attachment = AttachmentRef {
+ kind: AttachmentKind::Image,
+ name: "picture.png".into(),
+ mime_type: "image/png".into(),
+ bytes: 4,
+ storage_ref: StorageRef::SessionFile {
+ session_id: "session".into(),
+ relative_path: "artifact".into(),
+ },
+ };
+ let mut record = Artifact {
+ id: "artifact".into(),
+ session_id: "session".into(),
+ turn_id: "turn".into(),
+ created_at: 1,
+ name: attachment.name.clone(),
+ kind: ArtifactKind::Image,
+ size_bytes: attachment.bytes,
+ mime_type: Some(attachment.mime_type.clone()),
+ source: ArtifactSource::ToolResultProjection,
+ summary: None,
+ };
+ assert!(record.validate_attachment(&attachment).is_ok());
+ for changed in [
+ AttachmentRef {
+ name: "other.png".into(),
+ ..attachment.clone()
+ },
+ AttachmentRef {
+ mime_type: "image/jpeg".into(),
+ ..attachment.clone()
+ },
+ AttachmentRef {
+ bytes: 5,
+ ..attachment.clone()
+ },
+ AttachmentRef {
+ kind: AttachmentKind::Other,
+ ..attachment.clone()
+ },
+ ] {
+ assert!(record.validate_attachment(&changed).is_err());
+ }
+ record.kind = ArtifactKind::File;
+ assert!(record.validate_attachment(&attachment).is_err());
+ let mut text = attachment;
+ text.mime_type = "text/plain".into();
+ text.kind = AttachmentKind::Other;
+ record.mime_type = Some(text.mime_type.clone());
+ assert!(record.validate_attachment(&text).is_ok());
+ for kind in [ArtifactKind::Image, ArtifactKind::Pdf] {
+ record.kind = kind;
+ assert!(record.validate_attachment(&text).is_err());
+ }
+ }
#[test]
fn names_are_idempotent_with_javascript_whitespace_and_utf16_truncation() {
diff --git a/crates/runtime/src/workhub.rs b/crates/runtime/src/workhub.rs
index f14788fcb6..1f533b1767 100644
--- a/crates/runtime/src/workhub.rs
+++ b/crates/runtime/src/workhub.rs
@@ -19,6 +19,7 @@
//! Durable WorkHub delegation identities; execution remains ordinary queued work.
use crate::{
+ attachment::{AttachmentRef, StorageRef},
event::Invocation,
input::{DeliveredMessage, MessageInput},
message::{MessageDisposition, Placement, RootSourceMessage},
@@ -70,19 +71,22 @@ impl Delegation {
/// Derive the sole target message from its canonical source and recorded task.
pub fn message(&self, user: &MessageInput) -> Result<RootSourceMessage, &'static str> {
- if user
- .attachments
- .as_ref()
- .is_some_and(|items| !items.is_empty())
- {
- return Err("WorkHub attachment transfer is not installed");
- }
let content = MessageInput {
text: format!(
"User request:\n{}\n\nDelegated task:\n{}",
user.text, self.delegation_text
),
display_text: None,
+ attachments: user
+ .attachments
+ .as_ref()
+ .map(|items| {
+ items
+ .iter()
+ .map(|attachment| self.attachment(attachment))
+ .collect()
+ })
+ .transpose()?,
..user.clone()
};
if content.text_bytes() > 64 * 1024 {
@@ -107,4 +111,29 @@ impl Delegation {
message.validate()?;
Ok(message)
}
+
+ /// Stable destination; only the canonical coordination message supplies sources.
+ pub fn attachment(&self, source: &AttachmentRef) -> Result<AttachmentRef, &'static str> {
+ let StorageRef::SessionFile {
+ session_id,
+ relative_path,
+ } = &source.storage_ref
+ else {
+ return Err("WorkHub attachments require Session Artifact references");
+ };
+ if session_id != COORDINATION_SESSION_ID {
+ return Err("WorkHub attachment belongs to another Session");
+ }
+ crate::interaction::entity_id(relative_path)?;
+ Ok(AttachmentRef {
+ storage_ref: StorageRef::SessionFile {
+ session_id: self.target.session_id.clone(),
+ relative_path: crate::artifact::upload_artifact_id(
+ &self.target.session_id,
+ &format!("workhub:{}:{relative_path}", self.action_id),
+ ),
+ },
+ ..source.clone()
+ })
+ }
}
diff --git a/docs/rust-runtime.md b/docs/rust-runtime.md
index fba9c3639a..dae3b6d42a 100644
--- a/docs/rust-runtime.md
+++ b/docs/rust-runtime.md
@@ -159,8 +159,8 @@ live verification are deferred.
WorkHub supports session resolution, queries, model configuration and conversation
with scoped Desktop tools and attachment reads, plus candidate discovery and delegation
-to existing idle sessions. Task creation, linked operations, multi-selection and attachment
-transfer remain incomplete.
+to existing idle sessions with atomic attachment transfer. Task creation, linked operations
+and multi-selection remain incomplete.
Skill mutations and update previews,
advanced recovery/reconciliation, WorkHub coordination and orchestration,
diff --git a/docs/rust-runtime.zh-CN.md b/docs/rust-runtime.zh-CN.md
index 77b171d525..eee5920cef 100644
--- a/docs/rust-runtime.zh-CN.md
+++ b/docs/rust-runtime.zh-CN.md
@@ -132,7 +132,7 @@ Grep 差分测试需要 PATH 中有 `rg`；runtime 本身不依赖该可执行
模型流式交互与上下文压缩。Codex 订阅已接入执行；Copilot／xAI 推理适配和实测后置。
WorkHub 已支持专属会话解析、查询、模型配置、受限 Desktop 工具与附件读取，以及候选发现和向已有空闲会话委派任务。
-任务创建、关联操作、多选及附件转移仍未完成。
+委派时原子转移附件；任务创建、关联操作和多选仍未完成。
Skills 变更与更新预览、高级恢复与协调、WorkHub 协调、
编排、部分 capability 服务、受管升级及其它协议域
diff --git a/tests/fixtures/client-artifact-upload.mjs b/tests/fixtures/client-artifact-upload.mjs
new file mode 100644
index 0000000000..34030819f6
--- /dev/null
+++ b/tests/fixtures/client-artifact-upload.mjs
@@ -0,0 +1,39 @@
+/*
+ * 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-workflow.mjs b/tests/fixtures/client-attachment-workflow.mjs
index 5fe4b5c412..980c7f8d44 100644
--- a/tests/fixtures/client-attachment-workflow.mjs
+++ b/tests/fixtures/client-attachment-workflow.mjs
@@ -18,7 +18,7 @@
*/
import assert from 'node:assert/strict';
-import { createHash } from 'node:crypto';
+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';
@@ -35,24 +35,6 @@ async function rows(connection, sessionId) {
await observer.close();
}
}
-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;
-}
export async function verifyConsumption(connection, workspace, reopened) {
const request = (operation, input) => connection.request(operation, input, 3000);
const snapshot = join(workspace, 'attachments.json');
diff --git a/tests/fixtures/client-workhub-answer.mjs b/tests/fixtures/client-workhub-answer.mjs
index 8fc2c719bf..41a7a5ad6b 100644
--- a/tests/fixtures/client-workhub-answer.mjs
+++ b/tests/fixtures/client-workhub-answer.mjs
@@ -18,7 +18,7 @@
*/
import assert from 'node:assert/strict';
-import { createHash } from 'node:crypto';
+import { upload } from './client-artifact-upload.mjs';
import { once } from 'node:events';
import { createServer } from 'node:http';
import { readFile, writeFile } from 'node:fs/promises';
@@ -155,29 +155,16 @@ export async function verifyWorkhubAnswer(connection, workspace, reopened) {
const session = await request('workhub.coordination.query', {});
await writeFile(join(session.workspace.hostCwd, 'AGENTS.md'), forbidden);
const bytes = Buffer.from('ATTACHMENT_EVIDENCE\nsecond line\n');
- await request('artifact.ingest', {
- kind: 'begin',
+ const attachment = await upload(
+ request,
sessionId,
- uploadId: 'workhub-upload',
- name: 'evidence.txt',
- mimeType: 'text/plain',
- totalBytes: bytes.length,
- contentSha256: 'sha256:' + createHash('sha256').update(bytes).digest('hex'),
- });
- await request('artifact.ingest', {
- kind: 'chunk',
- sessionId,
- uploadId: 'workhub-upload',
- offset: 0,
- chunkBase64: bytes.toString('base64'),
- });
- const uploaded = await request('artifact.ingest', {
- kind: 'commit',
- sessionId,
- uploadId: 'workhub-upload',
- });
- attachmentPath = 'maka://runtime/attachments/' + uploaded.attachment.ref.relativePath;
- const input = { turnId, text: '请检查附件并确认 WorkHub', attachments: [uploaded.attachment] };
+ 'workhub-upload',
+ bytes,
+ 'evidence.txt',
+ 'text/plain',
+ );
+ attachmentPath = 'maka://runtime/attachments/' + attachment.ref.relativePath;
+ const input = { turnId, text: '请检查附件并确认 WorkHub', attachments: [attachment] };
await assert.rejects(
request('workhub.coordination.answer', input),
(e) => e.code === 'operation_unavailable',
diff --git a/tests/fixtures/client-workhub-delegation.mjs b/tests/fixtures/client-workhub-delegation.mjs
index 9285e64d84..d469bf22f2 100644
--- a/tests/fixtures/client-workhub-delegation.mjs
+++ b/tests/fixtures/client-workhub-delegation.mjs
@@ -25,6 +25,7 @@ import { join } from 'node:path';
import { decodeStoredMessage } from '../../packages/core/src/session.ts';
import { watchSession } from './client-subscription.mjs';
import { createInput, querySession } from './client-runtime-policy-fixture.mjs';
+import { upload } from './client-artifact-upload.mjs';
const sessionId = 'maka_workhub_coordination';
const turnId = 'delegate-request';
@@ -57,7 +58,8 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened) {
let failure,
calls = 0,
input,
- receipt;
+ receipt,
+ attachment;
const server = createServer(async (req, response) => {
try {
let body = '';
@@ -69,20 +71,29 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened) {
message.content.includes('Delegated task:\nImplement the requested task'),
);
const finished = data.messages.some((message) => message.role === 'tool');
- const delta = target
- ? { content: 'target completed' }
- : finished
- ? { content: 'delegation completed' }
- : {
- tool_calls: [
- {
- index: 0,
- id: 'delegate-call',
- type: 'function',
- function: { name: 'mcp__desktop_workhub__tasks', arguments: '{}' },
+ const path = target
+ ? body.match(/maka:\/\/runtime\/attachments\/[A-Za-z0-9_-]+/u)?.[0]
+ : undefined;
+ if (target) {
+ assert(path);
+ assert(!path.endsWith('/' + attachment.ref.relativePath));
+ if (finished) assert(body.includes('TRANSFER_EVIDENCE'));
+ }
+ const delta = finished
+ ? { content: target ? 'target completed' : 'delegation completed' }
+ : {
+ tool_calls: [
+ {
+ index: 0,
+ id: 'delegate-call',
+ type: 'function',
+ function: {
+ name: target ? 'Read' : 'mcp__desktop_workhub__tasks',
+ arguments: JSON.stringify(target ? { path } : {}),
},
- ],
- };
+ },
+ ],
+ };
if (target) {
assert(
data.messages.some(
@@ -102,7 +113,7 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened) {
});
response.writeHead(200, { 'Content-Type': 'text/event-stream', Connection: 'close' });
response.end(
- [chunk(delta, null), chunk({}, target || finished ? 'stop' : 'tool_calls')]
+ [chunk(delta, null), chunk({}, finished ? 'stop' : 'tool_calls')]
.map((event) => 'data: ' + JSON.stringify(event) + '\n\n')
.join('') + 'data: [DONE]\n\n',
);
@@ -152,6 +163,14 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened) {
await request('session.create', { ...createInput(workspace, id, 'bypass'), ...extra });
await request('session.lifecycle.set', { sessionId: 'archived', state: 'archived' });
await request('workhub.coordination.resolve', {});
+ attachment = await upload(
+ request,
+ sessionId,
+ 'transfer',
+ Buffer.from('TRANSFER_EVIDENCE'),
+ 'evidence.txt',
+ 'text/plain',
+ );
const initial = await request('workhub.coordination.candidates', {});
assert.deepEqual(
initial.candidates.map((candidate) => candidate.sessionId),
@@ -210,6 +229,10 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened) {
receipt = await act(input);
assert.equal(receipt.disposition, 'delegate_existing');
assert.equal(receipt.targetSessionId, 'target');
+ await request('artifact.delete', {
+ sessionId,
+ artifactId: attachment.ref.relativePath,
+ });
assert.deepEqual(await act(input), receipt);
await assert.rejects(
act({ ...input, delegationText: 'changed request' }),
@@ -230,6 +253,7 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened) {
await request('workhub.coordination.answer', {
turnId,
text: 'Please implement the requested task',
+ attachments: [attachment],
});
await sourceObserver.waitFor(
(frame) =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org