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/19 14:21:06 UTC
(maka) branch feat/runtime-host-rust updated: refactor(workhub): move ordinary delegation behind typed plugin commands
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 4d9668c782 refactor(workhub): move ordinary delegation behind typed plugin commands
4d9668c782 is described below
commit 4d9668c78256d489b4f5f56736a2fbdde6bb8666
Author: Wang <m4...@gmail.com>
AuthorDate: Sat Sep 19 22:20:52 2026 +0800
refactor(workhub): move ordinary delegation behind typed plugin commands
Generated-by: OpenAI Codex
---
crates/runtime-host/src/execution/workhub.rs | 1 +
.../runtime-host/src/execution/workhub/commands.rs | 54 ++++-
.../src/execution/workhub/delegation.rs | 192 ++++++++++++++++++
.../src/execution/workhub/delegation/receipt.rs | 78 ++++++++
crates/runtime-host/src/plugins/workhub.rs | 1 +
crates/runtime-host/src/plugins/workhub/control.rs | 10 +
.../runtime-host/src/plugins/workhub/delegation.rs | 99 ++++++++++
.../runtime-host/src/plugins/workhub/selection.rs | 1 +
crates/runtime-host/src/server.rs | 15 +-
crates/runtime-host/src/server/projects.rs | 15 +-
crates/runtime-host/src/server/projects/usage.rs | 73 +++++++
crates/runtime-host/src/server/workhub/action.rs | 219 +--------------------
docs/rust-parity.md | 2 +-
docs/rust-parity.zh-CN.md | 2 +-
14 files changed, 526 insertions(+), 236 deletions(-)
diff --git a/crates/runtime-host/src/execution/workhub.rs b/crates/runtime-host/src/execution/workhub.rs
index 3a7af54372..c6c88c285a 100644
--- a/crates/runtime-host/src/execution/workhub.rs
+++ b/crates/runtime-host/src/execution/workhub.rs
@@ -31,6 +31,7 @@ use std::sync::Arc;
use uuid::Uuid;
mod commands;
+mod delegation;
pub(super) mod profile;
mod resume;
pub(crate) use commands::WorkHubCommands;
diff --git a/crates/runtime-host/src/execution/workhub/commands.rs b/crates/runtime-host/src/execution/workhub/commands.rs
index ea9f6b0015..7556413b2d 100644
--- a/crates/runtime-host/src/execution/workhub/commands.rs
+++ b/crates/runtime-host/src/execution/workhub/commands.rs
@@ -30,10 +30,50 @@ use maka_event_log::{
};
use maka_plugins::fiber::Context;
use maka_protocol::OperationErrorCode as Code;
-use std::sync::Weak;
+use std::sync::{Arc, Weak};
-pub(crate) struct WorkHubCommands(pub Weak<Executions>);
+pub(crate) struct WorkHubCommands {
+ executions: Weak<Executions>,
+ pub(super) project_usage: crate::server::ProjectUsage,
+}
+impl WorkHubCommands {
+ pub(crate) fn new(
+ executions: &Arc<Executions>,
+ project_usage: crate::server::ProjectUsage,
+ ) -> Self {
+ Self {
+ executions: Arc::downgrade(executions),
+ project_usage,
+ }
+ }
+}
impl Commands for WorkHubCommands {
+ fn delegation(
+ &self,
+ caller: Context,
+ identity: crate::plugins::workhub::delegation::Identity,
+ ) -> BoxFuture<'_, Result<Option<maka_runtime::workhub::Delegation>>> {
+ Box::pin(async move {
+ let executions = self
+ .executions
+ .upgrade()
+ .ok_or_else(|| failure(Code::HostDraining, "Host is closed"))?;
+ super::delegation::probe(&executions, caller, identity).await
+ })
+ }
+ fn delegate(
+ &self,
+ caller: Context,
+ request: crate::plugins::workhub::delegation::Request,
+ ) -> BoxFuture<'_, Result<maka_runtime::workhub::Delegation>> {
+ Box::pin(async move {
+ let executions = self
+ .executions
+ .upgrade()
+ .ok_or_else(|| failure(Code::HostDraining, "Host is closed"))?;
+ super::delegation::execute(self, &executions, caller, request).await
+ })
+ }
fn target(
&self,
session: String,
@@ -42,7 +82,7 @@ impl Commands for WorkHubCommands {
{
Box::pin(async move {
let executions = self
- .0
+ .executions
.upgrade()
.ok_or_else(|| failure(Code::HostDraining, "Host is closed"))?;
executions.workhub_target(&session, eligible).await
@@ -54,7 +94,7 @@ impl Commands for WorkHubCommands {
) -> BoxFuture<'_, Result<crate::execution::Creation>> {
Box::pin(async move {
let executions = self
- .0
+ .executions
.upgrade()
.ok_or_else(|| failure(Code::HostDraining, "Host is closed"))?;
executions.prepare_session(request).await
@@ -69,7 +109,7 @@ impl Commands for WorkHubCommands {
) -> BoxFuture<'_, Result<crate::plugins::workhub::resume::Receipt>> {
Box::pin(async move {
let executions = self
- .0
+ .executions
.upgrade()
.ok_or_else(|| failure(Code::HostDraining, "Host is closed"))?;
super::resume::execute(&executions, caller, request, connection, eligible).await
@@ -81,7 +121,7 @@ impl Commands for WorkHubCommands {
) -> BoxFuture<'_, Result<Vec<Candidate<SessionConfiguration>>>> {
Box::pin(async move {
let executions = self
- .0
+ .executions
.upgrade()
.ok_or_else(|| failure(Code::HostDraining, "Host is closed"))?;
if executions.shutdown.is_cancelled() {
@@ -97,7 +137,7 @@ impl Commands for WorkHubCommands {
fn stop(&self, caller: Context, request: Stop) -> BoxFuture<'_, Result<StopRecord>> {
Box::pin(async move {
let executions = self
- .0
+ .executions
.upgrade()
.ok_or_else(|| failure(Code::HostDraining, "Host is closed"))?;
let (record, completed, _call) = {
diff --git a/crates/runtime-host/src/execution/workhub/delegation.rs b/crates/runtime-host/src/execution/workhub/delegation.rs
new file mode 100644
index 0000000000..210c256dfd
--- /dev/null
+++ b/crates/runtime-host/src/execution/workhub/delegation.rs
@@ -0,0 +1,192 @@
+/*
+ * 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.
+ */
+
+mod receipt;
+use super::super::{Executions, Result, failure};
+use super::commands::WorkHubCommands;
+use crate::plugins::workhub::{
+ delegation::{Identity, Request},
+ target::Target,
+};
+use maka_plugins::fiber::Context;
+use maka_protocol::OperationErrorCode as Code;
+use maka_runtime::{
+ event::{CommitError, EventWrite, Fact, Invocation, RuntimeEvent},
+ input::InvocationInput,
+ workhub::{Delegation, DelegationDelivery},
+};
+use std::sync::Arc;
+use uuid::Uuid;
+
+pub(super) async fn probe(
+ executions: &Arc<Executions>,
+ caller: Context,
+ identity: Identity,
+) -> Result<Option<Delegation>> {
+ let _gate = executions.lock_admission().await;
+ let _call = caller
+ .admit()
+ .map_err(|error| failure(Code::OperationUnavailable, &error.to_string()))?;
+ let receipt = receipt::read(executions, &identity).await?;
+ if receipt.is_none() {
+ if executions.shutdown.is_cancelled() {
+ return Err(failure(Code::HostDraining, "Host is draining"));
+ }
+ let source = executions.workhub_source(&identity.turn_id).await?;
+ if !matches!(source.root_input(), InvocationInput::Message { .. }) {
+ return Err(failure(
+ Code::OperationConflict,
+ "WorkHub action requires a user message",
+ ));
+ }
+ }
+ Ok(receipt)
+}
+
+pub(super) async fn execute(
+ commands: &WorkHubCommands,
+ executions: &Arc<Executions>,
+ caller: Context,
+ request: Request,
+) -> Result<Delegation> {
+ let mut admission = Some(executions.lock_admission().await);
+ let _call = caller
+ .admit()
+ .map_err(|error| failure(Code::OperationUnavailable, &error.to_string()))?;
+ if let Some(receipt) = receipt::read(executions, &request.identity).await? {
+ return Ok(receipt);
+ }
+ if executions.shutdown.is_cancelled() {
+ return Err(failure(Code::HostDraining, "Host is draining"));
+ }
+ let source = executions.workhub_source(&request.identity.turn_id).await?;
+ if let Some(selected) = &request.selected {
+ if selected.invocation != source.invocation {
+ return Err(failure(
+ Code::OperationConflict,
+ "The selecting Run is no longer active",
+ ));
+ }
+ crate::plugins::workhub::target::validate_selection(selected)?;
+ }
+ let InvocationInput::Message { content, .. } = source.root_input() else {
+ return Err(failure(
+ Code::OperationConflict,
+ "WorkHub action requires a user message",
+ ));
+ };
+ let target = request.target;
+ if let Target::Created { creation, .. } = &target {
+ executions.validate_creation(creation).await?;
+ commands
+ .project_usage
+ .record(&creation.configuration.workspace)
+ .await?;
+ }
+ let owner = executions.active_session_owner(target.id());
+ let delivery = match (&target, &owner) {
+ (
+ Target::Existing {
+ configuration_digest,
+ ..
+ },
+ Some(_),
+ ) => DelegationDelivery::Steering {
+ configuration_digest: configuration_digest.clone(),
+ },
+ (_, None) => DelegationDelivery::NewTurn,
+ (Target::Created { .. }, Some(_)) => {
+ return Err(failure(
+ Code::OperationConflict,
+ "Created target already has an execution owner",
+ ));
+ }
+ };
+ let delegation = Delegation {
+ kind: target.kind(),
+ description: Some(target.description()),
+ delivery,
+ action_id: request.identity.action_id,
+ request_fingerprint: request.identity.fingerprint,
+ source_message_event_id: source.root_opening_event_id().to_owned(),
+ target: owner.unwrap_or_else(|| Invocation {
+ session_id: target.id().to_owned(),
+ turn_id: Uuid::new_v4().to_string(),
+ run_id: Uuid::new_v4().to_string(),
+ invocation_id: Uuid::new_v4().to_string(),
+ }),
+ target_revision: target.revision(),
+ delegation_text: request.text.unwrap_or_else(|| content.text.clone()),
+ };
+ delegation
+ .message(content)
+ .map_err(|reason| failure(Code::OperationUnavailable, reason))?;
+ let action = EventWrite::plain(RuntimeEvent::new(
+ source.invocation,
+ Fact::WorkhubDelegated {
+ delegation: Box::new(delegation.clone()),
+ },
+ ))
+ .map_err(|error| failure(Code::InternalFailure, &error.to_string()))?;
+ let committed = match &target {
+ Target::Created { creation, .. } => executions
+ .log
+ .create_workhub_session(&action, &creation.configuration)
+ .await
+ .map_err(|error| match error {
+ maka_event_log::StoreError::CommitUnknown(_)
+ | maka_event_log::StoreError::OperationUnknown => {
+ CommitError::OutcomeUnknown(error.to_string())
+ }
+ other => CommitError::Rejected(other.to_string()),
+ }),
+ Target::Existing { .. } => executions.log.append(&action).await,
+ };
+ if let Err(error) = committed {
+ return Err(match error {
+ CommitError::OutcomeUnknown(reason) => {
+ executions.begin_drain();
+ failure(Code::CommitOutcomeUnknown, &reason)
+ }
+ CommitError::Rejected(reason) => {
+ let current = executions
+ .log
+ .get_session::<crate::session::SessionConfiguration>(target.id())
+ .await
+ .map_err(|error| super::commands::stored(executions, error))?;
+ let code = if let Target::Existing {
+ configuration_digest,
+ ..
+ } = &target
+ && current.is_none_or(|record| {
+ record.archived || record.configuration_digest != *configuration_digest
+ }) {
+ Code::CandidateSetStale
+ } else {
+ Code::OperationConflict
+ };
+ failure(code, &reason)
+ }
+ });
+ }
+ executions
+ .dispatch_pending(target.id(), &mut admission)
+ .await?;
+ Ok(delegation)
+}
diff --git a/crates/runtime-host/src/execution/workhub/delegation/receipt.rs b/crates/runtime-host/src/execution/workhub/delegation/receipt.rs
new file mode 100644
index 0000000000..7cb1c6311d
--- /dev/null
+++ b/crates/runtime-host/src/execution/workhub/delegation/receipt.rs
@@ -0,0 +1,78 @@
+/*
+ * 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::commands;
+use super::super::{Executions, Result, failure};
+use crate::plugins::workhub::delegation::Identity;
+use maka_protocol::OperationErrorCode as Code;
+use maka_runtime::{
+ event::Fact,
+ workhub::{COORDINATION_SESSION_ID, Delegation},
+};
+
+/// Caller holds admission; an existing canonical action takes precedence over
+/// source liveness, target changes and model availability.
+pub(super) async fn read(
+ executions: &Executions,
+ identity: &Identity,
+) -> Result<Option<Delegation>> {
+ if executions
+ .log
+ .workhub_stop(&identity.action_id)
+ .await
+ .map_err(|error| commands::stored(executions, error))?
+ .is_some()
+ || executions
+ .log
+ .workhub_correction(&identity.action_id)
+ .await
+ .map_err(|error| commands::stored(executions, error))?
+ .is_some()
+ {
+ return Err(failure(
+ Code::OperationConflict,
+ "WorkHub action already belongs to a control operation",
+ ));
+ }
+ // Receipt authority survives source termination, model removal and target changes.
+ if let Some(stored) = executions
+ .log
+ .workhub_action(&identity.action_id)
+ .await
+ .map_err(|error| commands::stored(executions, error))?
+ {
+ let Fact::WorkhubDelegated { delegation } = stored.event.fact else {
+ return Err(failure(
+ Code::OperationConflict,
+ "WorkHub action belongs to another operation",
+ ));
+ };
+ if stored.event.invocation.session_id != COORDINATION_SESSION_ID
+ || stored.event.invocation.turn_id != identity.turn_id
+ || delegation.request_fingerprint != identity.fingerprint
+ {
+ return Err(failure(
+ Code::OperationConflict,
+ "WorkHub action belongs to another request",
+ ));
+ }
+ return Ok(Some(*delegation));
+ }
+ Ok(None)
+}
diff --git a/crates/runtime-host/src/plugins/workhub.rs b/crates/runtime-host/src/plugins/workhub.rs
index cb78198779..833d6db6df 100644
--- a/crates/runtime-host/src/plugins/workhub.rs
+++ b/crates/runtime-host/src/plugins/workhub.rs
@@ -32,6 +32,7 @@ use serde_json::Value;
use std::sync::Arc;
pub(crate) mod candidates;
pub(crate) mod control;
+pub(crate) mod delegation;
pub(crate) mod resume;
pub(crate) mod selection;
pub(crate) mod target;
diff --git a/crates/runtime-host/src/plugins/workhub/control.rs b/crates/runtime-host/src/plugins/workhub/control.rs
index 2485ef1b16..7586ae872e 100644
--- a/crates/runtime-host/src/plugins/workhub/control.rs
+++ b/crates/runtime-host/src/plugins/workhub/control.rs
@@ -43,6 +43,16 @@ pub(crate) struct Stop {
/// Commands own admission, durable receipts and settlement. No locks, arbitrary
/// log writes or Host handles are exposed to the business implementation.
pub(crate) trait Commands: Send + Sync {
+ fn delegation(
+ &self,
+ caller: Context,
+ identity: super::delegation::Identity,
+ ) -> BoxFuture<'_, Result<Option<maka_runtime::workhub::Delegation>>>;
+ fn delegate(
+ &self,
+ caller: Context,
+ request: super::delegation::Request,
+ ) -> BoxFuture<'_, Result<maka_runtime::workhub::Delegation>>;
/// Apply the domain predicate inside one bounded read snapshot, before its limit.
fn candidates(
&self,
diff --git a/crates/runtime-host/src/plugins/workhub/delegation.rs b/crates/runtime-host/src/plugins/workhub/delegation.rs
new file mode 100644
index 0000000000..781c5c0a71
--- /dev/null
+++ b/crates/runtime-host/src/plugins/workhub/delegation.rs
@@ -0,0 +1,99 @@
+/*
+ * 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::{
+ control::{Result, failure, fingerprint},
+ selection::SelectedTarget,
+ target::Target,
+};
+use maka_protocol::{
+ OperationErrorCode as Code,
+ workhub::{ActInput, ActResult, Proposal},
+};
+use maka_runtime::workhub::{ActionId, Delegation, DelegationKind};
+
+#[derive(Clone)]
+pub(crate) struct Identity {
+ pub action_id: ActionId,
+ pub turn_id: String,
+ pub fingerprint: String,
+}
+pub(crate) struct Request {
+ pub identity: Identity,
+ pub target: Target,
+ pub selected: Option<SelectedTarget>,
+ pub text: Option<String>,
+}
+
+impl super::Control {
+ pub(crate) async fn delegate(
+ &self,
+ input: ActInput,
+ selected: Option<SelectedTarget>,
+ ) -> Result<ActResult> {
+ if !matches!(input.proposal, Proposal::Route(_)) {
+ return Err(failure(
+ Code::OperationConflict,
+ "Not an ordinary WorkHub delegation",
+ ));
+ }
+ let identity = Identity {
+ action_id: input.action_id.clone(),
+ turn_id: input.turn_id.clone(),
+ fingerprint: fingerprint(&input)?,
+ };
+ if let Some(delegation) = self
+ .commands
+ .delegation(self.caller.clone(), identity.clone())
+ .await?
+ {
+ return Ok(receipt(&delegation));
+ }
+ let target = self.prepare_target(&input, selected.as_ref()).await?;
+ let delegation = self
+ .commands
+ .delegate(
+ self.caller.clone(),
+ Request {
+ identity,
+ target,
+ selected,
+ text: input.delegation_text,
+ },
+ )
+ .await?;
+ Ok(receipt(&delegation))
+ }
+}
+
+fn receipt(delegation: &Delegation) -> ActResult {
+ let target_session_id = delegation.target.session_id.clone();
+ let target_turn_id = delegation.target.turn_id.clone();
+ match delegation.kind {
+ DelegationKind::Existing => ActResult::DelegateExisting {
+ target_session_id,
+ target_turn_id,
+ steered: delegation.delivery.is_steering(),
+ },
+ DelegationKind::Created => ActResult::CreateNew {
+ target_session_id,
+ target_turn_id,
+ },
+ }
+}
diff --git a/crates/runtime-host/src/plugins/workhub/selection.rs b/crates/runtime-host/src/plugins/workhub/selection.rs
index 71e0dff72a..a4c35c30f4 100644
--- a/crates/runtime-host/src/plugins/workhub/selection.rs
+++ b/crates/runtime-host/src/plugins/workhub/selection.rs
@@ -31,6 +31,7 @@ use maka_runtime::{
};
pub(crate) mod offer;
+#[derive(Clone)]
pub(crate) struct SelectedTarget {
pub invocation: Invocation,
pub candidate_ref: String,
diff --git a/crates/runtime-host/src/server.rs b/crates/runtime-host/src/server.rs
index 39ac742941..ebe5a65bb5 100644
--- a/crates/runtime-host/src/server.rs
+++ b/crates/runtime-host/src/server.rs
@@ -43,6 +43,7 @@ mod outbound;
mod plugin_remote;
mod projects;
mod scheduler;
+pub(crate) use projects::Usage as ProjectUsage;
pub(crate) use projects::resolve_record as resolve_project_workspace;
mod registration;
mod resources;
@@ -124,13 +125,14 @@ pub struct Host {
interactions: Arc<interactions::Interactions>,
uploads: artifacts::Uploads,
project_directories: projects::Directories,
+ project_usage: ProjectUsage,
// Log connections close before root authority is released.
log: Arc<EventLog>,
configuration: Arc<ConfigurationStore>,
connection_effects: connection_effects::ConnectionEffects,
oauth: oauth::Coordinator,
changes: broadcast::Sender<serde_json::Value>,
- change_revision: AtomicU64,
+ change_revision: Arc<AtomicU64>,
access_revocations: broadcast::Sender<String>,
access_changed: tokio::sync::Notify,
session_catalog: Arc<catalog_feed::CatalogFeed>,
@@ -214,6 +216,9 @@ impl Host {
let epoch = Uuid::new_v4().to_string();
log.begin_message_epoch(&epoch).await?;
let changes = broadcast::channel(64).0;
+ let change_revision = Arc::new(AtomicU64::new(0));
+ let project_usage =
+ ProjectUsage::new(log.clone(), changes.clone(), change_revision.clone());
let session_catalog = Arc::new(catalog_feed::CatalogFeed::new(
*log.subscribe_commits().borrow(),
));
@@ -259,9 +264,10 @@ impl Host {
crate::plugins::workhub::install(
&mut setup,
&executions.plugin_catalog,
- Arc::new(crate::execution::WorkHubCommands(Arc::downgrade(
+ Arc::new(crate::execution::WorkHubCommands::new(
&executions,
- ))),
+ project_usage.clone(),
+ )),
)?;
crate::plugins::scheduler::install(
&mut setup,
@@ -326,12 +332,13 @@ impl Host {
interactions,
uploads: Default::default(),
project_directories,
+ project_usage,
log,
configuration,
connection_effects: connection_effects::ConnectionEffects::default(),
oauth: oauth::Coordinator::default(),
changes,
- change_revision: AtomicU64::new(0),
+ change_revision,
access_revocations: broadcast::channel(64).0,
access_changed: tokio::sync::Notify::new(),
session_catalog,
diff --git a/crates/runtime-host/src/server/projects.rs b/crates/runtime-host/src/server/projects.rs
index 656cc6a007..e28948bb98 100644
--- a/crates/runtime-host/src/server/projects.rs
+++ b/crates/runtime-host/src/server/projects.rs
@@ -19,6 +19,7 @@
mod directories;
mod projection;
+mod usage;
use super::{Host, HostError};
use crate::session::SessionConfiguration;
pub(super) use directories::Directories;
@@ -32,6 +33,7 @@ use maka_protocol::session::{WorkspaceProjection, WorkspaceTarget};
use maka_protocol::{Operation, OperationError, OperationErrorCode as Code, Outcome, project::*};
use serde_json::{Value, json};
use std::{path::Path, sync::atomic::Ordering};
+pub(crate) use usage::Usage;
type Result<T> = std::result::Result<T, OperationError>;
@@ -235,18 +237,7 @@ pub(crate) async fn resolve_record(
}
pub(super) async fn record_usage(host: &Host, workspace: &WorkspaceProjection) -> Result<()> {
- if let WorkspaceTarget::Project { project_id } = &workspace.target {
- host.log
- .touch_project(
- project_id,
- &workspace.host_cwd,
- super::configuration::now().map_err(internal)?,
- )
- .await
- .map_err(stored)?;
- publish(host);
- }
- Ok(())
+ host.project_usage.record(workspace).await
}
fn publish(host: &Host) {
diff --git a/crates/runtime-host/src/server/projects/usage.rs b/crates/runtime-host/src/server/projects/usage.rs
new file mode 100644
index 0000000000..c4a4d14d1b
--- /dev/null
+++ b/crates/runtime-host/src/server/projects/usage.rs
@@ -0,0 +1,73 @@
+/*
+ * 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::{internal, stored};
+use maka_event_log::EventLog;
+use maka_protocol::{
+ OperationError,
+ session::{WorkspaceProjection, WorkspaceTarget},
+};
+use serde_json::{Value, json};
+use std::sync::{
+ Arc,
+ atomic::{AtomicU64, Ordering},
+};
+use tokio::sync::broadcast;
+
+/// Project usage and its notification share the Host's existing catalog revision.
+#[derive(Clone)]
+pub(crate) struct Usage {
+ log: Arc<EventLog>,
+ changes: broadcast::Sender<Value>,
+ revision: Arc<AtomicU64>,
+}
+impl Usage {
+ pub(crate) fn new(
+ log: Arc<EventLog>,
+ changes: broadcast::Sender<Value>,
+ revision: Arc<AtomicU64>,
+ ) -> Self {
+ Self {
+ log,
+ changes,
+ revision,
+ }
+ }
+
+ pub(crate) async fn record(
+ &self,
+ workspace: &WorkspaceProjection,
+ ) -> Result<(), OperationError> {
+ if let WorkspaceTarget::Project { project_id } = &workspace.target {
+ self.log
+ .touch_project(
+ project_id,
+ &workspace.host_cwd,
+ super::super::configuration::now().map_err(internal)?,
+ )
+ .await
+ .map_err(stored)?;
+ let revision = self.revision.fetch_add(1, Ordering::SeqCst) + 1;
+ let _ = self
+ .changes
+ .send(json!({"kind":"project.catalog.changed", "revision":revision}));
+ }
+ Ok(())
+ }
+}
diff --git a/crates/runtime-host/src/server/workhub/action.rs b/crates/runtime-host/src/server/workhub/action.rs
index 1c6a779f68..65606b1786 100644
--- a/crates/runtime-host/src/server/workhub/action.rs
+++ b/crates/runtime-host/src/server/workhub/action.rs
@@ -17,16 +17,11 @@
* under the License.
*/
-use super::{Host, failure, record, sessions};
+use super::{Host, failure, sessions};
use maka_protocol::{
OperationError, OperationErrorCode as Code,
workhub::{ActInput, ActResult},
};
-use maka_runtime::{
- event::{CommitError, EventWrite, Fact, Invocation, RuntimeEvent},
- input::InvocationInput,
- workhub::{COORDINATION_SESSION_ID, Delegation, DelegationDelivery, DelegationKind},
-};
use std::sync::Arc;
use uuid::Uuid;
@@ -65,7 +60,9 @@ pub(super) async fn act(
super::control(host)?.value.stop(input).await
}
},
- maka_protocol::workhub::Proposal::Route(_) => admit(host, input, None).await,
+ maka_protocol::workhub::Proposal::Route(_) => {
+ super::control(host)?.value.delegate(input, None).await
+ }
}
}
@@ -74,210 +71,10 @@ pub(super) async fn selected(
input: ActInput,
selected: &super::selection::SelectedTarget,
) -> Result<ActResult, OperationError> {
- admit(host, input, Some(selected)).await
-}
-
-async fn admit(
- host: &Arc<Host>,
- input: ActInput,
- selected: Option<&super::selection::SelectedTarget>,
-) -> Result<ActResult, OperationError> {
- let fingerprint = fingerprint(&input)?;
- let mut prepared = None;
- let mut provider = None;
- loop {
- let mut admission = Some(host.executions.lock_admission().await);
- if host
- .log
- .workhub_stop(&input.action_id)
- .await
- .map_err(sessions::stored)?
- .is_some()
- || host
- .log
- .workhub_correction(&input.action_id)
- .await
- .map_err(sessions::stored)?
- .is_some()
- {
- return Err(failure(
- Code::OperationConflict,
- "WorkHub action already belongs to a control operation",
- ));
- }
- // Receipt authority survives source termination, model removal and target changes.
- if let Some(stored) = host
- .log
- .workhub_action(&input.action_id)
- .await
- .map_err(sessions::stored)?
- {
- let Fact::WorkhubDelegated { delegation } = stored.event.fact else {
- return Err(failure(
- Code::OperationConflict,
- "WorkHub action belongs to another operation",
- ));
- };
- if stored.event.invocation.session_id != COORDINATION_SESSION_ID
- || stored.event.invocation.turn_id != input.turn_id
- || delegation.request_fingerprint != fingerprint
- {
- return Err(failure(
- Code::OperationConflict,
- "WorkHub action belongs to another request",
- ));
- }
- return Ok(receipt(&delegation));
- }
- let control = match &provider {
- Some(control) => control,
- None => provider.insert(super::control(host)?),
- };
- let _call = control
- .admit()
- .map_err(|error| failure(Code::OperationUnavailable, error.to_string()))?;
- if host.draining.is_cancelled() {
- return Err(failure(Code::HostDraining, "Host is draining"));
- }
- record(host)
- .await?
- .ok_or_else(|| failure(Code::NotFound, "WorkHub Session has not been resolved"))?;
- let source = host.executions.workhub_source(&input.turn_id).await?;
- if selected.is_some_and(|selected| selected.invocation != source.invocation) {
- return Err(failure(
- Code::OperationConflict,
- "The selecting Run is no longer active",
- ));
- }
- let InvocationInput::Message { content, .. } = source.root_input() else {
- return Err(failure(
- Code::OperationConflict,
- "WorkHub action requires a user message",
- ));
- };
- let Some(target) = prepared.take() else {
- drop(admission.take());
- prepared = Some(control.value.prepare_target(&input, selected).await?);
- continue;
- };
- target::validate(host, &target).await?;
- if let Some(selected) = selected {
- crate::plugins::workhub::target::validate_selection(selected)?;
- }
- let owner = host.executions.active_session_owner(target.id());
- let delivery = match (&target, &owner) {
- (
- target::Target::Existing {
- configuration_digest,
- ..
- },
- Some(_),
- ) => DelegationDelivery::Steering {
- configuration_digest: configuration_digest.clone(),
- },
- (_, None) => DelegationDelivery::NewTurn,
- (target::Target::Created { .. }, Some(_)) => {
- return Err(failure(
- Code::OperationConflict,
- "Created target already has an execution owner",
- ));
- }
- };
- let delegation = Delegation {
- kind: target.kind(),
- description: Some(target.description()),
- delivery,
- action_id: input.action_id,
- request_fingerprint: fingerprint,
- source_message_event_id: source.root_opening_event_id().to_owned(),
- target: owner.unwrap_or_else(|| Invocation {
- session_id: target.id().to_owned(),
- turn_id: Uuid::new_v4().to_string(),
- run_id: Uuid::new_v4().to_string(),
- invocation_id: Uuid::new_v4().to_string(),
- }),
- target_revision: target.revision(),
- delegation_text: input
- .delegation_text
- .unwrap_or_else(|| content.text.clone()),
- };
- delegation
- .message(content)
- .map_err(|reason| failure(Code::OperationUnavailable, reason))?;
- let result = receipt(&delegation);
- let action = EventWrite::plain(RuntimeEvent::new(
- source.invocation,
- Fact::WorkhubDelegated {
- delegation: Box::new(delegation),
- },
- ))
- .map_err(|error| failure(Code::InternalFailure, error.to_string()))?;
- let committed = match &target {
- target::Target::Created { creation, .. } => host
- .log
- .create_workhub_session(&action, &creation.configuration)
- .await
- .map_err(|error| match error {
- maka_event_log::StoreError::CommitUnknown(_)
- | maka_event_log::StoreError::OperationUnknown => {
- CommitError::OutcomeUnknown(error.to_string())
- }
- other => CommitError::Rejected(other.to_string()),
- }),
- target::Target::Existing { .. } => host.log.append(&action).await,
- };
- if let Err(error) = committed {
- return Err(match error {
- CommitError::OutcomeUnknown(reason) => {
- host.executions.begin_drain();
- failure(Code::CommitOutcomeUnknown, reason)
- }
- CommitError::Rejected(reason) => {
- let current = host
- .log
- .get_session::<crate::session::SessionConfiguration>(target.id())
- .await
- .map_err(sessions::stored)?;
- let code = if let target::Target::Existing {
- configuration_digest,
- ..
- } = &target
- && current.is_none_or(|record| {
- record.archived || record.configuration_digest != *configuration_digest
- }) {
- Code::CandidateSetStale
- } else {
- Code::OperationConflict
- };
- failure(code, reason)
- }
- });
- }
- host.executions
- .dispatch_pending(target.id(), &mut admission)
- .await?;
- return Ok(result);
- }
-}
-
-fn receipt(delegation: &Delegation) -> ActResult {
- let target_session_id = delegation.target.session_id.clone();
- let target_turn_id = delegation.target.turn_id.clone();
- match delegation.kind {
- DelegationKind::Existing => ActResult::DelegateExisting {
- target_session_id,
- target_turn_id,
- steered: delegation.delivery.is_steering(),
- },
- DelegationKind::Created => ActResult::CreateNew {
- target_session_id,
- target_turn_id,
- },
- }
-}
-
-fn fingerprint(input: &ActInput) -> Result<String, OperationError> {
- crate::plugins::workhub::control::fingerprint(input)
+ super::control(host)?
+ .value
+ .delegate(input, Some(selected.clone()))
+ .await
}
fn stored(host: &Host, error: maka_event_log::StoreError) -> OperationError {
diff --git a/docs/rust-parity.md b/docs/rust-parity.md
index f91fcf69e6..84839d57e1 100644
--- a/docs/rust-parity.md
+++ b/docs/rust-parity.md
@@ -57,7 +57,7 @@ Client changes are allowed when they remove an obsolete path, as with Graph.
| Domain / current coupling | Target and completion condition |
| --- | --- |
| **Skills:** `maka.skills` owns discovery, input preparation, per-step tool/context snapshots, governance, preference CAS, preview, import and workspace/user publication. Published Client Contributions own Session/new-workspace pickers, management and draft suggestions. | Desktop supplies target-bound Slots, generic Remote transport and authorized native file actions. The old scanner, importer, controller and Skills IPC/preload facade are removed. Host retains thin external protocol ada [...]
-| **WorkHub:** `maka.workhub` owns prompt/client-tool policy, Stop/Resume request/results, candidate filtering/references, selection forms/answers and target/default choices. Host retains exact-owner cancellation, bounded snapshot reads, canonical forms and final admission. Workflow orchestration, correction, recovery and UI remain to migrate. | Move the remaining workflow, queries, associations, selection, correction/resume policy and UI into the built-in domain. Host commands retain ex [...]
+| **WorkHub:** `maka.workhub` owns prompt/client-tool policy, ordinary delegation, Stop/Resume request/results, candidate filtering/references, selection forms/answers and target/default choices. Target preparation runs outside admission; Host commands recheck the instance and canonical source before atomic delivery. | Move correction/recovery, coordinator workflows, native tools, Remote and actual Desktop UI into the built-in domain. Host retains exact-message targeting, canonical forms [...]
| **Default assistant behavior:** `maka.assistant` publishes the default behavior, persona, personalization and workspace instructions. | Sources are frozen per logical model step; disabling the plugin removes its persona. Explicit Session/child instructions remain independent of replaceable prompt Contributions. Execution/compaction invariants remain Host-owned. |
| **Graph/Swarm and Scheduler:** already built-in plugins; behavior selection uses an open typed identity. Graph uses atomic activate/stop/retire-idle commands and read-only preferences, without `Executions`, configuration writers or Host locks. | Preserve existing orchestration and wakeup behavior. Reuse these narrow commands where semantics match; review Scheduler against the same ownership rule. Typed domain repositories remain legitimate. |
| **Code Mode:** mode selection, nested dispatch and history projection cross several crates. | After the first domain migrations, move its user-facing tool and mode policy where a real contribution boundary helps. Keep V8 ownership, nested-call permissions, dispatch/settlement and canonical history in the runtime. Do not invent a universal executor hook merely to move `exec`. |
diff --git a/docs/rust-parity.zh-CN.md b/docs/rust-parity.zh-CN.md
index e1fcf7e21a..70445b5464 100644
--- a/docs/rust-parity.zh-CN.md
+++ b/docs/rust-parity.zh-CN.md
@@ -50,7 +50,7 @@
| 领域／当前耦合 | 目标与完成条件 |
| --- | --- |
| **Skills：**`maka.skills` 拥有发现、输入准备、每步工具／上下文快照、治理、偏好 CAS、预览、导入及 workspace／user 发布；已发布的 Client Contribution 拥有 Session／新工作区选择器、管理页与草稿建议。 | Desktop 提供目标绑定的 Slot、通用 Remote 传输及授权的原生文件操作；旧扫描、导入、控制器和 Skills IPC／preload 门面已删除。Host 仅保留薄外部协议适配、准入与不可变回执，不再解析 Skill。 |
-| **WorkHub：**`maka.workhub` 拥有提示词／客户端工具策略、Stop／Resume 请求／结果、候选筛选／引用、选择表单／答案及目标／默认值决策。Host 保留精确 owner 取消、有界快照读取、规范表单及最终准入；流程编排、纠正、恢复和 UI 仍待迁移。 | 剩余流程、查询、关联、多选、纠正／resume 策略及 UI 迁入内置领域；Host 命令保留精确消息定位、原子准入／取消／回执及结算。复用领域记录与事务，不能把原子纠正拆成插件和 Host 各自写入。 |
+| **WorkHub：**`maka.workhub` 拥有提示词／客户端工具策略、普通委派、Stop／Resume 请求／结果、候选筛选／引用、选择表单／答案及目标／默认值决策。目标准备在准入锁外运行；Host 命令在原子投递前复查实例及 canonical source。 | 纠正／恢复、协调会话流程、原生工具、Remote 及实际 Desktop UI 迁入内置领域。Host 保留精确消息定位、规范表单、原子准入／取消／回执及结算；不能把纠正拆成插件和 Host 各自写入。 |
| **默认助手行为：**`maka.assistant` 发布默认 behavior、persona、个性化和工作区指令。 | 每逻辑模型步骤冻结来源；停用后不保留隐藏 persona。显式 Session／子任务指令独立于可替换的提示词 Contribution；执行／压缩不变量仍由 Host 维护。 |
| **Graph／Swarm 与 Scheduler：**已经是内置插件，behavior 按开放的类型化身份选择。Graph 使用原子激活／停止／空闲退休命令及只读偏好，不接收 `Executions`、配置写入器或 Host 锁。 | 保持已有编排与唤醒行为；语义相同时复用这些窄命令，按同一归属规则复核 Scheduler。类型化领域 repository 可以保留。 |
| **Code Mode：**模式选择、嵌套派发及历史投影跨越多个 crate。 | 首批领域迁移后，在有实际 Contribution 边界收益时迁移面向用户的工具和模式策略；V8 所有权、嵌套调用权限、派发／结算及规范历史保留 runtime。不为搬迁 `exec` 发明万能执行 hook。 |
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org