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/15 16:03:44 UTC
(maka) branch rust-preview updated: feat(runtime): share independent hosts across product UIs
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 9e2a192491 feat(runtime): share independent hosts across product UIs
9e2a192491 is described below
commit 9e2a1924914fb5cfcb22b7917ceb06e0dde759d8
Author: Wang <m4...@gmail.com>
AuthorDate: Fri Oct 16 00:03:23 2026 +0800
feat(runtime): share independent hosts across product UIs
Connect product frontends through authenticated local peer sessions while keeping business Root ownership in the Host. Drain original UI calls before detaching, preserve background work through explicit UI-only exit, and stop the Host on normal last-client quit.
Cover durable launcher recovery, concurrent clients, canceled control requests, replacement epochs, real terminal reconnection and native shutdown.
Generated-by: Codex
---
platform/event-log/src/applications.rs | 30 +-
platform/event-log/src/applications/launch.rs | 50 ++
.../event-log/src/applications/launch/tests.rs | 165 +++++
platform/host-runtime/src/application.rs | 13 +-
platform/host-runtime/src/application/host.rs | 154 +++-
.../host-runtime/src/application/host/access.rs | 4 +-
.../host-runtime/src/application/host/client.rs | 117 ++-
.../host-runtime/src/application/host/protocol.rs | 45 +-
.../src/application/transport/policy.rs | 18 +-
.../src/application/transport/server.rs | 52 +-
.../host-runtime/src/application/transport/wire.rs | 11 -
platform/render-terminal/src/application/run.rs | 168 ++++-
platform/transport/Cargo.toml | 2 +-
platform/transport/README.md | 4 +-
platform/transport/README.zh-CN.md | 4 +-
platform/transport/src/connections.rs | 263 ++++---
platform/transport/src/connections/host.rs | 140 ++--
platform/transport/src/lib.rs | 13 +-
platform/transport/src/startup.rs | 110 +++
platform/transport/src/transport.rs | 4 +-
platform/transport/src/transport/ipc.rs | 52 +-
platform/transport/src/transport/native.rs | 5 +-
platform/transport/src/transport/process.rs | 785 ---------------------
platform/transport/tests/owners.rs | 8 +-
platform/transport/tests/private.rs | 154 ++--
platform/transport/tests/public_host.rs | 578 ++++++++++++++-
plugins/shared/tui-messages.ts | 6 +-
products/maka/bundle/src/composition.ts | 2 +-
products/maka/cli/Cargo.toml | 2 +-
products/maka/cli/README.md | 4 +-
products/maka/cli/README.zh-CN.md | 4 +-
products/maka/cli/src/application.rs | 59 +-
products/maka/cli/src/application/product.rs | 32 +-
.../maka/cli/src/application/product/headless.rs | 107 ++-
.../maka/cli/src/application/product/launch.rs | 145 +++-
.../cli/src/application/product/launch/tests.rs | 28 +-
.../maka/cli/src/application/product/public.rs | 406 +++++++++++
.../maka/cli/src/application/product/restore.rs | 112 ++-
.../cli/src/application/product/test_public.rs | 175 +++++
.../maka/cli/src/application/product/update.rs | 38 +-
.../cli/src/application/product/update/tests.rs | 46 +-
products/maka/cli/src/application/serve.rs | 61 +-
products/maka/cli/src/args.rs | 6 +-
products/maka/cli/src/deployment.rs | 2 +-
products/maka/cli/src/deployment/local.rs | 67 +-
products/maka/cli/src/main.rs | 1 +
products/maka/cli/src/startup.rs | 89 +++
.../cli/tests/integration/terminal/application.rs | 304 +++++++-
.../tests/integration/terminal/bundled_product.rs | 59 +-
.../integration/terminal/bundled_product/chat.rs | 163 +++++
.../terminal/bundled_product/chat/code_mode.rs | 3 +
.../terminal/bundled_product/chat/computer.rs | 3 +
.../terminal/bundled_product/chat/goal.rs | 3 +
.../terminal/bundled_product/chat/graph.rs | 3 +
.../terminal/bundled_product/chat/health.rs | 3 +
.../terminal/bundled_product/chat/performance.rs | 11 +-
.../terminal/bundled_product/chat/plan.rs | 4 +
.../terminal/bundled_product/chat/recap.rs | 3 +
.../terminal/bundled_product/chat/title.rs | 3 +
.../terminal/bundled_product/chat/todo.rs | 3 +
products/maka/desktop/README.md | 2 +-
products/maka/desktop/README.zh-CN.md | 2 +-
products/maka/desktop/examples/native_lifecycle.rs | 179 ++++-
products/maka/desktop/src/actions.rs | 41 +-
products/maka/desktop/src/editor_tests.rs | 30 +-
products/maka/desktop/src/editor_tests/cohost.rs | 4 +-
products/maka/desktop/src/editor_tests/reading.rs | 3 +-
products/maka/desktop/src/main.rs | 47 +-
.../assets/starter/verify-host.py | 91 ++-
.../maka-plugin-authoring/references/runtime-ui.md | 2 +-
70 files changed, 3764 insertions(+), 1543 deletions(-)
diff --git a/platform/event-log/src/applications.rs b/platform/event-log/src/applications.rs
index 116ca69dd6..9965e75adb 100644
--- a/platform/event-log/src/applications.rs
+++ b/platform/event-log/src/applications.rs
@@ -108,6 +108,10 @@ fn validate(
observations: &[RemoteObservation],
bytes: &[Package],
) -> Result<Plan, StoreError> {
+ let packages = validate_packages(bytes)?;
+ validate_plan(input, observations, &packages)
+}
+fn validate_packages(bytes: &[Package]) -> Result<Packages, StoreError> {
let mut packages = Packages::default();
let mut references = BTreeSet::new();
for package in bytes {
@@ -123,6 +127,13 @@ fn validate(
.insert(package.application_ref(), declaration.clone())
.map_err(|e| invalid(e.to_string()))?;
}
+ Ok(packages)
+}
+fn validate_plan(
+ input: &Input,
+ observations: &[RemoteObservation],
+ packages: &Packages,
+) -> Result<Plan, StoreError> {
let mut remote = RemoteExports::new();
for observation in observations {
if remote
@@ -135,7 +146,7 @@ fn validate(
return Err(invalid("duplicate remote capability observation"));
}
}
- Plan::prepare(input, &packages, &remote).map_err(|e| invalid(e.to_string()))
+ Plan::prepare(input, packages, &remote).map_err(|e| invalid(e.to_string()))
}
fn check_document(
input: &Input,
@@ -168,6 +179,23 @@ pub fn validate_application(
)?;
validate(input, observations, bytes).map(|_| ())
}
+
+/// Check several sources against the same immutable package set. Each source
+/// keeps its own document budget, remote observations and resulting plan.
+pub fn validate_applications(
+ sources: &[(&Input, &[RemoteObservation])],
+ bytes: &[Package],
+) -> Result<Vec<Plan>, StoreError> {
+ let references: Vec<_> = bytes.iter().map(Package::application_ref).collect();
+ for (input, observations) in sources {
+ check_document(input, observations, references.clone())?;
+ }
+ let packages = validate_packages(bytes)?;
+ sources
+ .iter()
+ .map(|(input, observations)| validate_plan(input, observations, &packages))
+ .collect()
+}
impl EventLog {
fn application_domain(&self) -> Result<String, StoreError> {
self.validate_root()?;
diff --git a/platform/event-log/src/applications/launch.rs b/platform/event-log/src/applications/launch.rs
index c09d2220ef..2ab1db4b7a 100644
--- a/platform/event-log/src/applications/launch.rs
+++ b/platform/event-log/src/applications/launch.rs
@@ -35,6 +35,33 @@ pub struct Launch {
pub pending: Option<Vec<u8>>,
}
impl EventLog {
+ /// Trusted Host configuration changes are separate from source replacement.
+ /// Check both authorities in one transaction; the same desired value is an
+ /// idempotent retry only while the observed source is still exactly current.
+ pub async fn compare_and_set_application_launch_configuration(
+ &self,
+ source: maka_plugin_core::deployment::DeploymentRef,
+ expected: Value,
+ configuration: Value,
+ ) -> Result<bool, StoreError> {
+ source.validate().map_err(|error| invalid(&error))?;
+ if source.state_domain != self.application_domain()? {
+ return Err(invalid("foreign Host configuration source"));
+ }
+ let expected = encode(&expected, MAX_CONFIGURATION)?;
+ let configuration = encode(&configuration, MAX_CONFIGURATION)?;
+ self.connection.run(move |connection| Box::pin(async move {
+ let mut tx = connection.begin_with("BEGIN IMMEDIATE").await?;
+ let changed = sqlx::query("UPDATE application_launch SET configuration=? WHERE singleton=1 AND deployment=? AND pending IS NULL AND (configuration=? OR configuration=?) AND EXISTS(SELECT 1 FROM application_deployments WHERE id=? AND revision=?)")
+ .bind(&configuration).bind(source.deployment.to_string())
+ .bind(expected).bind(&configuration)
+ .bind(source.deployment.to_string()).bind(source.revision as i64)
+ .execute(&mut *tx).await?.rows_affected() == 1;
+ tx.commit().await.map_err(StoreError::CommitUnknown)?;
+ Ok(changed)
+ })).await
+ }
+
/// Native product authoring inputs and the last installed default sources.
pub async fn application_product(&self) -> Result<Option<Value>, StoreError> {
self.application_domain()?;
@@ -104,6 +131,29 @@ impl EventLog {
})).await
}
+ /// Select an existing committed source without starting source preparation.
+ /// The observed source must still be current, and an existing selection is
+ /// never overwritten.
+ pub async fn select_committed_application_launch(
+ &self,
+ source: maka_plugin_core::deployment::DeploymentRef,
+ configuration: Value,
+ ) -> Result<(), StoreError> {
+ source.validate().map_err(|error| invalid(&error))?;
+ if source.state_domain != self.application_domain()? {
+ return Err(invalid("foreign launcher source"));
+ }
+ let configuration = encode(&configuration, MAX_CONFIGURATION)?;
+ self.connection.run(move |connection| Box::pin(async move {
+ let mut tx = connection.begin_with("BEGIN IMMEDIATE").await?;
+ let changed = sqlx::query("INSERT INTO application_launch (singleton,deployment,configuration,pending) SELECT 1,id,?,NULL FROM application_deployments WHERE id=? AND revision=?")
+ .bind(configuration).bind(source.deployment.to_string()).bind(source.revision as i64)
+ .execute(&mut *tx).await?.rows_affected();
+ if changed != 1 { return Err(invalid("committed launcher source changed or is missing")); }
+ tx.commit().await.map_err(StoreError::CommitUnknown)
+ })).await
+ }
+
/// Create once, before any cross-Root source effects. An unknown COMMIT
/// remains unknown; reopening reads this same record instead of reauthoring.
pub async fn begin_application_launch(&self, launch: Launch) -> Result<(), StoreError> {
diff --git a/platform/event-log/src/applications/launch/tests.rs b/platform/event-log/src/applications/launch/tests.rs
index cb74d2c48b..51ba4a07bc 100644
--- a/platform/event-log/src/applications/launch/tests.rs
+++ b/platform/event-log/src/applications/launch/tests.rs
@@ -211,3 +211,168 @@ async fn product_baseline_and_connections_publish_atomically_with_completion() {
assert_eq!(current.pending, Some(vec![9]));
store.close().await.unwrap();
}
+
+#[tokio::test]
+async fn host_configuration_cas_checks_source_and_original_authority_on_retries() {
+ let (_directory, store) = crate::applications::tests::store().await;
+ let deployment = Uuid::new_v4();
+ let source = store
+ .commit_application(
+ deployment,
+ 0,
+ crate::applications::tests::input(),
+ vec![],
+ vec![],
+ )
+ .await
+ .unwrap();
+ let previous = json!({"kind":"public_host","native":[],"endpoints":{"old":true}});
+ let desired = json!({"kind":"public_host","native":[],"endpoints":{"new":true}});
+ store
+ .begin_application_launch(Launch {
+ deployment,
+ configuration: previous.clone(),
+ pending: Some(vec![1]),
+ })
+ .await
+ .unwrap();
+ assert!(
+ !store
+ .compare_and_set_application_launch_configuration(
+ source.clone(),
+ previous.clone(),
+ desired.clone()
+ )
+ .await
+ .unwrap()
+ );
+ store.complete_application_launch(deployment).await.unwrap();
+ assert!(
+ !store
+ .compare_and_set_application_launch_configuration(
+ source.clone(),
+ json!({"foreign":true}),
+ desired.clone()
+ )
+ .await
+ .unwrap()
+ );
+ assert!(
+ store
+ .compare_and_set_application_launch_configuration(
+ source.clone(),
+ previous.clone(),
+ desired.clone()
+ )
+ .await
+ .unwrap()
+ );
+ // The same desired authority is an idempotent retry at this exact source.
+ assert!(
+ store
+ .compare_and_set_application_launch_configuration(
+ source.clone(),
+ previous.clone(),
+ desired.clone()
+ )
+ .await
+ .unwrap()
+ );
+ let newer = store
+ .commit_application(
+ deployment,
+ 1,
+ crate::applications::tests::input(),
+ vec![],
+ vec![],
+ )
+ .await
+ .unwrap();
+ assert!(
+ !store
+ .compare_and_set_application_launch_configuration(source, previous, desired.clone())
+ .await
+ .unwrap()
+ );
+ let latest = json!({"kind":"public_host","native":[],"endpoints":{"latest":true}});
+ assert!(
+ store
+ .compare_and_set_application_launch_configuration(newer, desired, latest.clone())
+ .await
+ .unwrap()
+ );
+ let record = store.application_launch().await.unwrap().unwrap();
+ assert_eq!(record.configuration, latest);
+ assert!(record.pending.is_none());
+ store.close().await.unwrap();
+}
+
+#[tokio::test]
+async fn committed_selection_requires_current_source_and_never_overwrites_selection() {
+ let (_directory, store) = crate::applications::tests::store().await;
+ let deployment = Uuid::new_v4();
+ let original = store
+ .commit_application(
+ deployment,
+ 0,
+ crate::applications::tests::input(),
+ vec![],
+ vec![],
+ )
+ .await
+ .unwrap();
+ let mut missing = original.clone();
+ missing.deployment = Uuid::new_v4();
+ assert!(
+ store
+ .select_committed_application_launch(missing, json!({}))
+ .await
+ .is_err()
+ );
+ let current = store
+ .commit_application(
+ deployment,
+ 1,
+ crate::applications::tests::input(),
+ vec![],
+ vec![],
+ )
+ .await
+ .unwrap();
+ assert!(
+ store
+ .select_committed_application_launch(original, json!({}))
+ .await
+ .is_err()
+ );
+ assert!(store.application_launch().await.unwrap().is_none());
+ let configuration = json!({"trusted":"committed"});
+ store
+ .select_committed_application_launch(current.clone(), configuration.clone())
+ .await
+ .unwrap();
+ let selected = store.application_launch().await.unwrap().unwrap();
+ assert_eq!(selected.deployment, deployment);
+ assert_eq!(selected.configuration, configuration);
+ assert!(selected.pending.is_none());
+ assert!(
+ store
+ .select_committed_application_launch(current.clone(), json!({"replacement":true}))
+ .await
+ .is_err()
+ );
+ assert_eq!(
+ store
+ .application_launch()
+ .await
+ .unwrap()
+ .unwrap()
+ .configuration,
+ configuration
+ );
+ assert_eq!(
+ store.application_head(deployment).await.unwrap(),
+ Some(current)
+ );
+ store.close().await.unwrap();
+}
diff --git a/platform/host-runtime/src/application.rs b/platform/host-runtime/src/application.rs
index 7be0d90e26..8fbd0139e6 100644
--- a/platform/host-runtime/src/application.rs
+++ b/platform/host-runtime/src/application.rs
@@ -194,7 +194,7 @@ impl NativeRegistry {
pub(super) struct Method {
operation: Option<maka_event_log::applications::operations::OperationTemplate>,
- input: jsonschema::Validator,
+ input: Arc<jsonschema::Validator>,
output: Arc<jsonschema::Validator>,
progress: Option<Arc<jsonschema::Validator>>,
}
@@ -225,16 +225,13 @@ impl Service {
.progress
.as_ref()
.map(|schema| {
- jsonschema::validator_for(schema)
- .map(Arc::new)
- .map_err(|e| e.to_string())
+ maka_plugin_core::validator(schema).map_err(|e| e.to_string())
})
.transpose()?,
- input: jsonschema::validator_for(&method.input)
+ input: maka_plugin_core::validator(&method.input)
+ .map_err(|e| e.to_string())?,
+ output: maka_plugin_core::validator(&method.output)
.map_err(|e| e.to_string())?,
- output: Arc::new(
- jsonschema::validator_for(&method.output).map_err(|e| e.to_string())?,
- ),
},
))
})
diff --git a/platform/host-runtime/src/application/host.rs b/platform/host-runtime/src/application/host.rs
index 36b074ad3e..9bda734bfd 100644
--- a/platform/host-runtime/src/application/host.rs
+++ b/platform/host-runtime/src/application/host.rs
@@ -25,9 +25,9 @@ mod lifecycle;
pub mod websocket;
pub use access::Pairing;
pub mod protocol;
-pub use client::{Client, LocalCall};
+pub use client::{CallChannelOpener, Client, LocalCall};
pub use discovery::Registration;
-pub use protocol::{LifecycleMode, Status, VERSION};
+pub use protocol::{Configuration, LifecycleMode, Status, VERSION};
use super::transport::{
Catalogue, Endpoints, HostEndpoint, NativeGrant, Peer, PeerPolicy, Peers, Server, wire,
@@ -73,13 +73,12 @@ pub struct Host {
application: Application,
store: Arc<EventLog>,
access: Arc<maka_config::ConfigurationStore>,
- access_grant: String,
+ access_grant: Mutex<String>,
access_gate: tokio::sync::Mutex<()>,
epoch: Uuid,
source: DeploymentRef,
policy: Arc<PeerPolicy>,
- endpoints: Endpoints,
- selections: BTreeMap<String, HostEndpoint>,
+ selections: Mutex<BTreeMap<String, HostEndpoint>>,
sessions: Mutex<BTreeMap<Uuid, Arc<Session>>>,
stop: CancellationToken,
residency: Mutex<lifecycle::Residency>,
@@ -146,13 +145,10 @@ impl Host {
scripts,
)
.await?;
- let endpoints = match Endpoints::start(&application.client(), &selections).await {
- Ok(endpoints) => endpoints,
- Err(error) => {
- application.shutdown().await?;
- return Err(error);
- }
- };
+ if let Err(error) = Endpoints::start(&application.client(), &selections).await {
+ application.shutdown().await?;
+ return Err(error);
+ }
let access = match maka_config::ConfigurationStore::for_root(root.clone()).await {
Ok(access) => Arc::new(access),
Err(error) => {
@@ -163,8 +159,14 @@ impl Host {
let access_grant = format!(
"application.owner:{:x}",
sha2::Sha256::digest(
- serde_json::to_vec(&(application.source(), &selections))
- .map_err(|error| error.to_string())?
+ serde_json::to_vec(&(
+ application.source(),
+ Configuration::PublicHost {
+ endpoints: selections.clone(),
+ native: policy.native_grants(),
+ }
+ ))
+ .map_err(|error| error.to_string())?
)
);
Ok(Arc::new(Self {
@@ -174,17 +176,66 @@ impl Host {
application,
store,
access,
- access_grant,
+ access_grant: Mutex::new(access_grant),
access_gate: tokio::sync::Mutex::new(()),
policy,
- endpoints,
epoch: Uuid::new_v4(),
- selections,
+ selections: Mutex::new(selections),
sessions: Mutex::default(),
stop: CancellationToken::new(),
residency: Mutex::new(lifecycle::Residency::default()),
}))
}
+ /// This configuration step is durable independently of source replacement.
+ /// The launcher's original pending operation coordinates their retries.
+ async fn configure(
+ &self,
+ source: DeploymentRef,
+ previous: Configuration,
+ configuration: Configuration,
+ ) -> Result<()> {
+ previous.validate()?;
+ configuration.validate()?;
+ if source.state_domain != self.root_id() || source.deployment != self.source.deployment {
+ return Err("foreign Host configuration source".into());
+ }
+ let _gate = self.access_gate.lock().await;
+ if self.stop.is_cancelled() {
+ return Err("Host is closing".into());
+ }
+ let current = Configuration::PublicHost {
+ endpoints: self.selections.lock().unwrap().clone(),
+ native: self.policy.native_grants(),
+ };
+ let current = serde_json::to_value(current).map_err(|error| error.to_string())?;
+ let previous = serde_json::to_value(previous).map_err(|error| error.to_string())?;
+ let desired = serde_json::to_value(&configuration).map_err(|error| error.to_string())?;
+ if current != previous && current != desired {
+ return Err("Host configuration conflicts with current authority".into());
+ }
+ let access_grant = format!(
+ "application.owner:{:x}",
+ sha2::Sha256::digest(
+ serde_json::to_vec(&(Some(&self.source), &configuration))
+ .map_err(|error| error.to_string())?
+ )
+ );
+ if !self
+ .store
+ .compare_and_set_application_launch_configuration(source, previous, desired)
+ .await
+ .map_err(|error| error.to_string())?
+ {
+ return Err("Host configuration source or persisted authority changed".into());
+ }
+ // No await between durable confirmation and publication of live authority.
+ // Existing sessions retain their captured bindings and accepted jobs.
+ let Configuration::PublicHost { endpoints, native } = configuration;
+ *self.selections.lock().unwrap() = endpoints;
+ self.policy.replace_native_grants(native);
+ *self.access_grant.lock().unwrap() = access_grant;
+ Ok(())
+ }
pub fn root_id(&self) -> &str {
self.root.root_id()
}
@@ -337,6 +388,15 @@ impl Host {
drop(gate);
wire::write(&mut output, &Response::Status(self.status())).await
}
+ Request::Configure {
+ source,
+ previous,
+ configuration,
+ } => {
+ authority.local()?;
+ let result = self.configure(source, previous, configuration).await;
+ wire::write(&mut output, &Response::Configured(result)).await
+ }
Request::PrepareAccess { principal } => {
authority.local()?;
wire::write(
@@ -371,24 +431,51 @@ impl Host {
let result = self.retire(interrupt, allow_idle_connections).await;
wire::write(&mut output, &Response::Retired(result)).await
}
+ Request::Quit { session: id } => {
+ authority.local()?;
+ let session = self.sessions.lock().unwrap().get(&id).cloned();
+ if let Some(session) = session {
+ session.peer.replace(vec![])?;
+ session.transport.close().await?;
+ }
+ let stop = {
+ // Admission and the last-session decision share the same gate.
+ let _gate = self.access_gate.lock().await;
+ let mut sessions = self.sessions.lock().unwrap();
+ sessions.remove(&id);
+ let stop = sessions.is_empty();
+ if stop {
+ self.residency.lock().unwrap().paused = true;
+ self.stop.cancel();
+ }
+ stop
+ };
+ if stop {
+ self.shutdown().await?;
+ }
+ wire::write(&mut output, &Response::Closed).await
+ }
request @ (Request::Connect | Request::ConnectPeer { .. }) => {
let gate = self.access_gate.lock().await;
let principal = self
.identify(&authority, opening.client_instance.as_deref())
.await?;
let (mode, peer, endpoints, selections) = match request {
- Request::Connect => (
- SessionMode::Owned,
- Peer::new(
- self.source.clone(),
- self.selections
- .values()
- .map(|endpoint| endpoint.consumer.clone())
- .collect(),
- )?,
- self.endpoints.clone(),
- self.selections.clone(),
- ),
+ Request::Connect => {
+ let selections = self.selections.lock().unwrap().clone();
+ (
+ SessionMode::Owned,
+ Peer::new(
+ self.source.clone(),
+ selections
+ .values()
+ .map(|endpoint| endpoint.consumer.clone())
+ .collect(),
+ )?,
+ Endpoints::capture(&self.application.client(), &selections).await?,
+ selections,
+ )
+ }
Request::ConnectPeer {
source,
consumers,
@@ -547,7 +634,10 @@ impl Host {
// drains all original calls before releasing the session.
session.peer.replace(vec![])?;
session.transport.close().await?;
- self.sessions.lock().unwrap().remove(&id);
+ {
+ let _gate = self.access_gate.lock().await;
+ self.sessions.lock().unwrap().remove(&id);
+ }
// EOF or a partial control frame still requests this same
// session's drain. Only the completed cleanup above authorizes
// its acknowledgement; writing that acknowledgement may fail.
@@ -596,10 +686,10 @@ impl Host {
if selections.is_empty() || selections.len() > 128 {
return Err("invalid peer endpoint selections".into());
}
+ let published = self.selections.lock().unwrap();
for (name, selection) in selections {
selection.validate()?;
- let published = self
- .selections
+ let published = published
.get(name)
.ok_or("Host endpoint is not published")?;
if selection.consumer != published.consumer
diff --git a/platform/host-runtime/src/application/host/access.rs b/platform/host-runtime/src/application/host/access.rs
index 6ac7ac5030..4025121f41 100644
--- a/platform/host-runtime/src/application/host/access.rs
+++ b/platform/host-runtime/src/application/host/access.rs
@@ -67,7 +67,7 @@ impl Host {
.map_err(|error| error.to_string())?
.filter(|value| {
value.principal_kind == ManagedPrincipalKind::RemoteOwner
- && value.grants.contains(&self.access_grant)
+ && value.grants.contains(&*self.access_grant.lock().unwrap())
})
.ok_or("credential does not grant this application source".into())
}
@@ -128,7 +128,7 @@ impl Host {
principal_id: principal,
principal_kind: ManagedPrincipalKind::RemoteOwner,
grants: vec![
- self.access_grant.clone(),
+ self.access_grant.lock().unwrap().clone(),
"access.credential.finalize".into(),
],
can_publish_client_capabilities: false,
diff --git a/platform/host-runtime/src/application/host/client.rs b/platform/host-runtime/src/application/host/client.rs
index 6139e70c9d..ac19ab6062 100644
--- a/platform/host-runtime/src/application/host/client.rs
+++ b/platform/host-runtime/src/application/host/client.rs
@@ -19,7 +19,7 @@
//! Public application caller: an anchored authenticated session, no legacy operations.
use super::{
- Status, VERSION,
+ Configuration, Status, VERSION,
protocol::{Connected, Opening, Request, Response, SessionControl},
};
use crate::{
@@ -142,6 +142,9 @@ impl Client {
pub fn epoch(&self) -> Uuid {
self.connected.epoch
}
+ pub fn source(&self) -> &DeploymentRef {
+ &self.connected.source
+ }
/// Open inert custody for one exact captured endpoint. A caller retains this
/// Call through errors; reconnection never retries an unconfirmed mutation.
pub async fn call(
@@ -180,29 +183,19 @@ impl Client {
/// Establish only the authenticated session channel. The native adapter retains
/// its original checked Source, Call and caller-side RecordedIntent.
pub async fn open_call_channel(&self) -> Result<BufReader<LocalStream>> {
+ self.call_channel_opener()?.open().await
+ }
+ /// Freeze only this already authenticated session's physical target. It owns
+ /// no anchor, caller approval, source, or operation token and cannot reconnect.
+ pub fn call_channel_opener(&self) -> Result<CallChannelOpener> {
if self.phase != Phase::Ready {
return Err("Host session control is not ready".into());
}
- let mut stream = ipc::connect(&self.address)
- .await
- .map_err(|error| error.to_string())?;
- wire::write(
- &mut stream,
- &Opening {
- client_instance: None,
- version: VERSION,
- epoch: Some(self.connected.epoch),
- request: Request::Call {
- session: self.connected.session,
- },
- },
- )
- .await?;
- let mut reader = wire::Reader::new(stream);
- if !matches!(reader.read::<Response>().await?, Response::Accepted) {
- return Err("Host did not accept application channel".into());
- }
- Ok(reader.inner)
+ Ok(CallChannelOpener {
+ address: self.address.clone(),
+ epoch: self.connected.epoch,
+ session: self.connected.session,
+ })
}
fn ready_peer(&self) -> Result<&DeploymentRef> {
if self.phase != Phase::Ready {
@@ -356,6 +349,56 @@ impl Client {
}
}
}
+ /// Close this local session and stop its Host when no other session remains.
+ /// A separate epoch-pinned channel also works after losing a control observer;
+ /// a partial frame on the original anchor must never receive another command.
+ pub async fn quit(&mut self) -> Result<()> {
+ if self.phase == Phase::Closed {
+ return Ok(());
+ }
+ self.phase = Phase::Lost;
+ match control(
+ &self.address,
+ Some(self.connected.epoch),
+ Request::Quit {
+ session: self.connected.session,
+ },
+ )
+ .await?
+ {
+ Response::Closed => {
+ self.phase = Phase::Closed;
+ Ok(())
+ }
+ _ => Err("Host quit cleanup remains unconfirmed".into()),
+ }
+ }
+ /// Persist trusted configuration and apply it to this same Host incarnation.
+ /// A lost response must be retried with the same source and desired value.
+ pub async fn configure(
+ &self,
+ source: DeploymentRef,
+ previous: Configuration,
+ configuration: Configuration,
+ ) -> Result<()> {
+ if self.phase != Phase::Ready || self.peer.is_some() {
+ return Err("Host configuration requires a ready local owner session".into());
+ }
+ match control(
+ &self.address,
+ Some(self.connected.epoch),
+ Request::Configure {
+ source,
+ previous,
+ configuration,
+ },
+ )
+ .await?
+ {
+ Response::Configured(result) => result,
+ _ => Err("Host configuration response mismatch".into()),
+ }
+ }
pub async fn prepare_access(
address: &Path,
epoch: Uuid,
@@ -410,6 +453,38 @@ impl Client {
}
}
}
+/// An immutable physical target captured from an authenticated Client. The
+/// embedding must fence native calls before detaching that Client's anchor.
+#[derive(Clone)]
+pub struct CallChannelOpener {
+ address: PathBuf,
+ epoch: Uuid,
+ session: Uuid,
+}
+impl CallChannelOpener {
+ pub async fn open(&self) -> Result<BufReader<LocalStream>> {
+ let mut stream = ipc::connect(&self.address)
+ .await
+ .map_err(|error| error.to_string())?;
+ wire::write(
+ &mut stream,
+ &Opening {
+ client_instance: None,
+ version: VERSION,
+ epoch: Some(self.epoch),
+ request: Request::Call {
+ session: self.session,
+ },
+ },
+ )
+ .await?;
+ let mut reader = wire::Reader::new(stream);
+ if !matches!(reader.read::<Response>().await?, Response::Accepted) {
+ return Err("Host did not accept application channel".into());
+ }
+ Ok(reader.inner)
+ }
+}
async fn control(address: &Path, epoch: Option<Uuid>, request: Request) -> Result<Response> {
let stream = ipc::connect(address)
.await
diff --git a/platform/host-runtime/src/application/host/protocol.rs b/platform/host-runtime/src/application/host/protocol.rs
index 9e957699a2..ab102ccc99 100644
--- a/platform/host-runtime/src/application/host/protocol.rs
+++ b/platform/host-runtime/src/application/host/protocol.rs
@@ -18,12 +18,45 @@
*/
use super::super::Result;
-use super::super::transport::{Catalogue, HostEndpoint};
+use super::super::transport::{Catalogue, HostEndpoint, NativeGrant};
use maka_plugin_core::{composition::InstanceId, deployment::DeploymentRef};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use uuid::Uuid;
+/// Persisted trusted-local Host publication and native invocation authority.
+#[derive(Clone, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum Configuration {
+ PublicHost {
+ endpoints: BTreeMap<String, HostEndpoint>,
+ native: Vec<NativeGrant>,
+ },
+}
+impl Configuration {
+ pub fn validate(&self) -> Result<()> {
+ let Self::PublicHost { endpoints, native } = self;
+ if endpoints.is_empty()
+ || endpoints.len() > 128
+ || endpoints
+ .keys()
+ .any(|name| name.is_empty() || name.len() > 256)
+ {
+ return Err("invalid public Host endpoint configuration".into());
+ }
+ for endpoint in endpoints.values() {
+ endpoint.validate()?;
+ }
+ if native.len() > 128 {
+ return Err("maximum 128 native service grants".into());
+ }
+ for grant in native {
+ grant.validate()?;
+ }
+ Ok(())
+ }
+}
+
pub const VERSION: u32 = 1;
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
@@ -48,6 +81,15 @@ pub enum Request {
session: Uuid,
},
Status,
+ Quit {
+ session: Uuid,
+ },
+ /// Current-account owner only. Source/packages remain a separate CAS operation.
+ Configure {
+ source: DeploymentRef,
+ previous: Configuration,
+ configuration: Configuration,
+ },
Retire {
#[serde(default)]
interrupt: bool,
@@ -79,6 +121,7 @@ pub enum Response {
result: Result<()>,
},
Status(Status),
+ Configured(Result<()>),
Retired(Result<bool>),
Closed,
Pairing(Result<super::Pairing>),
diff --git a/platform/host-runtime/src/application/transport/policy.rs b/platform/host-runtime/src/application/transport/policy.rs
index 3a881b19d4..c09a74cbeb 100644
--- a/platform/host-runtime/src/application/transport/policy.rs
+++ b/platform/host-runtime/src/application/transport/policy.rs
@@ -35,7 +35,7 @@ use serde::{Deserialize, Serialize};
use std::{
collections::{BTreeMap, BTreeSet},
sync::{
- Arc, Mutex,
+ Arc, Mutex, RwLock,
atomic::{AtomicBool, Ordering},
},
};
@@ -281,7 +281,7 @@ struct Evidence {
#[derive(Default)]
pub struct PeerPolicy {
issuer: Issuer,
- native: Vec<NativeGrant>,
+ native: RwLock<Vec<NativeGrant>>,
}
impl PeerPolicy {
pub fn with_native_grants(native: Vec<NativeGrant>) -> Result<Self> {
@@ -292,10 +292,18 @@ impl PeerPolicy {
grant.validate()?;
}
Ok(Self {
- native,
+ native: RwLock::new(native),
..Self::default()
})
}
+ /// New native admissions use the current trusted configuration. Already
+ /// accepted execution retains its original owner and is never cancelled here.
+ pub(crate) fn replace_native_grants(&self, native: Vec<NativeGrant>) {
+ *self.native.write().unwrap() = native;
+ }
+ pub(crate) fn native_grants(&self) -> Vec<NativeGrant> {
+ self.native.read().unwrap().clone()
+ }
pub fn issuer(&self) -> Issuer {
self.issuer.clone()
}
@@ -357,6 +365,8 @@ impl Policy for PeerPolicy {
if let Some(origin) = invocation.native_origin {
return self
.native
+ .read()
+ .unwrap()
.iter()
.any(|grant| {
&grant.origin == origin.instance()
@@ -400,6 +410,8 @@ impl Policy for PeerPolicy {
if let Some(origin) = invocation.native_origin {
return self
.native
+ .read()
+ .unwrap()
.iter()
.any(|grant| {
grant.recovery
diff --git a/platform/host-runtime/src/application/transport/server.rs b/platform/host-runtime/src/application/transport/server.rs
index 57038495da..88bb4ec4c9 100644
--- a/platform/host-runtime/src/application/transport/server.rs
+++ b/platform/host-runtime/src/application/transport/server.rs
@@ -21,7 +21,7 @@
use super::{
catalogue::Catalogue,
policy::{Peers, PolicyOwner},
- wire::{self, Accepted, Control, Hello, Open, Reader, Reply},
+ wire::{self, Control, Open, Reader, Reply},
};
use crate::application::Result;
use crate::application::{Client, QueryCompletion, RESOURCE_CHUNK_LIMIT, ResourceDescriptor};
@@ -96,26 +96,7 @@ impl Server {
R: AsyncRead + Unpin + Send + 'static,
W: AsyncWrite + Unpin + Send + 'static,
{
- self.serve_inner(reader, writer, peer, None, false)
- }
- /// An inherited-bootstrap nonce pairs a private current-account channel
- /// with this original Host incarnation. Hello uses the same tracked
- /// connection owner as Open; it never creates request authority.
- pub fn serve_private<R, W>(
- self: &Arc<Self>,
- reader: R,
- writer: W,
- peer: Arc<Peers>,
- nonce: uuid::Uuid,
- ) -> Result<impl Future<Output = Result<()>> + Send + 'static>
- where
- R: AsyncRead + Unpin + Send + 'static,
- W: AsyncWrite + Unpin + Send + 'static,
- {
- if nonce.is_nil() {
- return Err("private channel nonce must be nonnil".into());
- }
- self.serve_inner(reader, writer, peer, Some(nonce), false)
+ self.serve_inner(reader, writer, peer, false)
}
/// The embedding has authenticated a public principal and selected its
/// Host gateway consumers and endpoint grants. This mode records caller
@@ -130,14 +111,13 @@ impl Server {
R: AsyncRead + Unpin + Send + 'static,
W: AsyncWrite + Unpin + Send + 'static,
{
- self.serve_inner(reader, writer, peer, None, true)
+ self.serve_inner(reader, writer, peer, true)
}
fn serve_inner<R, W>(
self: &Arc<Self>,
reader: R,
writer: W,
peer: Arc<Peers>,
- nonce: Option<uuid::Uuid>,
owned: bool,
) -> Result<impl Future<Output = Result<()>> + Send + 'static>
where
@@ -160,7 +140,7 @@ impl Server {
lost: false,
pending: None,
};
- let execution = AssertUnwindSafe(session.run(&server, peer, nonce, owned))
+ let execution = AssertUnwindSafe(session.run(&server, peer, owned))
.catch_unwind()
.await;
let panicked = execution.is_err();
@@ -354,29 +334,7 @@ impl<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> Session<R, W> {
})
.await
}
- async fn run(
- &mut self,
- server: &Server,
- peer: Arc<Peers>,
- nonce: Option<uuid::Uuid>,
- owned: bool,
- ) -> Result<()> {
- if let Some(nonce) = nonce {
- let hello: Hello = tokio::select! {
- biased;
- () = self.stop.cancelled() => return Err("Application transport is closing".into()),
- result = self.reader.read_bounded(512) => result?,
- };
- if hello.nonce != nonce {
- return Err("private channel belongs to a different Host incarnation".into());
- }
- let accepted = Accepted { nonce };
- tokio::select! {
- biased;
- () = self.stop.cancelled() => return Err("Application transport is closing".into()),
- result = wire::write(&mut self.writer, &accepted) => result?,
- }
- }
+ async fn run(&mut self, server: &Server, peer: Arc<Peers>, owned: bool) -> Result<()> {
let open: Open = self.read().await?;
let descriptors: BTreeMap<String, ResourceDescriptor> = self.read().await?;
let Some(endpoint) = server
diff --git a/platform/host-runtime/src/application/transport/wire.rs b/platform/host-runtime/src/application/transport/wire.rs
index a1ff8a153e..6e80a25d82 100644
--- a/platform/host-runtime/src/application/transport/wire.rs
+++ b/platform/host-runtime/src/application/transport/wire.rs
@@ -31,17 +31,6 @@ use std::io::Write;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader};
use uuid::Uuid;
-#[derive(Serialize, Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct Hello {
- pub nonce: Uuid,
-}
-#[derive(Serialize, Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct Accepted {
- pub nonce: Uuid,
-}
-
pub const FRAME_LIMIT: usize = 2 * 1024 * 1024;
/// Observations supplied by the caller adapter, never transport authentication.
#[derive(Clone, Debug, Serialize, Deserialize)]
diff --git a/platform/render-terminal/src/application/run.rs b/platform/render-terminal/src/application/run.rs
index fcf3ad214c..0e2e6de763 100644
--- a/platform/render-terminal/src/application/run.rs
+++ b/platform/render-terminal/src/application/run.rs
@@ -22,6 +22,7 @@ use super::{PaintedScene, Presenter, registry, sanitize};
use crate::terminal;
use crossterm::event::{self, Event, EventStream, KeyCode, KeyEventKind, KeyModifiers};
use futures_util::StreamExt;
+use maka_transport::ExitMode;
use serde_json::json;
use std::time::{Duration, Instant};
use std::{
@@ -122,7 +123,7 @@ pub fn run(
execution: maka_ui_runtime::ExecutionOptions,
startup: Option<crate::startup::Startup>,
mut startup_errors: Vec<String>,
-) -> Result<(), String> {
+) -> Result<ExitMode, String> {
let mut presenter = scheduler
.block_on(Presenter::start_loaded(
store,
@@ -175,6 +176,7 @@ pub fn run(
let mut clipboard_output = VecDeque::new();
let mut issued_sequence = 0u64;
let mut written_sequence = 0u64;
+ let mut exit_mode = ExitMode::Quit;
let mut exit_key = ExitKey::default();
let mut interrupt = InterruptWindow::default();
let mut queued_input: Option<QueuedInput> = None;
@@ -302,7 +304,8 @@ pub fn run(
continue;
}
let result = (|| -> Result<bool, String> {
- if exit_key.ready(None, Instant::now()) {
+ if let Some(mode) = exit_key.ready(None, Instant::now()) {
+ exit_mode = mode;
return Ok(true);
}
let screen = screen.as_mut().expect("active terminal");
@@ -365,7 +368,7 @@ pub fn run(
} else {
let at = [
presenter.next_frame.filter(|_| pending_output.is_none()),
- exit_key.0.map(|(_, at)| at),
+ exit_key.0.map(|(_, _, at)| at),
updates.next_deadline(),
]
.into_iter()
@@ -406,7 +409,8 @@ pub fn run(
}
}
}
- if exit_key.ready(Some(&input), Instant::now()) {
+ if let Some(mode) = exit_key.ready(Some(&input), Instant::now()) {
+ exit_mode = mode;
return Ok(true);
}
interrupt.observe(&input);
@@ -473,7 +477,14 @@ pub fn run(
if matches!(key.code, KeyCode::Char('c' | 'C')) && !interrupt.request(now) {
return Ok(false);
}
- return Ok(exit_key.request(key.code, guard.reports_key_releases(), now));
+ let mode = requested_exit_mode(&Event::Key(key)).expect("exit gesture");
+ if let Some(mode) =
+ exit_key.request(key.code, mode, guard.reports_key_releases(), now)
+ {
+ exit_mode = mode;
+ return Ok(true);
+ }
+ return Ok(false);
}
// Captured terminal input belongs to its child process, never
// to a previously armed application exit gesture.
@@ -572,7 +583,7 @@ pub fn run(
eprintln!("{diagnostic}");
}
}
- return failure.map_or(Ok(()), Err);
+ return failure.map_or(Ok(exit_mode), Err);
}
Err(maka_ui_runtime::Error::Core(maka_plugin_core::Error::CleanupPending)) => {}
Err(error) => presenter.report(&error),
@@ -610,8 +621,24 @@ pub(super) fn is_interrupt(event: &Event) -> bool {
matches!(event,Event::Key(key) if key.kind==KeyEventKind::Press && matches!(key.code,KeyCode::Char('c' | 'C')) && key.modifiers==KeyModifiers::CONTROL)
}
fn is_exit(event: &Event) -> bool {
- is_interrupt(event)
- || matches!(event,Event::Key(key) if key.kind==KeyEventKind::Press && key.code==KeyCode::Char('q') && key.modifiers==KeyModifiers::CONTROL)
+ requested_exit_mode(event).is_some()
+}
+fn requested_exit_mode(event: &Event) -> Option<ExitMode> {
+ let Event::Key(key) = event else { return None };
+ if key.kind != KeyEventKind::Press {
+ return None;
+ }
+ if matches!(key.code, KeyCode::Char('q' | 'Q'))
+ && key.modifiers == KeyModifiers::CONTROL | KeyModifiers::ALT
+ {
+ Some(ExitMode::KeepHost)
+ } else if is_interrupt(event)
+ || (key.code == KeyCode::Char('q') && key.modifiers == KeyModifiers::CONTROL)
+ {
+ Some(ExitMode::Quit)
+ } else {
+ None
+ }
}
#[derive(Default)]
@@ -646,31 +673,35 @@ impl InterruptWindow {
/// Keep a negotiated physical quit gesture inside its original raw terminal.
/// A lost release must not strand shutdown; another physical action cancels it.
#[derive(Default)]
-struct ExitKey(Option<(KeyCode, Instant)>);
+struct ExitKey(Option<(KeyCode, ExitMode, Instant)>);
impl ExitKey {
- fn request(&mut self, key: KeyCode, releases: bool, now: Instant) -> bool {
+ fn request(
+ &mut self,
+ key: KeyCode,
+ mode: ExitMode,
+ releases: bool,
+ now: Instant,
+ ) -> Option<ExitMode> {
if !releases {
- return true;
+ return Some(mode);
}
- self.0 = Some((key, now + Duration::from_secs(2)));
- false
+ self.0 = Some((key, mode, now + Duration::from_secs(2)));
+ None
}
- fn ready(&mut self, event: Option<&Event>, now: Instant) -> bool {
- let Some((code, deadline)) = self.0 else {
- return false;
- };
+ fn ready(&mut self, event: Option<&Event>, now: Instant) -> Option<ExitMode> {
+ let (code, mode, deadline) = self.0?;
if now >= deadline
|| matches!(event, Some(Event::Key(key)) if key.kind == KeyEventKind::Release && key.code == code)
{
self.0 = None;
- return true;
+ return Some(mode);
}
if matches!(event, Some(Event::Key(key)) if key.kind == KeyEventKind::Press)
|| matches!(event, Some(Event::Mouse(mouse)) if matches!(mouse.kind, event::MouseEventKind::Down(_)))
{
self.0 = None;
}
- false
+ None
}
}
@@ -714,8 +745,14 @@ mod tests {
fn quit_release_keeps_original_key_and_cancels_on_a_new_gesture() {
let now = Instant::now();
let mut exit = ExitKey::default();
- assert!(exit.request(KeyCode::Char('q'), false, now));
- assert!(!exit.request(KeyCode::Char('q'), true, now));
+ assert_eq!(
+ exit.request(KeyCode::Char('q'), ExitMode::Quit, false, now),
+ Some(ExitMode::Quit)
+ );
+ assert_eq!(
+ exit.request(KeyCode::Char('q'), ExitMode::Quit, true, now),
+ None
+ );
let released = |code| {
Event::Key(KeyEvent::new_with_kind(
code,
@@ -723,23 +760,90 @@ mod tests {
KeyEventKind::Release,
))
};
- assert!(!exit.ready(Some(&released(KeyCode::Char('c'))), now));
- assert!(exit.ready(Some(&released(KeyCode::Char('q'))), now));
- assert!(!exit.ready(Some(&released(KeyCode::Char('q'))), now));
- assert!(!exit.request(KeyCode::Char('c'), true, now));
+ assert_eq!(exit.ready(Some(&released(KeyCode::Char('c'))), now), None);
+ assert_eq!(
+ exit.ready(Some(&released(KeyCode::Char('q'))), now),
+ Some(ExitMode::Quit)
+ );
+ assert_eq!(exit.ready(Some(&released(KeyCode::Char('q'))), now), None);
+ assert_eq!(
+ exit.request(KeyCode::Char('c'), ExitMode::Quit, true, now),
+ None
+ );
let repeat = Event::Key(KeyEvent::new_with_kind(
KeyCode::Char('c'),
KeyModifiers::CONTROL,
KeyEventKind::Repeat,
));
- assert!(!exit.ready(Some(&repeat), now));
+ assert_eq!(exit.ready(Some(&repeat), now), None);
let other = Event::Key(KeyEvent::new(KeyCode::Char('x'), KeyModifiers::NONE));
- assert!(!exit.ready(Some(&other), now));
- assert!(!exit.ready(Some(&released(KeyCode::Char('c'))), now));
- assert!(!exit.ready(None, now + Duration::from_secs(3)));
- assert!(!exit.request(KeyCode::Char('q'), true, now));
- assert!(!exit.ready(None, now + Duration::from_secs(1)));
- assert!(exit.ready(None, now + Duration::from_secs(2)));
+ assert_eq!(exit.ready(Some(&other), now), None);
+ assert_eq!(exit.ready(Some(&released(KeyCode::Char('c'))), now), None);
+ assert_eq!(exit.ready(None, now + Duration::from_secs(3)), None);
+ assert_eq!(
+ exit.request(KeyCode::Char('q'), ExitMode::Quit, true, now),
+ None
+ );
+ assert_eq!(exit.ready(None, now + Duration::from_secs(1)), None);
+ assert_eq!(
+ exit.ready(None, now + Duration::from_secs(2)),
+ Some(ExitMode::Quit)
+ );
+ }
+ #[test]
+ fn explicit_ui_only_exit_is_a_distinct_physical_gesture() {
+ let gesture = Event::Key(KeyEvent::new(
+ KeyCode::Char('q'),
+ KeyModifiers::CONTROL | KeyModifiers::ALT,
+ ));
+ assert_eq!(requested_exit_mode(&gesture), Some(ExitMode::KeepHost));
+ assert!(is_exit(&gesture));
+ assert!(!is_interrupt(&gesture));
+ assert_eq!(copy_gesture(&gesture, false), None);
+ for modifiers in [KeyModifiers::CONTROL, KeyModifiers::NONE, KeyModifiers::ALT] {
+ let input = Event::Key(KeyEvent::new(KeyCode::Char('q'), modifiers));
+ let expected = (modifiers == KeyModifiers::CONTROL).then_some(ExitMode::Quit);
+ assert_eq!(requested_exit_mode(&input), expected);
+ }
+ for kind in [KeyEventKind::Repeat, KeyEventKind::Release] {
+ let input = Event::Key(KeyEvent::new_with_kind(
+ KeyCode::Char('q'),
+ KeyModifiers::CONTROL | KeyModifiers::ALT,
+ kind,
+ ));
+ assert_eq!(requested_exit_mode(&input), None);
+ }
+ }
+ #[test]
+ fn cancelled_ui_only_exit_does_not_change_the_next_default_quit() {
+ let now = Instant::now();
+ let mut exit = ExitKey::default();
+ assert_eq!(
+ exit.request(KeyCode::Char('q'), ExitMode::KeepHost, true, now),
+ None
+ );
+ let other = Event::Key(KeyEvent::new(KeyCode::Char('x'), KeyModifiers::NONE));
+ assert_eq!(exit.ready(Some(&other), now), None);
+ assert_eq!(exit.ready(None, now + Duration::from_secs(3)), None);
+ assert_eq!(
+ exit.request(KeyCode::Char('q'), ExitMode::Quit, true, now),
+ None
+ );
+ let release = Event::Key(KeyEvent::new_with_kind(
+ KeyCode::Char('q'),
+ KeyModifiers::CONTROL,
+ KeyEventKind::Release,
+ ));
+ assert_eq!(exit.ready(Some(&release), now), Some(ExitMode::Quit));
+ assert_eq!(
+ exit.request(KeyCode::Char('q'), ExitMode::KeepHost, true, now),
+ None
+ );
+ assert_eq!(exit.ready(Some(&release), now), Some(ExitMode::KeepHost));
+ assert_eq!(
+ exit.request(KeyCode::Char('q'), ExitMode::KeepHost, false, now),
+ Some(ExitMode::KeepHost)
+ );
}
#[test]
fn source_copy_does_not_overlap_exit_or_release() {
diff --git a/platform/transport/Cargo.toml b/platform/transport/Cargo.toml
index c4d8c0a3de..4d0391c7dc 100644
--- a/platform/transport/Cargo.toml
+++ b/platform/transport/Cargo.toml
@@ -35,9 +35,9 @@ tokio = { workspace = true, features = ["net", "process"] }
tokio-util = { workspace = true, features = ["io-util"] }
uuid.workspace = true
futures-util.workspace = true
-tempfile.workspace = true
[dev-dependencies]
+tempfile.workspace = true
maka-ui-js.workspace = true
maka-host-js.workspace = true
diff --git a/platform/transport/README.md b/platform/transport/README.md
index f89e9efbaf..9209c550fc 100644
--- a/platform/transport/README.md
+++ b/platform/transport/README.md
@@ -21,8 +21,8 @@
[简体中文](README.zh-CN.md)
-Connects a foreground UI to independently owned Host deployments over local IPC. The launcher supplies trusted executable paths, connection configuration and method/recovery grants. `connections` owns those connections; `Updates` coordinates approved source changes.
+Connects a foreground UI to independently owned Host deployments over local IPC. The launcher supplies trusted connection configuration and method/recovery grants. Public connections discover a Root-bound Host or ask the trusted CLI launcher to start it; cohosts borrow the UI Store. `connections` owns those connections; `Updates` coordinates approved source changes.
-Remote registration data describes available services. It cannot create permission, choose a different caller or retarget an accepted operation. Shutdown waits for the original UI work, Host cleanup and child-process exit.
+Remote registration data describes available services. It cannot create permission, choose a different caller or retarget an accepted operation. Closing the UI waits for its original physical calls and locally owned cohost cleanup. Normal quit asks each public Host to stop if no other session remains, atomically with session admission. Explicit UI-only exit detaches and preserves background work.
[Source](src/lib.rs) · [Cargo manifest](Cargo.toml)
diff --git a/platform/transport/README.zh-CN.md b/platform/transport/README.zh-CN.md
index 0d6b432bfc..54b59aac1a 100644
--- a/platform/transport/README.zh-CN.md
+++ b/platform/transport/README.zh-CN.md
@@ -21,8 +21,8 @@
[English](README.md)
-通过本地 IPC 将前台 UI 连接到独立持有的 Host 部署。启动器提供可信可执行文件路径、连接配置以及方法和回执恢复权限。`connections` 管理连接，`Updates` 协调获准的源配置更新。
+通过本地 IPC 将前台 UI 连接到独立持有的 Host 部署。启动器提供可信连接配置以及方法和回执恢复权限。公共连接发现绑定 Root 的 Host；不存在时，通过可信 CLI 启动器自动启动。同进程 Host 借用 UI Store。`connections` 管理连接，`Updates` 协调获准的源配置更新。
-远端注册信息只描述可用服务，不能授予权限、替换调用方或改写已接受操作的目标。关闭流程需要等待原始 UI 工作、Host 清理和子进程退出。
+远端注册信息只描述可用服务，不能授予权限、替换调用方或改写已接受操作的目标。关闭 UI 时，先等待原始物理调用及本地持有的同进程 Host 清理。默认退出要求公共 Host 在没有其他会话时停止；此判断与会话接入原子执行。显式选择仅退出 UI 时，断开会话并保留后台工作。
[源码](src/lib.rs) · [Cargo 配置](Cargo.toml)
diff --git a/platform/transport/src/connections.rs b/platform/transport/src/connections.rs
index 9ee6da1042..6b8e19226e 100644
--- a/platform/transport/src/connections.rs
+++ b/platform/transport/src/connections.rs
@@ -19,16 +19,14 @@
use crate::{
Result,
- transport::{
- ipc, native,
- process::{Bootstrap, Control},
- },
+ transport::{ipc, native},
};
use futures_util::{
FutureExt,
future::{BoxFuture, join_all},
};
use maka_event_log::{EventLog, root::RootNamespaces};
+use maka_host_runtime::application::host::{Client, discovery};
use maka_plugin_core::{
composition::{ExecutionLocation, InstanceId, PresentationTarget},
deployment::{DeploymentRef, RemoteObservation},
@@ -38,17 +36,14 @@ use maka_ui_runtime::actions::RemoteService;
use serde::Deserialize;
use std::{
collections::{BTreeMap, BTreeSet},
- ffi::OsString,
io::Read,
path::{Path, PathBuf},
- process::Stdio,
sync::Arc,
};
-use tokio::process::Command;
use uuid::Uuid;
mod host;
-use host::{Address, CohostConnector, HostOwner, OwnedHost};
+use host::{CohostConnector, ConnectedHost, HostLink};
/// Trusted launcher input, never a guest-created credential or deployment edit.
#[derive(Default, Deserialize)]
@@ -67,14 +62,14 @@ pub struct Host {
pub native: Vec<crate::transport::NativeGrant>,
}
pub use maka_host_runtime::application::transport::HostEndpoint;
-/// Explicit executor location chosen by the trusted launcher. Cohost borrows
+/// Explicit Host attachment chosen by the trusted launcher. Cohost borrows
/// the already owned UI Store; it never opens a filesystem Root.
#[derive(Clone, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum HostLocation {
- Process {
+ Public {
root: PathBuf,
- /// Optional launcher pin, checked by the child after acquiring the Root.
+ /// Optional launcher pin, checked against discovery and live status.
#[serde(default, rename = "stateDomain")]
state_domain: Option<String>,
},
@@ -119,7 +114,7 @@ impl Configuration {
{
return Err("invalid or duplicate local Host declaration".into());
}
- if let HostLocation::Process {
+ if let HostLocation::Public {
state_domain: Some(domain),
..
} = &host.location
@@ -144,20 +139,6 @@ impl Configuration {
}
}
-/// Trusted launcher-selected executable and its compiled Host role. No plugin
-/// configuration can choose this program or transport native factory closures.
-pub struct HostExecutable {
- path: PathBuf,
- arguments: Vec<OsString>,
-}
-impl HostExecutable {
- pub fn new(path: PathBuf, arguments: Vec<OsString>) -> Result<Self> {
- Ok(Self {
- path: std::path::absolute(path).map_err(|e| e.to_string())?,
- arguments,
- })
- }
-}
struct PreparedHost {
configuration: Host,
consumers: BTreeSet<InstanceId>,
@@ -181,7 +162,7 @@ enum Phase {
Closing,
Closed,
}
-/// Actual child and original physical-call custody belong to the launcher.
+/// UI physical-call custody belongs to the launcher; public Hosts remain independent.
/// Prepare first, retain this object, then observe start through a mutable borrow.
/// Close the UI and its actual accepted work before draining these connections.
pub struct Connections {
@@ -193,9 +174,9 @@ pub struct Connections {
cohost_native: maka_host_runtime::application::NativeRegistry,
cohost_scripts: Option<Arc<dyn maka_host_runtime::application::backend::Factory>>,
namespaces: RootNamespaces,
- executable: HostExecutable,
+ helper: PathBuf,
prepared: Vec<PreparedHost>,
- hosts: Vec<OwnedHost>,
+ hosts: Vec<ConnectedHost>,
registrations: Vec<RemoteService>,
startup_errors: Vec<String>,
phase: Phase,
@@ -210,7 +191,7 @@ impl Connections {
target: PresentationTarget,
configuration: Configuration,
namespaces: RootNamespaces,
- executable: HostExecutable,
+ helper: PathBuf,
mut cohost_native: maka_host_runtime::application::NativeRegistry,
cohost_scripts: Option<Arc<dyn maka_host_runtime::application::backend::Factory>>,
) -> Result<Self> {
@@ -243,7 +224,7 @@ impl Connections {
for mut host in configuration.hosts {
let consumers = consumers(plan, &host.name);
match &mut host.location {
- HostLocation::Process { root, .. } => {
+ HostLocation::Public { root, .. } => {
*root = std::path::absolute(&*root).map_err(|e| e.to_string())?
}
HostLocation::Cohost {} => {
@@ -280,7 +261,7 @@ impl Connections {
observations: loaded.source().remote_exports.clone(),
initial: Some(loaded),
namespaces,
- executable,
+ helper: std::path::absolute(helper).map_err(|error| error.to_string())?,
prepared,
hosts: Vec::new(),
registrations: Vec::new(),
@@ -288,8 +269,8 @@ impl Connections {
phase: Phase::Prepared,
})
}
- /// The launcher owns self before this future can spawn a child. Dropping the
- /// observer releases only its borrow; it never drops or restarts that child.
+ /// The launcher retains actual local custody before awaiting startup. Dropping
+ /// the observer never restarts a public Host or closes an accepted UI call.
pub async fn start(&mut self) -> Result<()> {
if self.phase != Phase::Prepared {
return Err("Host connections cannot start in this phase".into());
@@ -360,77 +341,141 @@ impl Connections {
source: DeploymentRef,
) -> Result<crate::transport::Catalogue> {
match host.location.clone() {
- HostLocation::Process { root, .. } => {
- self.start_process(host, consumers, root, source).await
+ HostLocation::Public { root, .. } => {
+ self.connect_public(host, consumers, root, source).await
}
HostLocation::Cohost {} => self.start_cohost(host, consumers, source).await,
}
}
- async fn start_process(
+ async fn connect_public(
&mut self,
- host: Host,
+ configuration: Host,
consumers: BTreeSet<InstanceId>,
root: PathBuf,
source: DeploymentRef,
) -> Result<crate::transport::Catalogue> {
- let configuration = host.clone();
- let nonce = Uuid::new_v4();
- let address = Address::create(nonce)?;
- let state_domain = match &host.location {
- HostLocation::Process { state_domain, .. } => state_domain.clone(),
- HostLocation::Cohost {} => unreachable!("process location required"),
+ tokio::time::timeout(
+ std::time::Duration::from_secs(40),
+ self.connect_public_inner(configuration, consumers, root, source),
+ )
+ .await
+ .map_err(|_| "local Host connection/startup deadline elapsed".to_owned())?
+ }
+ async fn connect_public_inner(
+ &mut self,
+ configuration: Host,
+ consumers: BTreeSet<InstanceId>,
+ root: PathBuf,
+ source: DeploymentRef,
+ ) -> Result<crate::transport::Catalogue> {
+ let mut startup = None;
+ let mut discovered = match self.discover(&root).await {
+ Ok(discovered) => discovered,
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+ startup = Some(self.start_public_helper(&configuration, &root).await?);
+ self.discover(&root)
+ .await
+ .map_err(|error| error.to_string())?
+ }
+ Err(error) => return Err(error.to_string()),
};
- let bootstrap = Bootstrap {
- root,
- state_domain,
- deployment: host.deployment,
- namespaces: self.namespaces.clone(),
- source,
- consumers,
- endpoints: host.endpoints.clone(),
- native: host.native.clone(),
- address: address.path.clone(),
- nonce,
+ let HostLocation::Public { state_domain, .. } = &configuration.location else {
+ unreachable!()
};
- bootstrap.validate()?;
+ if discovered.source.deployment != configuration.deployment
+ || state_domain
+ .as_ref()
+ .is_some_and(|expected| *expected != discovered.root_id)
+ {
+ return Err(
+ "discovered Host differs from its configured deployment or State Root".into(),
+ );
+ }
+ let stream = match maka_host_runtime::ipc::connect(&discovered.endpoint).await {
+ Ok(stream) => stream,
+ Err(error)
+ if startup.is_none()
+ && error.downcast_ref::<std::io::Error>().is_some_and(|error| {
+ error.kind() == std::io::ErrorKind::ConnectionRefused
+ || error.kind() == std::io::ErrorKind::NotFound
+ }) =>
+ {
+ startup = Some(self.start_public_helper(&configuration, &root).await?);
+ discovered = self
+ .discover(&root)
+ .await
+ .map_err(|error| error.to_string())?;
+ maka_host_runtime::ipc::connect(&discovered.endpoint)
+ .await
+ .map_err(|error| error.to_string())?
+ }
+ Err(error) => return Err(error.to_string()),
+ };
+ let status = Client::status_stream(stream, Some(discovered.host_epoch)).await?;
+ discovered
+ .verify(&status)
+ .map_err(|error| error.to_string())?;
+ if discovered.source.deployment != configuration.deployment
+ || state_domain
+ .as_ref()
+ .is_some_and(|expected| *expected != discovered.root_id)
+ || startup.as_ref().is_some_and(|(_, ready)| {
+ discovered.host_epoch != ready.epoch || discovered.root_id != ready.root_id
+ })
+ {
+ return Err("local Host identity changed during startup".into());
+ }
+ let client = Client::connect_peer(
+ &discovered.endpoint,
+ Some(discovered.host_epoch),
+ source.clone(),
+ consumers.clone(),
+ configuration.endpoints.clone(),
+ )
+ .await?;
+ let catalogue = client.catalogue().clone();
let connection =
- native::Connection::new(Arc::new(ipc::Connector::new(address.path.clone(), nonce)?));
- let mut child = Command::new(&self.executable.path)
- .args(&self.executable.arguments)
- .stdin(Stdio::piped())
- .stdout(Stdio::piped())
- .stderr(Stdio::inherit())
- .spawn()
- .map_err(|e| e.to_string())?;
- let control = Control::new(
- child.stdout.take().expect("piped child stdout"),
- child.stdin.take().expect("piped child stdin"),
- nonce,
- )?;
- // Publish actual custody before polling bootstrap IO.
- self.hosts.push(OwnedHost {
- owner: HostOwner::Process {
- child,
- control: Box::new(control),
- _address: address,
+ native::Connection::new(Arc::new(ipc::Connector::new(client.call_channel_opener()?)));
+ self.hosts.push(ConnectedHost {
+ owner: HostLink::Public {
+ client: Some(Box::new(client)),
},
connection: Some(connection),
- ready: false,
+ ready: true,
published: false,
epochs: BTreeMap::new(),
configuration,
- peers: crate::transport::Peers::new(crate::transport::Peer::new(
- bootstrap.source.clone(),
- bootstrap.consumers.clone(),
- )?),
+ peers: crate::transport::Peers::new(crate::transport::Peer::new(source, consumers)?),
});
- let owned = self.hosts.last_mut().unwrap();
- let HostOwner::Process { control, .. } = &mut owned.owner else {
+ drop(startup);
+ Ok(catalogue)
+ }
+ async fn discover(&self, root: &Path) -> std::io::Result<discovery::Discovery> {
+ let namespaces = self.namespaces.clone();
+ let root = root.to_owned();
+ tokio::task::spawn_blocking(move || discovery::read_with_namespaces(&root, &namespaces))
+ .await
+ .map_err(std::io::Error::other)?
+ }
+ async fn start_public_helper(
+ &self,
+ configuration: &Host,
+ root: &Path,
+ ) -> Result<(crate::startup::Anchor, crate::startup::Ready)> {
+ let HostLocation::Public { state_domain, .. } = &configuration.location else {
unreachable!()
};
- let catalogue = control.start(&bootstrap).await?;
- owned.ready = true;
- Ok(catalogue)
+ crate::startup::launch(
+ &self.helper,
+ crate::startup::Request {
+ root: root.to_owned(),
+ state_domain: state_domain.clone(),
+ deployment: configuration.deployment,
+ endpoints: configuration.endpoints.clone(),
+ native: configuration.native.clone(),
+ },
+ )
+ .await
}
async fn start_cohost(
&mut self,
@@ -439,8 +484,8 @@ impl Connections {
source: DeploymentRef,
) -> Result<crate::transport::Catalogue> {
let peers = crate::transport::Peers::new(crate::transport::Peer::new(source, consumers)?);
- self.hosts.push(OwnedHost {
- owner: HostOwner::Cohost {
+ self.hosts.push(ConnectedHost {
+ owner: HostLink::Cohost {
application: None,
server: None,
captured: None,
@@ -453,7 +498,7 @@ impl Connections {
peers: peers.clone(),
});
let owned = self.hosts.last_mut().unwrap();
- let HostOwner::Cohost {
+ let HostLink::Cohost {
application,
server,
..
@@ -609,14 +654,14 @@ impl Connections {
{
let prepared = self.prepared.remove(index);
let callers = consumers(&plan, &prepared.configuration.name);
- // Newly needed executors belong to this controller before bootstrap
- // is observed. A failure retains actual custody, not a retryable recipe.
+ // Retain newly selected local custody before observing startup.
+ // Public connections never restart or replace the discovered Host.
let owners = self.hosts.len();
let result = self
.start_selected(prepared.configuration.clone(), callers, source.clone())
.await;
- // All pre-custody work is synchronous. Cancellation at the first
- // suspension already belongs to an actual owner, never this recipe.
+ // A failed connection without published custody may remain prepared.
+ // This never replays an operation or replaces an existing connection.
if result.is_err() && self.hosts.len() == owners {
self.prepared.insert(index, prepared);
}
@@ -706,7 +751,7 @@ impl Connections {
return Err("UI source selects an endpoint outside its configured grants".into());
}
// Complete the whole batch's known validation before changing any grant.
- // Physical IO failure still belongs to each original child, not rollback.
+ // Physical IO failure still belongs to its original connection, not rollback.
for (owned, approval) in self.hosts.iter_mut().zip(staged) {
let Some((approved, consumers, catalogue)) = approval else {
owned.peers.update(None, &BTreeSet::new())?;
@@ -762,10 +807,12 @@ impl Connections {
}
}
- /// Revoke every connection synchronously, then drain all original children.
- /// Both same-child Closed and actual exit are necessary. Failed original
- /// Calls and caller tokens remain in their physical owner even after exit.
+ /// Fence UI calls synchronously, then confirm their original physical cleanup.
+ /// Public session detachment never claims Host cleanup or waits for Host exit.
pub fn close(&mut self) -> BoxFuture<'_, Result<()>> {
+ self.close_with(crate::ExitMode::KeepHost)
+ }
+ pub fn close_with(&mut self, mode: crate::ExitMode) -> BoxFuture<'_, Result<()>> {
self.phase = Phase::Closing;
let physical: Vec<_> = self
.hosts
@@ -777,21 +824,16 @@ impl Connections {
})
.collect();
async move {
+ let namespaces = &self.namespaces;
let closes: Vec<_> = self
.hosts
.iter_mut()
.zip(physical)
.map(|(host, physical)| async move {
- let (physical, process) = tokio::join!(
- async {
- match physical {
- Some(physical) => physical.await,
- None => Ok(()),
- }
- },
- host.close_owner()
- );
- physical.and(process)
+ if let Some(physical) = physical {
+ physical.await?;
+ }
+ host.close_owner(mode, namespaces).await
})
.collect();
let results = join_all(closes).await;
@@ -811,12 +853,15 @@ impl Connections {
}
.boxed()
}
- /// Unconfirmed cleanup keeps the same owners and runtime alive. Never kill a
- /// child, release a failed Call, or infer cleanup from historical receipts.
+ /// Unconfirmed UI cleanup retains its original owners and runtime. Historical
+ /// receipts or a detached public session cannot settle a failed physical Call.
pub async fn drain(&mut self) {
+ self.drain_with(crate::ExitMode::KeepHost).await;
+ }
+ pub async fn drain_with(&mut self, mode: crate::ExitMode) {
let mut reported = BTreeSet::new();
loop {
- match self.close().await {
+ match self.close_with(mode).await {
Ok(()) => return,
Err(error) => {
if reported.insert(error.clone()) {
diff --git a/platform/transport/src/connections/host.rs b/platform/transport/src/connections/host.rs
index 7abc405e47..2414aa0a19 100644
--- a/platform/transport/src/connections/host.rs
+++ b/platform/transport/src/connections/host.rs
@@ -17,47 +17,13 @@
* under the License.
*/
-//! Original process/cohost execution and physical-call custody.
+//! Public session attachment and locally owned cohost execution.
use super::*;
-use tokio::process::{Child, ChildStdin, ChildStdout};
-
-pub(super) struct Address {
- pub(super) path: PathBuf,
- // The parent retains the private directory until this same child has closed
- // and actually exited. Failed cleanup never recycles an old address.
- #[cfg(unix)]
- _directory: tempfile::TempDir,
-}
-impl Address {
- pub(super) fn create(nonce: Uuid) -> Result<Self> {
- #[cfg(unix)]
- {
- use std::os::unix::fs::PermissionsExt;
- // Short account-private paths also fit Darwin's sockaddr_un limit.
- let directory = tempfile::Builder::new()
- .prefix("maka-")
- .permissions(std::fs::Permissions::from_mode(0o700))
- .tempdir_in("/tmp")
- .map_err(|e| e.to_string())?;
- let _ = nonce;
- Ok(Self {
- path: directory.path().join("host"),
- _directory: directory,
- })
- }
- #[cfg(windows)]
- {
- Ok(Self {
- path: PathBuf::from(format!(r"\\.\pipe\maka-application-{nonce}")),
- })
- }
- }
-}
-pub(super) struct OwnedHost {
- pub(super) owner: HostOwner,
+pub(super) struct ConnectedHost {
+ pub(super) owner: HostLink,
pub(super) connection: Option<Arc<native::Connection>>,
pub(super) ready: bool,
- // Publication is distinct from bootstrap readiness: an unused controller
+ // Publication is distinct from connection readiness: an unused controller
// must keep custody without obstructing healthy UI-source retention.
pub(super) published: bool,
// Last approved endpoint observations, never executable registrations.
@@ -65,11 +31,9 @@ pub(super) struct OwnedHost {
pub(super) configuration: Host,
pub(super) peers: Arc<crate::transport::Peers>,
}
-pub(super) enum HostOwner {
- Process {
- child: Child,
- control: Box<Control<ChildStdout, ChildStdin>>,
- _address: Address,
+pub(super) enum HostLink {
+ Public {
+ client: Option<Box<maka_host_runtime::application::host::Client>>,
},
Cohost {
application: Option<maka_host_runtime::application::Application>,
@@ -77,11 +41,17 @@ pub(super) enum HostOwner {
captured: Option<crate::transport::Endpoints>,
},
}
-impl OwnedHost {
+impl ConnectedHost {
pub(super) async fn registrations_changed(&mut self) -> Result<bool> {
let current = match &mut self.owner {
- HostOwner::Process { control, .. } => control.epochs().await?,
- HostOwner::Cohost { application, .. } => crate::transport::Endpoints::capture(
+ HostLink::Public { client } => {
+ client
+ .as_mut()
+ .ok_or("public Host session is detached")?
+ .epochs()
+ .await?
+ }
+ HostLink::Cohost { application, .. } => crate::transport::Endpoints::capture(
&application
.as_ref()
.ok_or("cohost is not initialized")?
@@ -98,8 +68,14 @@ impl OwnedHost {
}
pub(super) async fn capture(&mut self) -> Result<crate::transport::Catalogue> {
match &mut self.owner {
- HostOwner::Process { control, .. } => control.capture().await,
- HostOwner::Cohost {
+ HostLink::Public { client } => {
+ client
+ .as_mut()
+ .ok_or("public Host session is detached")?
+ .capture()
+ .await
+ }
+ HostLink::Cohost {
application,
captured,
..
@@ -126,10 +102,11 @@ impl OwnedHost {
catalogue: &Option<crate::transport::Catalogue>,
) -> Result<()> {
match &self.owner {
- HostOwner::Process { control, .. } => {
- control.check_approval(source, consumers, retain, catalogue)
- }
- HostOwner::Cohost {
+ HostLink::Public { client } => client
+ .as_ref()
+ .ok_or("public Host session is detached")?
+ .check_approval(source, consumers, retain, catalogue),
+ HostLink::Cohost {
server, captured, ..
} => {
if server.is_none() || consumers.is_empty() != catalogue.is_none() {
@@ -156,10 +133,14 @@ impl OwnedHost {
) -> Result<()> {
self.check_approval(&source, &consumers, &retain, &catalogue)?;
match &mut self.owner {
- HostOwner::Process { control, .. } => {
- control.approve(source, consumers, retain, catalogue).await
+ HostLink::Public { client } => {
+ client
+ .as_mut()
+ .ok_or("public Host session is detached")?
+ .approve(source, consumers, retain, catalogue)
+ .await
}
- HostOwner::Cohost {
+ HostLink::Cohost {
server, captured, ..
} => {
if catalogue.is_some() {
@@ -172,19 +153,31 @@ impl OwnedHost {
}
}
}
- pub(super) async fn close_owner(&mut self) -> Result<()> {
+ pub(super) async fn close_owner(
+ &mut self,
+ mode: crate::ExitMode,
+ namespaces: &RootNamespaces,
+ ) -> Result<()> {
match &mut self.owner {
- HostOwner::Process { child, control, .. } => {
- let closed = control.close().await;
- let exited = child.wait().await.map_err(|e| e.to_string());
- closed?;
- let status = exited?;
- if self.ready && !status.success() {
- return Err(format!("original Host exited unsuccessfully: {status}"));
+ HostLink::Public { client } => {
+ if mode == crate::ExitMode::Quit
+ && let Some(client) = client.as_mut()
+ && let Err(error) = client.quit().await
+ {
+ let HostLocation::Public { root, .. } = &self.configuration.location else {
+ unreachable!();
+ };
+ if !released(root, namespaces, client).await {
+ return Err(error);
+ }
}
+ // The caller has already confirmed all original physical Calls.
+ // Keeping the Host only detaches this anchor. Normal quit instead
+ // receives the original session's last-client decision above.
+ drop(client.take());
Ok(())
}
- HostOwner::Cohost {
+ HostLink::Cohost {
server,
application,
..
@@ -202,6 +195,25 @@ impl OwnedHost {
}
}
}
+/// An externally stopped Host cannot answer quit. Prove the original owner is
+/// gone without opening its database or treating a disconnected socket as cleanup.
+async fn released(root: &Path, namespaces: &RootNamespaces, client: &Client) -> bool {
+ let domain = &client.source().state_domain;
+ if let Ok(owner) = maka_event_log::root::RootOwner::open(root, namespaces) {
+ return owner.root_id() == domain;
+ }
+ // A replacement may already hold the Root; never send it the old UI's quit.
+ let Ok(found) = discovery::read_with_namespaces(root, namespaces) else {
+ return false;
+ };
+ if found.root_id != *domain || found.host_epoch == client.epoch() {
+ return false;
+ }
+ let Ok(status) = Client::status(&found.endpoint, Some(found.host_epoch)).await else {
+ return false;
+ };
+ found.verify(&status).is_ok()
+}
pub(super) struct CohostConnector {
pub(super) server: Arc<crate::transport::Server>,
pub(super) peers: Arc<crate::transport::Peers>,
diff --git a/platform/transport/src/lib.rs b/platform/transport/src/lib.rs
index c7790bea2e..0550d2e6f3 100644
--- a/platform/transport/src/lib.rs
+++ b/platform/transport/src/lib.rs
@@ -19,18 +19,25 @@
//! Trusted local connections between separate UI and Host deployment owners.
mod connections;
+pub mod startup;
pub mod stdio;
pub mod transport;
mod updates;
-pub use connections::{
- Configuration, Connections, Grant, Host, HostEndpoint, HostExecutable, HostLocation,
-};
+pub use connections::{Configuration, Connections, Grant, Host, HostEndpoint, HostLocation};
pub use maka_host_runtime::application::transport::{GATEWAY, with_gateway};
pub use updates::{ApprovedUpdate, Updates};
/// Compiled fixed-UI-target management availability for explicit cohost entries.
pub const UI_DEPLOYMENT_MANAGER: &str = "maka.application.ui-management";
pub type Result<T> = std::result::Result<T, String>;
+/// The user's choice when closing a UI, after its physical calls have drained.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub enum ExitMode {
+ #[default]
+ Quit,
+ KeepHost,
+}
+
/// Product startup must not exit after losing confirmation of an acquired SQL
/// owner's close. That original worker retains the Root and writer authority.
pub async fn open_store(
diff --git a/platform/transport/src/startup.rs b/platform/transport/src/startup.rs
new file mode 100644
index 0000000000..f38a4cfa5d
--- /dev/null
+++ b/platform/transport/src/startup.rs
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+//! A finite trusted CLI helper holds startup residency until peer admission.
+use crate::{HostEndpoint, Result};
+use maka_host_runtime::application::transport::{NativeGrant, wire};
+use serde::{Deserialize, Serialize};
+use std::{
+ collections::BTreeMap,
+ path::{Path, PathBuf},
+ process::Stdio,
+};
+use tokio::process::{Child, ChildStdin, Command};
+use uuid::Uuid;
+
+pub const REQUEST_LIMIT: usize = 512 * 1024;
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Request {
+ pub root: PathBuf,
+ pub state_domain: Option<String>,
+ pub deployment: Uuid,
+ pub endpoints: BTreeMap<String, HostEndpoint>,
+ pub native: Vec<NativeGrant>,
+}
+impl Request {
+ pub fn validate(&self) -> Result<()> {
+ if !self.root.is_absolute()
+ || self.deployment.is_nil()
+ || self
+ .state_domain
+ .as_ref()
+ .is_some_and(|domain| !crate::connections::valid_domain(domain))
+ || self.endpoints.is_empty()
+ || self.endpoints.len() > 128
+ || self.native.len() > 128
+ {
+ return Err("invalid local Host startup selection".into());
+ }
+ for (name, endpoint) in &self.endpoints {
+ if name.is_empty() || name.len() > 256 {
+ return Err("invalid Host endpoint name".into());
+ }
+ endpoint.validate()?;
+ }
+ for grant in &self.native {
+ grant.validate()?;
+ }
+ Ok(())
+ }
+}
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Ready {
+ pub root_id: String,
+ pub epoch: Uuid,
+}
+/// Owns only a finite helper's startup pipe, never a Host process or UI calls.
+/// Dropping it releases startup residency after the real peer is connected.
+pub(crate) struct Anchor {
+ _child: Child,
+ _input: ChildStdin,
+}
+pub(crate) async fn launch(executable: &Path, request: Request) -> Result<(Anchor, Ready)> {
+ request.validate()?;
+ wire::encode_bounded(&request, REQUEST_LIMIT)?;
+ let mut child = Command::new(executable)
+ .args(["host", "ensure"])
+ .stdin(Stdio::piped())
+ .stdout(Stdio::piped())
+ .stderr(Stdio::inherit())
+ .spawn()
+ .map_err(|error| format!("local Host startup helper: {error}"))?;
+ let mut input = child.stdin.take().expect("piped startup input");
+ let output = child.stdout.take().expect("piped startup output");
+ wire::write_bounded(&mut input, &request, REQUEST_LIMIT).await?;
+ let ready: Ready = wire::Reader::new(output).read_bounded(4096).await?;
+ if !crate::connections::valid_domain(&ready.root_id)
+ || ready.epoch.is_nil()
+ || request
+ .state_domain
+ .as_ref()
+ .is_some_and(|expected| *expected != ready.root_id)
+ {
+ return Err("local Host startup returned a foreign identity".into());
+ }
+ Ok((
+ Anchor {
+ _child: child,
+ _input: input,
+ },
+ ready,
+ ))
+}
diff --git a/platform/transport/src/transport.rs b/platform/transport/src/transport.rs
index ef0e4786e1..8b1d132efd 100644
--- a/platform/transport/src/transport.rs
+++ b/platform/transport/src/transport.rs
@@ -17,11 +17,9 @@
* under the License.
*/
-//! UI adapters and trusted process bootstrap for the shared Application transport.
+//! UI adapters for authenticated independent Host sessions.
pub mod ipc;
pub mod native;
-pub mod process;
-use maka_host_runtime::application::transport::wire;
pub use maka_host_runtime::application::transport::{
Call, Catalogue, Description, Endpoint, Endpoints, NativeGrant, Peer, PeerPolicy, Peers, Reply,
Selection, Server, Source,
diff --git a/platform/transport/src/transport/ipc.rs b/platform/transport/src/transport/ipc.rs
index bac22c6256..7986da9ab1 100644
--- a/platform/transport/src/transport/ipc.rs
+++ b/platform/transport/src/transport/ipc.rs
@@ -17,58 +17,34 @@
* under the License.
*/
-//! Physical pairing on private current-account IPC, not network authentication.
-use super::{native, wire};
+//! Fixed physical channels for an authenticated independent Host session.
+use super::native;
use crate::Result;
use futures_util::future::BoxFuture;
-use std::path::PathBuf;
use tokio_util::sync::CancellationToken;
-use uuid::Uuid;
-use wire::{Accepted, Hello};
-/// Only the trusted launcher constructs this from its actual child bootstrap.
-/// The address and nonce stay fixed; a lost Host is never retargeted by alias.
+/// Physical channels for one independently owned, already authenticated Host
+/// session. This snapshot never retains the anchor or retargets a restarted Host.
pub struct Connector {
- path: PathBuf,
- nonce: Uuid,
+ opener: maka_host_runtime::application::host::CallChannelOpener,
}
impl Connector {
- pub fn new(path: PathBuf, nonce: Uuid) -> Result<Self> {
- if nonce.is_nil() {
- return Err("private channel nonce must be nonnil".into());
- }
- Ok(Self { path, nonce })
+ pub fn new(opener: maka_host_runtime::application::host::CallChannelOpener) -> Self {
+ Self { opener }
}
}
impl native::Connector for Connector {
fn connect(&self, stop: CancellationToken) -> BoxFuture<'static, Result<native::Stream>> {
- let path = self.path.clone();
- let nonce = self.nonce;
+ let opener = self.opener.clone();
Box::pin(async move {
- // Before Open, no Application request or immutable resource can be
- // admitted. Dropped OS-opening/Hello custody remains within the
- // original embedding's tracked Server, which its child owner drains.
- let mut stream = tokio::select! {
+ // No business Open or resources have been sent. Cancellation drops
+ // only inert IPC; the independent Host retains its own channel tasks.
+ let stream = tokio::select! {
biased;
- () = stop.cancelled() => return Err("private opening cancelled".into()),
- result = maka_host_runtime::ipc::connect(&path) => result.map_err(|e| e.to_string())?,
+ () = stop.cancelled() => return Err("public opening cancelled".into()),
+ stream = opener.open() => stream?,
};
- let hello = Hello { nonce };
- tokio::select! {
- biased;
- () = stop.cancelled() => return Err("private opening cancelled".into()),
- result = wire::write(&mut stream, &hello) => result?,
- }
- let mut reader = wire::Reader::new(stream);
- let accepted: Accepted = tokio::select! {
- biased;
- () = stop.cancelled() => return Err("private opening cancelled".into()),
- result = reader.read_bounded(512) => result?,
- };
- if accepted.nonce != nonce {
- return Err("private channel belongs to a different Host incarnation".into());
- }
- Ok(Box::new(reader.inner) as native::Stream)
+ Ok(Box::new(stream) as native::Stream)
})
}
}
diff --git a/platform/transport/src/transport/native.rs b/platform/transport/src/transport/native.rs
index 98d90eb9df..257efe47b3 100644
--- a/platform/transport/src/transport/native.rs
+++ b/platform/transport/src/transport/native.rs
@@ -60,8 +60,9 @@ pub type Stream = Box<dyn Channel>;
/// Trusted physical admission only. A fresh channel belongs to the same
/// authenticated Host incarnation that published the frozen Selection.
/// Cancellation requests settlement of this same opening. Its future must
-/// complete before custody is released; an error confirms that no execution or
-/// physical custody remains outside the embedding. Dropping or panicking an
+/// complete before custody is released; an error confirms that no caller input,
+/// operation token or admitted execution escaped. An independent Host may still
+/// own inert pre-Open channels. Dropping or panicking an
/// opening observer does not establish that confirmation.
pub trait Connector: Send + Sync + 'static {
fn connect(&self, stop: CancellationToken) -> BoxFuture<'static, Result<Stream>>;
diff --git a/platform/transport/src/transport/process.rs b/platform/transport/src/transport/process.rs
deleted file mode 100644
index c9a7d7d91e..0000000000
--- a/platform/transport/src/transport/process.rs
+++ /dev/null
@@ -1,785 +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.
- */
-
-//! Shared Host process bootstrap and actual lifecycle acknowledgements.
-use super::{Catalogue, Endpoints, Peer, PeerPolicy, Server, wire};
-use crate::{HostEndpoint, Result};
-use maka_event_log::{
- EventLog,
- root::{RootNamespaces, RootOwner},
-};
-use maka_host_runtime::{
- application::{Application, NativeRegistry},
- ipc::LocalListener,
-};
-use maka_plugin_core::{composition::InstanceId, deployment::DeploymentRef};
-use serde::{Deserialize, Serialize};
-use std::{
- collections::{BTreeMap, BTreeSet},
- path::PathBuf,
- sync::Arc,
-};
-use tokio::io::{AsyncRead, AsyncWrite};
-use uuid::Uuid;
-const BOOTSTRAP_LIMIT: usize = 512 * 1024;
-
-/// Trusted parent input on its selected child's inherited stdin, never an IPC
-/// request or plugin-created grant. The child owns exactly one original Host.
-#[derive(Serialize, Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct Bootstrap {
- pub root: PathBuf,
- #[serde(default)]
- pub state_domain: Option<String>,
- pub deployment: Uuid,
- pub namespaces: RootNamespaces,
- pub source: DeploymentRef,
- pub consumers: BTreeSet<InstanceId>,
- pub endpoints: BTreeMap<String, HostEndpoint>,
- pub native: Vec<super::NativeGrant>,
- pub address: PathBuf,
- pub nonce: Uuid,
-}
-impl Bootstrap {
- pub fn validate(&self) -> Result<()> {
- Peer::new(self.source.clone(), self.consumers.clone())?;
- if self
- .state_domain
- .as_deref()
- .is_some_and(|domain| !crate::connections::valid_domain(domain))
- {
- return Err("invalid Host State Root identity".into());
- }
- if self.deployment.is_nil()
- || self.nonce.is_nil()
- || self.endpoints.is_empty()
- || self.endpoints.len() > 128
- || self
- .endpoints
- .keys()
- .any(|name| name.is_empty() || name.len() > 256)
- {
- return Err("invalid Host bootstrap".into());
- }
- for endpoint in self.endpoints.values() {
- endpoint.validate()?;
- }
- if self.native.len() > 128 {
- return Err("maximum 128 native service grants".into());
- }
- for grant in &self.native {
- grant.validate()?;
- }
- // The parent checks the actual wire bound before spawning any child.
- // An oversized bootstrap must not leave a child waiting on partial IO.
- wire::encode_bounded(self, BOOTSTRAP_LIMIT)?;
- Ok(())
- }
-}
-#[derive(Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
-enum Event {
- Epochs {
- nonce: Uuid,
- result: Result<BTreeMap<String, Option<Uuid>>>,
- },
- Ready {
- nonce: Uuid,
- catalogue: Catalogue,
- },
- Closed {
- nonce: Uuid,
- #[serde(default, skip_serializing_if = "Option::is_none")]
- error: Option<String>,
- },
- Captured {
- nonce: Uuid,
- result: Result<Catalogue>,
- },
- Approved {
- nonce: Uuid,
- source: DeploymentRef,
- result: Result<()>,
- },
-}
-#[derive(Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
-enum Command {
- Epochs {
- nonce: Uuid,
- },
- Close {
- nonce: Uuid,
- },
- Capture {
- nonce: Uuid,
- },
- Approve {
- nonce: Uuid,
- source: DeploymentRef,
- consumers: BTreeSet<InstanceId>,
- retain: BTreeSet<u64>,
- captured: Option<Catalogue>,
- },
-}
-#[derive(PartialEq)]
-enum Phase {
- Inert,
- Started,
- Closing,
- Closed,
- Lost,
-}
-/// Own alongside the actual Child handle. Closed only proves this same child's
-/// aggregate cleanup; normal process exit is a separate required observation.
-/// Neither observation settles an old Lost Call or its actual caller token.
-pub struct Control<R, W> {
- reader: wire::Reader<R>,
- writer: Option<W>,
- nonce: Uuid,
- phase: Phase,
-}
-impl<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> Control<R, W> {
- pub fn new(reader: R, writer: W, nonce: Uuid) -> Result<Self> {
- if nonce.is_nil() {
- return Err("Host control nonce must be nonnil".into());
- }
- Ok(Self {
- reader: wire::Reader::new(reader),
- writer: Some(writer),
- nonce,
- phase: Phase::Inert,
- })
- }
- pub async fn start(&mut self, bootstrap: &Bootstrap) -> Result<Catalogue> {
- if self.phase != Phase::Inert || bootstrap.nonce != self.nonce {
- return Err("Host control cannot bootstrap in this phase".into());
- }
- bootstrap.validate()?;
- self.phase = Phase::Lost;
- wire::write_bounded(self.writer.as_mut().unwrap(), bootstrap, BOOTSTRAP_LIMIT).await?;
- match self.reader.read().await? {
- Event::Ready { nonce, catalogue } if nonce == self.nonce => {
- catalogue.validate()?;
- self.phase = Phase::Started;
- Ok(catalogue)
- }
- Event::Closed { nonce, error } if nonce == self.nonce => {
- self.phase = Phase::Closed;
- Err(error.unwrap_or_else(|| {
- "Host closed before publishing its original registrations".into()
- }))
- }
- _ => Err("Host control did not publish its original registrations".into()),
- }
- }
- /// Bounded observation only. The original captured bindings and approval
- /// exchange still own every subsequent publication.
- pub(crate) async fn epochs(&mut self) -> Result<BTreeMap<String, Option<Uuid>>> {
- if self.phase != Phase::Started {
- return Err("Host control is not ready for observation".into());
- }
- self.phase = Phase::Lost;
- wire::write(
- self.writer.as_mut().unwrap(),
- &Command::Epochs { nonce: self.nonce },
- )
- .await?;
- match self.reader.read().await? {
- Event::Epochs { nonce, result } if nonce == self.nonce => {
- self.phase = Phase::Started;
- result
- }
- _ => Err("Host observation differs from its original incarnation".into()),
- }
- }
- /// Capture actual registrations without publishing endpoints or approvals.
- pub async fn capture(&mut self) -> Result<Catalogue> {
- if self.phase != Phase::Started {
- return Err("Host control is not ready for capture".into());
- }
- self.phase = Phase::Lost;
- wire::write(
- self.writer.as_mut().unwrap(),
- &Command::Capture { nonce: self.nonce },
- )
- .await?;
- match self.reader.read().await? {
- Event::Captured { nonce, result } if nonce == self.nonce => {
- self.phase = Phase::Started;
- let catalogue = result?;
- catalogue.validate()?;
- Ok(catalogue)
- }
- _ => Err("Host capture acknowledgement differs from its original incarnation".into()),
- }
- }
- pub(crate) fn check_approval(
- &self,
- source: &DeploymentRef,
- consumers: &BTreeSet<InstanceId>,
- retain: &BTreeSet<u64>,
- captured: &Option<Catalogue>,
- ) -> Result<()> {
- if self.phase != Phase::Started {
- return Err("Host control is not ready for approval".into());
- }
- source.validate()?;
- if !consumers.is_empty() {
- Peer::new(source.clone(), consumers.clone())?;
- }
- if consumers.is_empty() != captured.is_none() {
- return Err("approval requires its captured endpoints".into());
- }
- if retain.len() > 256 {
- return Err("maximum 256 retained source approvals".into());
- }
- wire::encode_bounded(
- &Command::Approve {
- nonce: self.nonce,
- source: source.clone(),
- consumers: consumers.clone(),
- retain: retain.clone(),
- captured: captured.clone(),
- },
- wire::FRAME_LIMIT,
- )?;
- Ok(())
- }
- /// Publish the previously captured endpoints and exact caller approval.
- /// An empty consumer set performs only retention, without rebinding endpoints.
- pub async fn approve(
- &mut self,
- source: DeploymentRef,
- consumers: BTreeSet<InstanceId>,
- retain: BTreeSet<u64>,
- captured: Option<Catalogue>,
- ) -> Result<()> {
- self.check_approval(&source, &consumers, &retain, &captured)?;
- let command = Command::Approve {
- nonce: self.nonce,
- source: source.clone(),
- consumers,
- retain,
- captured,
- };
- wire::encode_bounded(&command, wire::FRAME_LIMIT)?;
- // An abandoned/partial exchange cannot append another control frame.
- self.phase = Phase::Lost;
- wire::write(self.writer.as_mut().unwrap(), &command).await?;
- match self.reader.read().await? {
- Event::Approved {
- nonce,
- source: approved,
- result,
- } if nonce == self.nonce && approved == source => {
- self.phase = Phase::Started;
- result
- }
- _ => Err("Host approval acknowledgement differs from its original request".into()),
- }
- }
- pub async fn close(&mut self) -> Result<()> {
- if self.phase == Phase::Closed {
- return Ok(());
- }
- if self.phase == Phase::Started {
- self.phase = Phase::Lost;
- if wire::write(
- self.writer.as_mut().unwrap(),
- &Command::Close { nonce: self.nonce },
- )
- .await
- .is_ok()
- {
- self.phase = Phase::Closing;
- } else {
- // Preserve this same child's possible buffered acknowledgement.
- // Never append a command after an uncertain prefix.
- drop(self.writer.take());
- }
- } else if self.phase != Phase::Closing {
- // A partial control write cannot append another command. EOF asks
- // the same child to drain; retain its stdout to observe real Closed.
- drop(self.writer.take());
- }
- loop {
- let event = match self.reader.read().await {
- Ok(event) => event,
- Err(error) => {
- self.phase = Phase::Lost;
- drop(self.writer.take());
- return Err(error);
- }
- };
- match event {
- Event::Closed { nonce, .. } if nonce == self.nonce => {
- self.phase = Phase::Closed;
- return Ok(());
- }
- Event::Ready { nonce, .. }
- | Event::Epochs { nonce, .. }
- | Event::Captured { nonce, .. }
- | Event::Approved { nonce, .. }
- if nonce == self.nonce && self.phase == Phase::Lost => {}
- _ => {
- self.phase = Phase::Lost;
- drop(self.writer.take());
- return Err(
- "Host lifecycle acknowledgement changed its original incarnation".into(),
- );
- }
- }
- }
- }
-}
-/// Process entrypoint. Availability is supplied by this product's compiled
-/// binary; persisted Host composition selects all executed instances.
-pub async fn serve(
- native: NativeRegistry,
- scripts: Option<Arc<dyn maka_host_runtime::application::backend::Factory>>,
-) -> Result<()> {
- let native = crate::with_gateway(native)?;
- let crate::stdio::Stdio {
- input,
- mut output,
- output_done,
- } = crate::stdio::Stdio::open().map_err(|e| e.to_string())?;
- let mut input = wire::Reader::new(input);
- let bootstrap: Bootstrap = input.read_bounded(BOOTSTRAP_LIMIT).await?;
- bootstrap.validate()?;
- let peer = super::Peers::new(Peer::new(
- bootstrap.source.clone(),
- bootstrap.consumers.clone(),
- )?);
- let initialized = async {
- let listener = LocalListener::bind(&bootstrap.address).map_err(|e| e.to_string())?;
- let owner = Arc::new(
- RootOwner::open(&bootstrap.root, &bootstrap.namespaces).map_err(|e| e.to_string())?,
- );
- if bootstrap
- .state_domain
- .as_deref()
- .is_some_and(|domain| owner.root_id() != domain)
- {
- return Err("original Host State Root identity changed".into());
- }
- let store = match EventLog::for_root(owner).await {
- Ok(store) => Arc::new(store),
- Err(error @ maka_event_log::StoreError::StartupCleanupFailed { .. }) => {
- // The original SQL worker still holds its actual authority.
- // Neither process exit nor an aggregate Closed can settle it.
- eprintln!("Host SQL startup cleanup unconfirmed: {error}");
- std::future::pending::<()>().await;
- unreachable!();
- }
- Err(error) => return Err(error.to_string()),
- };
- maka_host_runtime::recovery::recover(&store)
- .await
- .map_err(|error| error.to_string())?;
- Ok::<_, String>((listener, store))
- }
- .await;
- let (mut listener, store) = match initialized {
- Ok(initialized) => initialized,
- Err(error) => {
- // No Application was admitted. Every successfully acquired SQL
- // owner has confirmed its real close before reaching this branch.
- wire::write(
- &mut output,
- &Event::Closed {
- nonce: bootstrap.nonce,
- error: Some(startup_diagnostic(&error)),
- },
- )
- .await?;
- drop(output);
- tokio::time::timeout(std::time::Duration::from_secs(5), output_done)
- .await
- .map_err(|e| e.to_string())?
- .map_err(|e| e.to_string())?
- .map_err(|e| e.to_string())?;
- return Err(error);
- }
- };
- let policy = Arc::new(PeerPolicy::with_native_grants(bootstrap.native)?);
- let application = Application::start_from_store(
- store.clone(),
- bootstrap.deployment,
- native,
- policy.issuer(),
- policy.clone(),
- scripts,
- )
- .await;
- let result = match application {
- Ok(application) => {
- let result = async {
- let client = application.client();
- let endpoints = Endpoints::start(&client, &bootstrap.endpoints).await?;
- let server = Server::new(client.clone(), policy, endpoints)?;
- let ready = Event::Ready {
- nonce: bootstrap.nonce,
- catalogue: server.catalogue(),
- };
- let published = wire::write(&mut output, &ready).await;
- let mut control_error = None;
- let mut captured = None;
- if published.is_ok() {
- loop {
- tokio::select! {
- biased;
- command = input.read::<Command>() => {
- match command {
- Ok(Command::Epochs { nonce }) if nonce == bootstrap.nonce => {
- let result = Endpoints::capture(&client, &bootstrap.endpoints).await.map(|endpoints| endpoints.epochs());
- if let Err(error) = wire::write(&mut output, &Event::Epochs { nonce, result }).await {
- control_error = Some(error); break;
- }
- }
- Ok(Command::Capture { nonce }) if nonce == bootstrap.nonce => {
- let result = Endpoints::capture(&client, &bootstrap.endpoints).await;
- let result = match result {
- Ok(endpoints) => {
- let catalogue = Catalogue::captured(&endpoints);
- captured = Some(endpoints);
- Ok(catalogue)
- }
- Err(error) => { captured = None; Err(error) }
- };
- let reply = capture_reply(nonce, result);
- if matches!(&reply, Event::Captured { result: Err(_), .. }) { captured = None; }
- if let Err(error) = wire::write(&mut output, &reply).await {
- control_error = Some(error); break;
- }
- }
- Ok(Command::Approve { nonce, source, consumers, retain, captured: expected }) if nonce == bootstrap.nonce => {
- let result = (|| {
- source.validate()?;
- if source.state_domain != bootstrap.source.state_domain || source.deployment != bootstrap.source.deployment {
- return Err("foreign approval deployment".into());
- }
- let approved = if consumers.is_empty() { None } else { Some(Peer::new(source.clone(), consumers)?) };
- if approved.is_none() != expected.is_none() { return Err("approval requires its captured endpoints".into()); }
- peer.check_update(approved.clone(), &retain)?;
- if let Some(expected) = expected {
- let endpoints = captured.take().ok_or("Host endpoints were not captured")?;
- if Catalogue::captured(&endpoints) != expected { return Err("Host captured endpoints differ".into()); }
- // Publish original Bindings, never rebind an alias after preflight.
- server.replace_endpoints(endpoints)?;
- }
- peer.update(approved, &retain)
- })();
- if let Err(error) = wire::write(&mut output, &Event::Approved { nonce, source, result }).await {
- control_error = Some(error); break;
- }
- }
- Ok(Command::Close { nonce }) if nonce == bootstrap.nonce => break,
- _ => { control_error = Some("Host control ended or named a foreign incarnation".into()); break; }
- }
- }
- accepted = listener.accept() => {
- let stream = match accepted {
- Ok(stream) => stream,
- Err(reason) => {
- control_error = Some(format!("Host IPC accept failed: {:?}", reason.kind()));
- break;
- }
- };
- let (reader, writer) = tokio::io::split(stream);
- // Hello belongs to original wire capacity/tracked custody.
- // Dropping this observer never drops the actual work.
- if let Ok(observer) = server.serve_private(reader, writer, peer.clone(), bootstrap.nonce) {
- drop(observer);
- }
- }
- }
- }
- }
- // Revoke acceptance before waiting for actual invocation cleanup.
- drop(listener);
- drain_server(&server).await;
- drop(server);
- published?;
- control_error.map_or(Ok(()), Err)
- }.await;
- drain_application(&application).await;
- drop(application);
- result
- }
- Err(reason) => {
- // No Application owner was returned. Revoke physical acceptance and
- // still observe the original SQL close before publishing Closed.
- drop(listener);
- Err(reason)
- }
- };
- store.drain().await;
- drop(store);
- // Root, SQL and all accepted scopes have actually closed before this frame.
- wire::write(
- &mut output,
- &Event::Closed {
- nonce: bootstrap.nonce,
- error: result.as_ref().err().map(|error| startup_diagnostic(error)),
- },
- )
- .await?;
- drop(output);
- tokio::time::timeout(std::time::Duration::from_secs(5), output_done)
- .await
- .map_err(|e| e.to_string())?
- .map_err(|e| e.to_string())?
- .map_err(|e| e.to_string())?;
- result
-}
-fn startup_diagnostic(error: &str) -> String {
- // At most 64 KiB before JSON escaping, including non-ASCII diagnostics.
- error.chars().take(16 * 1024).collect()
-}
-fn capture_reply(nonce: Uuid, result: Result<Catalogue>) -> Event {
- let event = Event::Captured { nonce, result };
- match wire::encode_bounded(&event, wire::FRAME_LIMIT) {
- Ok(_) => event,
- // Known serialization bounds fail before IO; they must not close the Host.
- Err(_) => Event::Captured {
- nonce,
- result: Err("Host captured catalogue exceeds control frame limit".into()),
- },
- }
-}
-async fn drain_server(server: &Server) {
- let mut reported = BTreeSet::new();
- loop {
- match server.close().await {
- Ok(()) => return,
- Err(reason) => {
- if reported.insert(reason.clone()) {
- eprintln!("Host physical cleanup unconfirmed: {reason:?}");
- }
- }
- }
- tokio::time::sleep(std::time::Duration::from_millis(50)).await;
- }
-}
-async fn drain_application(application: &Application) {
- let mut reported = BTreeSet::new();
- loop {
- match application.shutdown().await {
- Ok(()) => return,
- Err(reason) => {
- if reported.insert(reason.clone()) {
- eprintln!("Host Application cleanup unconfirmed: {reason:?}");
- }
- }
- }
- tokio::time::sleep(std::time::Duration::from_millis(50)).await;
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- #[tokio::test]
- async fn abandoned_observation_keeps_the_original_close_acknowledgement() {
- let nonce = Uuid::new_v4();
- // Separate pipes match ChildStdin/ChildStdout: dropping only the
- // original writer must deliver EOF while its reader remains owned.
- let (writer, child_reader) = tokio::io::duplex(4096);
- let (mut child_writer, reader) = tokio::io::duplex(4096);
- let mut control = Control::new(reader, writer, nonce).unwrap();
- control.phase = Phase::Started;
- let received = std::sync::Arc::new(tokio::sync::Notify::new());
- let observed = received.clone();
- let peer = tokio::spawn(async move {
- let mut reader = wire::Reader::new(child_reader);
- assert!(
- matches!(reader.read::<Command>().await.unwrap(), Command::Epochs { nonce: actual } if actual == nonce)
- );
- observed.notify_one();
- // An abandoned exchange closes its original writer. It must not
- // append another command behind a potentially partial observation.
- assert!(reader.read::<Command>().await.is_err());
- wire::write(
- &mut child_writer,
- &Event::Epochs {
- nonce,
- result: Ok(BTreeMap::new()),
- },
- )
- .await
- .unwrap();
- wire::write(&mut child_writer, &Event::Closed { nonce, error: None })
- .await
- .unwrap();
- });
- {
- let observation = control.epochs();
- tokio::pin!(observation);
- tokio::select! {
- _ = &mut observation => panic!("peer has not acknowledged observation"),
- _ = received.notified() => {}
- }
- }
- tokio::time::timeout(std::time::Duration::from_secs(5), control.close())
- .await
- .unwrap()
- .unwrap();
- peer.await.unwrap();
- }
-
- #[tokio::test]
- async fn failed_bootstrap_returns_original_error_and_confirms_cleanup() {
- let nonce = Uuid::new_v4();
- let consumer = InstanceId(vec![Uuid::new_v4()]);
- let bootstrap = Bootstrap {
- root: "/unused/host".into(),
- state_domain: None,
- deployment: Uuid::new_v4(),
- namespaces: RootNamespaces {
- ownership: "/unused/owners".into(),
- control: "/unused/control".into(),
- },
- source: DeploymentRef {
- state_domain: "a".repeat(64),
- deployment: Uuid::new_v4(),
- revision: 1,
- },
- consumers: [consumer.clone()].into(),
- endpoints: [(
- "agent".into(),
- HostEndpoint {
- consumer,
- import: "agent".into(),
- grant: maka_host_runtime::application::transport::Grant {
- methods: ["read".into()].into(),
- recovery: false,
- },
- },
- )]
- .into(),
- native: vec![],
- address: "/unused/address".into(),
- nonce,
- };
- let (writer, child_reader) = tokio::io::duplex(4096);
- let (mut child_writer, reader) = tokio::io::duplex(4096);
- let mut control = Control::new(reader, writer, nonce).unwrap();
- let peer = tokio::spawn(async move {
- let mut reader = wire::Reader::new(child_reader);
- assert_eq!(reader.read::<Bootstrap>().await.unwrap().nonce, nonce);
- wire::write(
- &mut child_writer,
- &Event::Closed {
- nonce,
- error: Some("original Host State Root identity changed".into()),
- },
- )
- .await
- .unwrap();
- });
- assert_eq!(
- control.start(&bootstrap).await.unwrap_err(),
- "original Host State Root identity changed"
- );
- assert!(control.phase == Phase::Closed);
- control.close().await.unwrap();
- peer.await.unwrap();
- }
-
- #[test]
- fn oversized_capture_is_a_bounded_error_acknowledgement() {
- let contract = serde_json::json!({
- "name":"large.Service", "version":1, "transport":"remote",
- "interface":{"kind":"service","methods":{"read":{"mode":"query","input":{"description":"x".repeat(wire::FRAME_LIMIT)},"output":{},"resources":{}}}}
- });
- // Read-only descriptors cannot manufacture any Binding or grant.
- let catalogue: Catalogue = serde_json::from_value(serde_json::json!({"endpoints":{"large":{
- "provider":[Uuid::from_u128(1)],"export":"Large","contract":contract,"registration_epoch":Uuid::from_u128(2)
- }}})).unwrap();
- let nonce = Uuid::from_u128(3);
- let reply = capture_reply(nonce, Ok(catalogue));
- assert!(
- matches!(&reply, Event::Captured { nonce: actual, result: Err(_) } if *actual == nonce)
- );
- wire::encode_bounded(&reply, wire::FRAME_LIMIT).unwrap();
- }
-
- #[tokio::test]
- async fn captured_catalogue_above_bootstrap_limit_can_be_approved() {
- let catalogue: Catalogue = serde_json::from_value(serde_json::json!({"endpoints":{"large":{
- "provider":[Uuid::from_u128(1)],"export":"Large","contract":{
- "name":"large.Service","version":1,"transport":"remote",
- "interface":{"kind":"service","methods":{"read":{"mode":"query",
- "input":{"description":"x".repeat(BOOTSTRAP_LIMIT)},"output":{},"resources":{}}}}
- },"registration_epoch":Uuid::from_u128(2)
- }}})).unwrap();
- let nonce = Uuid::new_v4();
- let source = DeploymentRef {
- state_domain: "a".repeat(64),
- deployment: Uuid::new_v4(),
- revision: 2,
- };
- let consumers = BTreeSet::from([InstanceId(vec![Uuid::new_v4()])]);
- let retain = BTreeSet::from([1, 2]);
- let (writer, child_reader) = tokio::io::duplex(4096);
- let (mut child_writer, reader) = tokio::io::duplex(4096);
- let mut control = Control::new(reader, writer, nonce).unwrap();
- control.phase = Phase::Started;
- let expected = catalogue.clone();
- let expected_source = source.clone();
- let peer = tokio::spawn(async move {
- let command: Command = wire::Reader::new(child_reader).read().await.unwrap();
- let Command::Approve {
- nonce: actual,
- source,
- captured,
- ..
- } = command
- else {
- panic!("expected approval")
- };
- assert_eq!(actual, nonce);
- assert_eq!(source, expected_source);
- assert_eq!(captured, Some(expected));
- wire::write(
- &mut child_writer,
- &Event::Approved {
- nonce,
- source,
- result: Ok(()),
- },
- )
- .await
- .unwrap();
- });
- // Capture uses the normal frame bound, and the matching approval must too.
- assert!(matches!(
- capture_reply(nonce, Ok(catalogue.clone())),
- Event::Captured { result: Ok(_), .. }
- ));
- control
- .approve(source, consumers, retain, Some(catalogue))
- .await
- .unwrap();
- assert!(control.phase == Phase::Started);
- peer.await.unwrap();
- }
-}
diff --git a/platform/transport/tests/owners.rs b/platform/transport/tests/owners.rs
index c716645510..1ace4d0ba8 100644
--- a/platform/transport/tests/owners.rs
+++ b/platform/transport/tests/owners.rs
@@ -188,7 +188,7 @@ async fn cancelled_cohost_start_keeps_original_initializer_native_file_and_sql_u
composition::PresentationTarget,
package::{EntryTarget, Implementation},
};
- use maka_transport::{Connections, HostExecutable};
+ use maka_transport::Connections;
let directory = tempfile::tempdir().unwrap();
let root = directory.path().join("cohost");
let namespaces = RootNamespaces {
@@ -284,7 +284,7 @@ async fn cancelled_cohost_start_keeps_original_initializer_native_file_and_sql_u
PresentationTarget::Desktop,
configuration,
namespaces.clone(),
- HostExecutable::new("absent-cohost-executable".into(), vec![]).unwrap(),
+ std::path::PathBuf::from("unused-cohost-startup-helper"),
native,
Some(maka_host_js::factory(Default::default())),
)
@@ -348,7 +348,7 @@ async fn cancelled_cohost_start_keeps_original_initializer_native_file_and_sql_u
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cohost_retains_independent_receipts_when_selected_business_cannot_start() {
use maka_plugin_core::composition::PresentationTarget;
- use maka_transport::{Connections, HostExecutable};
+ use maka_transport::Connections;
tokio::time::timeout(std::time::Duration::from_secs(20), async {
let directory = tempfile::tempdir().unwrap();
let root = directory.path().join("cohost");
@@ -366,7 +366,7 @@ async fn cohost_retains_independent_receipts_when_selected_business_cannot_start
"name":"local", "location":{"kind":"cohost"}, "deployment":host, "endpoints":{"business":{"consumer":support::identity(13),"import":"business","grant":{"methods":["read","write"],"recovery":false}},"receipts":{"consumer":support::identity(14),"import":"receipts","grant":{"methods":["read"],"recovery":true}}}
}]})).unwrap();
let mut connections = Connections::prepare(store.clone(), ui.deployment, PresentationTarget::Tui,
- configuration, namespaces.clone(), HostExecutable::new("unused-cohost-program".into(), vec![]).unwrap(), NativeRegistry::default(),
+ configuration, namespaces.clone(), std::path::PathBuf::from("unused-cohost-startup-helper"), NativeRegistry::default(),
Some(maka_host_js::factory(Default::default())),).await.unwrap();
connections.start().await.unwrap();
assert_eq!(connections.registrations().len(), 1);
diff --git a/platform/transport/tests/private.rs b/platform/transport/tests/private.rs
index 7c9ffc4773..6b83cd3ef3 100644
--- a/platform/transport/tests/private.rs
+++ b/platform/transport/tests/private.rs
@@ -483,10 +483,21 @@ async fn native_ui(
Arc<maka_transport::transport::native::Connection>,
Arc<std::sync::Mutex<Option<maka_ui_runtime::actions::ServiceSource>>>,
Arc<dyn maka_ui_runtime::actions::NativeService>,
+) {
+ native_ui_at(f, connector, f.server.catalogue()).await
+}
+async fn native_ui_at(
+ f: &Fixture,
+ connector: Arc<dyn maka_transport::transport::native::Connector>,
+ catalogue: maka_transport::transport::Catalogue,
+) -> (
+ maka_ui_runtime::Application,
+ Arc<maka_transport::transport::native::Connection>,
+ Arc<std::sync::Mutex<Option<maka_ui_runtime::actions::ServiceSource>>>,
+ Arc<dyn maka_ui_runtime::actions::NativeService>,
) {
use maka_transport::transport::native::{Connection, Service};
let connection = Connection::new(connector);
- let catalogue = f.server.catalogue();
let source = Arc::new(std::sync::Mutex::new(None));
let business = Arc::new(
Service::new(
@@ -639,36 +650,75 @@ async fn stop_ui(mut ui: maka_ui_runtime::Application, root: &maka_plugin_core::
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_adapter_runs_checked_ui_actions_actual_caller_sql_and_original_recovery() {
+ use maka_host_runtime::application::{
+ host::{Client, Host},
+ transport::HostEndpoint,
+ };
let f = Fixture::new().await;
- let nonce = Uuid::new_v4();
+ let deployment = f.app.source().unwrap().deployment;
+ f.server.close().await.unwrap();
+ f.app.shutdown().await.unwrap();
+ let selections = BTreeMap::from([
+ (
+ "business".into(),
+ HostEndpoint {
+ consumer: support::identity(13),
+ import: "business".into(),
+ grant: Grant {
+ methods: ["read".into(), "write".into()].into(),
+ recovery: false,
+ },
+ },
+ ),
+ (
+ "receipts".into(),
+ HostEndpoint {
+ consumer: support::identity(13),
+ import: "receipts".into(),
+ grant: Grant {
+ methods: ["read".into()].into(),
+ recovery: true,
+ },
+ },
+ ),
+ ]);
+ let host = Host::open(
+ f.store.clone(),
+ deployment,
+ NativeRegistry::default(),
+ Some(maka_host_js::factory(Default::default())),
+ selections.clone(),
+ vec![],
+ )
+ .await
+ .unwrap();
#[cfg(unix)]
let path = f._directory.path().join("native.sock");
#[cfg(windows)]
- let path = std::path::PathBuf::from(format!(r"\\.\pipe\maka-native-test-{nonce}"));
- let mut listener = maka_host_runtime::ipc::LocalListener::bind(&path).unwrap();
- let stop = tokio_util::sync::CancellationToken::new();
- let accept_stop = stop.clone();
- let server = f.server.clone();
- let peer = f.peer.clone();
- let acceptance = tokio::spawn(async move {
- loop {
- let stream = tokio::select! {
- biased;
- ()=accept_stop.cancelled()=>break,
- stream=listener.accept()=>stream.unwrap(),
- };
- let (reader, writer) = tokio::io::split(stream);
- drop(
- server
- .serve_private(reader, writer, peer.clone(), nonce)
- .unwrap(),
- );
+ let path = std::path::PathBuf::from(format!(r"\\.\pipe\maka-native-test-{}", Uuid::new_v4()));
+ let listener = maka_host_runtime::ipc::LocalListener::bind(&path).unwrap();
+ let serving = tokio::spawn({
+ let host = host.clone();
+ async move {
+ host.serve_local(listener, tokio_util::sync::CancellationToken::new())
+ .await
}
- drop(listener);
});
- let (mut ui, connection, _, _) = native_ui(
+ let anchor = Client::connect_peer(
+ &path,
+ Some(host.epoch()),
+ f.source.clone(),
+ [support::identity(2)].into(),
+ selections,
+ )
+ .await
+ .unwrap();
+ let (mut ui, connection, _, _) = native_ui_at(
&f,
- Arc::new(maka_transport::transport::ipc::Connector::new(path, nonce).unwrap()),
+ Arc::new(maka_transport::transport::ipc::Connector::new(
+ anchor.call_channel_opener().unwrap(),
+ )),
+ anchor.catalogue().clone(),
)
.await;
let root = ui.mount_launch(serde_json::json!({})).unwrap();
@@ -702,9 +752,26 @@ async fn native_adapter_runs_checked_ui_actions_actual_caller_sql_and_original_r
stop_ui(ui, &root).await;
connection.close().await.unwrap();
drop(connection);
- stop.cancel();
- acceptance.await.unwrap();
- f.server.close().await.unwrap();
+ drop(anchor);
+ // Detaching the UI does not stop the independent business Host.
+ let mut observer = Client::connect(&path, Some(host.epoch())).await.unwrap();
+ let mut read = observer
+ .call(
+ "business",
+ "read".into(),
+ serde_json::Value::Null,
+ Default::default(),
+ )
+ .await
+ .unwrap();
+ read.start().await.unwrap();
+ assert!(
+ matches!(read.receive().await.unwrap(), Reply::Finished { result: Ok(ref value), cleanup: Ok(()) } if value["effects"] == 1)
+ );
+ observer.close().await.unwrap();
+ assert!(Client::retire(&path, host.epoch()).await.unwrap());
+ serving.await.unwrap().unwrap();
+ drop(host);
f.cleanup().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -1613,14 +1680,13 @@ async fn lost_close_before_dispatch_poll_keeps_the_actual_confirmed_sql_token()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn private_hello_accepts_bursts_and_close_cancels_partial_handshakes() {
+async fn application_close_cancels_bursts_of_partial_open_frames() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let f = Fixture::new().await;
- let nonce = Uuid::new_v4();
#[cfg(unix)]
let path = f._directory.path().join("hello.sock");
#[cfg(windows)]
- let path = std::path::PathBuf::from(format!(r"\\.\pipe\maka-hello-test-{nonce}"));
+ let path = std::path::PathBuf::from(format!(r"\\.\pipe\maka-open-test-{}", Uuid::new_v4()));
let mut listener = maka_host_runtime::ipc::LocalListener::bind(&path).unwrap();
let mut clients = Vec::new();
let mut observers = Vec::new();
@@ -1628,37 +1694,11 @@ async fn private_hello_accepts_bursts_and_close_cancels_partial_handshakes() {
let mut client = maka_host_runtime::ipc::connect(&path).await.unwrap();
let accepted = listener.accept().await.unwrap();
let (reader, writer) = tokio::io::split(accepted);
- observers.push(
- f.server
- .serve_private(reader, writer, f.peer.clone(), nonce)
- .unwrap(),
- );
+ observers.push(f.server.serve(reader, writer, f.peer.clone()).unwrap());
// Leave a real physical header incomplete. No Scope/Open is admitted.
client.write_all(&[0]).await.unwrap();
clients.push(client);
}
- // A foreign nonce still fails before Scope admission and without an Open.
- let mut foreign = maka_host_runtime::ipc::connect(&path).await.unwrap();
- let accepted = listener.accept().await.unwrap();
- let (reader, writer) = tokio::io::split(accepted);
- let observer = f
- .server
- .serve_private(reader, writer, f.peer.clone(), nonce)
- .unwrap();
- let hello = serde_json::to_vec(&serde_json::json!({"nonce":Uuid::new_v4()})).unwrap();
- foreign
- .write_all(&(hello.len() as u32).to_be_bytes())
- .await
- .unwrap();
- foreign.write_all(&hello).await.unwrap();
- assert!(
- tokio::time::timeout(std::time::Duration::from_secs(2), observer)
- .await
- .unwrap()
- .unwrap_err()
- .contains("different Host incarnation")
- );
- drop(foreign);
drop(listener);
tokio::time::timeout(std::time::Duration::from_secs(2), f.server.close())
.await
diff --git a/platform/transport/tests/public_host.rs b/platform/transport/tests/public_host.rs
index b3928e5fe0..f5ec08df41 100644
--- a/platform/transport/tests/public_host.rs
+++ b/platform/transport/tests/public_host.rs
@@ -242,17 +242,30 @@ impl PeerHost {
.commit_application(deployment, 0, input, vec![], packages)
.await
.unwrap();
- let selections = BTreeMap::from([(
- "business".into(),
- HostEndpoint {
- consumer: support::identity(13),
- import: "business".into(),
- grant: Grant {
- methods: ["read".into(), "write".into()].into(),
- recovery: false,
+ let selections = BTreeMap::from([
+ (
+ "business".into(),
+ HostEndpoint {
+ consumer: support::identity(13),
+ import: "business".into(),
+ grant: Grant {
+ methods: ["read".into(), "write".into()].into(),
+ recovery: false,
+ },
},
- },
- )]);
+ ),
+ (
+ "receipts".into(),
+ HostEndpoint {
+ consumer: support::identity(13),
+ import: "receipts".into(),
+ grant: Grant {
+ methods: ["read".into()].into(),
+ recovery: true,
+ },
+ },
+ ),
+ ]);
let host = Host::open(
store.clone(),
deployment,
@@ -308,6 +321,78 @@ impl PeerHost {
(store, source, consumer)
}
}
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn quit_keeps_other_clients_and_stops_after_the_last_client() {
+ tokio::time::timeout(std::time::Duration::from_secs(15), async {
+ let fixture = PeerHost::start().await;
+ let epoch = Some(fixture.host.epoch());
+ let mut detached = Client::connect(&fixture.address, epoch).await.unwrap();
+ detached.close().await.unwrap();
+ let mut first = Client::connect(&fixture.address, epoch).await.unwrap();
+ let mut second = Client::connect(&fixture.address, epoch).await.unwrap();
+ first.quit().await.unwrap();
+ let mut read = second
+ .call("business", "read".into(), Value::Null, Default::default())
+ .await
+ .unwrap();
+ read.start().await.unwrap();
+ assert!(matches!(read.receive().await.unwrap(), Reply::Finished { result: Ok(ref value), cleanup: Ok(()) } if value["effects"] == 0));
+ second.quit().await.unwrap();
+ fixture.serving.await.unwrap().unwrap();
+ drop(fixture.host);
+ drop(fixture.store);
+ drop(RootOwner::open(&fixture.directory.path().join("host"), &fixture.namespaces).unwrap());
+ })
+ .await
+ .unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn concurrent_quits_stop_the_host() {
+ tokio::time::timeout(std::time::Duration::from_secs(15), async {
+ let fixture = PeerHost::start().await;
+ let epoch = Some(fixture.host.epoch());
+ let mut first = Client::connect(&fixture.address, epoch).await.unwrap();
+ let mut second = Client::connect(&fixture.address, epoch).await.unwrap();
+ let (first, second) = tokio::join!(first.quit(), second.quit());
+ first.unwrap();
+ second.unwrap();
+ fixture.serving.await.unwrap().unwrap();
+ })
+ .await
+ .unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn reconnect_captures_the_replacement_host_registration() {
+ tokio::time::timeout(Duration::from_secs(20), async {
+ let fixture = PeerHost::start().await;
+ let mut original = Client::connect(&fixture.address, Some(fixture.host.epoch())).await.unwrap();
+ let before = original.catalogue().describe("business").unwrap().registration_epoch.unwrap();
+ let (mut input, packages) = support::host();
+ let provider = Uuid::new_v4();
+ input.instances.get_mut("business").unwrap().key = provider;
+ fixture.store.commit_application(fixture.host.source().deployment, 1, input, vec![], packages).await.unwrap();
+ let mut connected = loop {
+ let mut connected = Client::connect(&fixture.address, Some(fixture.host.epoch())).await.unwrap();
+ let observed = connected.catalogue().describe("business").unwrap();
+ if observed.registration_epoch.is_some_and(|epoch| epoch != before) {
+ assert_eq!(observed.provider.0, vec![provider]);
+ break connected;
+ }
+ connected.close().await.unwrap();
+ tokio::task::yield_now().await;
+ };
+ let mut read = connected.call("business", "read".into(), Value::Null, Default::default()).await.unwrap();
+ read.start().await.unwrap();
+ assert!(matches!(read.receive().await.unwrap(), Reply::Finished { result: Ok(ref value), cleanup: Ok(()) } if value["effects"] == 0));
+ original.close().await.unwrap();
+ connected.close().await.unwrap();
+ assert!(Client::retire(&fixture.address, fixture.host.epoch()).await.unwrap());
+ fixture.serving.await.unwrap().unwrap();
+ }).await.unwrap();
+}
+
async fn peer_call(
client: &Client,
source: &maka_plugin_core::deployment::DeploymentRef,
@@ -459,31 +544,55 @@ async fn cancelled_peer_control_never_appends_another_command_to_its_anchor() {
survivor.close().await.unwrap();
}
+ for quitting in [false, true] {
#[cfg(unix)]
- let address = fixture.directory.path().join("control.sock");
+ let address = fixture.directory.path().join(if quitting { "quit-control.sock" } else { "control.sock" });
#[cfg(windows)]
let address = std::path::PathBuf::from(format!(r"\\.\pipe\maka-peer-control-test-{}", Uuid::new_v4()));
let mut listener = LocalListener::bind(&address).unwrap();
let (received, command_received) = tokio::sync::oneshot::channel();
+ let (quit_received, independent_quit_received) = tokio::sync::oneshot::channel();
+ let (release_ack, ack_released) = tokio::sync::oneshot::channel();
+ let session = Uuid::new_v4();
+ let catalogue = catalogue.clone();
let epoch = fixture.host.epoch();
let connected_source = fixture.host.source().clone();
+ let connected_consumers = fixture.selections.iter().map(|(name, selection)| (name.clone(), selection.consumer.clone())).collect();
let fake = tokio::spawn(async move {
let stream = listener.accept().await.unwrap();
let mut reader = wire::Reader::new(stream);
assert!(matches!(reader.read::<Opening>().await.unwrap().request, Request::ConnectPeer { .. }));
wire::write(&mut reader.inner, &Response::Connected(Connected {
- version: VERSION, epoch, session: Uuid::new_v4(), source: connected_source,
- consumers: BTreeMap::from([("business".into(), support::identity(13))]), catalogue: catalogue.clone(),
+ version: VERSION, epoch, session, source: connected_source,
+ consumers: connected_consumers, catalogue: catalogue.clone(),
})).await.unwrap();
assert!(matches!(reader.read::<SessionControl>().await.unwrap(), SessionControl::Capture));
received.send(()).unwrap();
- // EOF proves no later control frame was appended after lost observation.
- assert!(reader.read::<SessionControl>().await.is_err());
- #[cfg(unix)]
- {
- // Unix retains the read half for original replies and final cleanup.
+ if quitting {
+ let stream = tokio::select! {
+ stream = listener.accept() => stream.unwrap(),
+ _ = reader.read::<SessionControl>() => panic!("quit appended to or closed the lost anchor"),
+ };
+ let mut control = wire::Reader::new(stream);
+ let opening = control.read::<Opening>().await.unwrap();
+ assert_eq!(opening.epoch, Some(epoch));
+ assert!(matches!(opening.request, Request::Quit { session: actual } if actual == session));
+ // Old anchor replies cannot acknowledge the independent quit.
wire::write(&mut reader.inner, &Response::Captured(Ok(catalogue))).await.unwrap();
wire::write(&mut reader.inner, &Response::Closed).await.unwrap();
+ quit_received.send(()).unwrap();
+ ack_released.await.unwrap();
+ wire::write(&mut control.inner, &Response::Closed).await.unwrap();
+ assert!(reader.read::<SessionControl>().await.is_err());
+ } else {
+ // EOF proves no later control frame was appended after lost observation.
+ assert!(reader.read::<SessionControl>().await.is_err());
+ #[cfg(unix)]
+ {
+ // Unix retains the read half for original replies and final cleanup.
+ wire::write(&mut reader.inner, &Response::Captured(Ok(catalogue))).await.unwrap();
+ wire::write(&mut reader.inner, &Response::Closed).await.unwrap();
+ }
}
});
let mut client = Client::connect_peer(&address, Some(epoch), source.clone(), [consumer.clone()].into(), fixture.selections.clone()).await.unwrap();
@@ -495,23 +604,438 @@ async fn cancelled_peer_control_never_appends_another_command_to_its_anchor() {
drop(capture);
assert!(client.epochs().await.is_err());
assert!(client.capture().await.is_err());
- assert!(client.approve(source, [consumer].into(), Default::default(), None).await.is_err());
+ assert!(client.approve(source.clone(), [consumer.clone()].into(), Default::default(), None).await.is_err());
assert!(client.open_call_channel().await.is_err());
- #[cfg(unix)]
- {
- client.close().await.unwrap();
+ if quitting {
+ let mut quit = Box::pin(client.quit());
+ tokio::select! {
+ result = &mut quit => panic!("quit completed before its independent acknowledgement: {result:?}"),
+ result = independent_quit_received => result.unwrap(),
+ }
+ assert!(futures_util::poll!(quit.as_mut()).is_pending(), "old anchor replies acknowledged the independent quit");
+ // Only this independently gated Closed response confirms cleanup.
+ release_ack.send(()).unwrap();
+ quit.await.unwrap();
+ client.quit().await.unwrap();
client.close().await.unwrap();
- }
- #[cfg(windows)]
- {
- assert!(client.close().await.is_err());
- assert!(client.close().await.is_err());
+ } else {
+ #[cfg(unix)]
+ {
+ client.close().await.unwrap();
+ client.close().await.unwrap();
+ }
+ #[cfg(windows)]
+ {
+ assert!(client.close().await.is_err());
+ assert!(client.close().await.is_err());
+ }
}
assert!(client.open_call_channel().await.is_err());
drop(client);
fake.await.unwrap();
+ }
assert!(Client::retire(&fixture.address, fixture.host.epoch()).await.unwrap());
fixture.serving.await.unwrap().unwrap();
ui.shutdown().await.unwrap();
}).await.unwrap();
}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn discovered_connections_release_ui_roots_without_stopping_the_shared_host() {
+ use maka_plugin_core::composition::PresentationTarget;
+ use maka_transport::{Configuration, Connections, HostLocation};
+ tokio::time::timeout(Duration::from_secs(30), async {
+ let fixture = PeerHost::start().await;
+ let registration = fixture.host.publish_registration(&fixture.address, None).unwrap();
+ let (ui, source, _) = fixture.ui("tui-connections", PresentationTarget::Tui).await;
+ let configuration = || Configuration { hosts: vec![maka_transport::Host {
+ name: "local".into(),
+ location: HostLocation::Public { root: fixture.directory.path().join("host"), state_domain: Some(fixture.host.root_id().into()) },
+ deployment: fixture.host.source().deployment,
+ endpoints: fixture.selections.clone(), native: vec![],
+ }] };
+ for invalid in 0..3 {
+ let mut input = configuration();
+ match invalid {
+ 0 => input.hosts[0].deployment = Uuid::new_v4(),
+ 1 => input.hosts[0].location = HostLocation::Public { root: fixture.directory.path().join("host"), state_domain: Some("0".repeat(64)) },
+ _ => input.hosts[0].location = HostLocation::Public { root: fixture.directory.path().join("never-create"), state_domain: None },
+ }
+ let mut connections = Connections::prepare(ui.clone(), source.deployment, PresentationTarget::Tui, input, fixture.namespaces.clone(), std::path::PathBuf::from("unused-local-startup-helper"), NativeRegistry::default(), Some(maka_host_js::factory(Default::default()))).await.unwrap();
+ assert!(connections.start().await.is_err());
+ connections.close().await.unwrap();
+ }
+ assert!(!fixture.directory.path().join("never-create").exists());
+ let mut connections = Connections::prepare(ui.clone(), source.deployment, PresentationTarget::Tui, configuration(), fixture.namespaces.clone(), std::path::PathBuf::from("unused-local-startup-helper"), NativeRegistry::default(), Some(maka_host_js::factory(Default::default()))).await.unwrap();
+ connections.start().await.unwrap();
+ let frozen = connections.registrations();
+ assert_eq!(frozen.len(), 2);
+ let (desktop, desktop_source, _) = fixture.ui("desktop-connections", PresentationTarget::Desktop).await;
+ let mut other = Connections::prepare(desktop.clone(), desktop_source.deployment, PresentationTarget::Desktop, configuration(), fixture.namespaces.clone(), std::path::PathBuf::from("unused-local-startup-helper"), NativeRegistry::default(), Some(maka_host_js::factory(Default::default()))).await.unwrap();
+ other.start().await.unwrap();
+ assert_eq!(fixture.host.status().sessions, 2);
+ connections.close().await.unwrap();
+ ui.shutdown().await.unwrap();
+ drop(ui);
+ drop(RootOwner::open(&fixture.directory.path().join("tui-connections"), &fixture.namespaces).unwrap());
+ while fixture.host.status().sessions != 1 {
+ tokio::task::yield_now().await;
+ }
+ assert!(!fixture.host.status().closing);
+ // Retained registrations carry no anchor lease after the owning UI closed.
+ assert_eq!(frozen.len(), 2);
+ assert_eq!(other.approve_source(&desktop, desktop_source.clone(), Default::default()).await.unwrap().len(), 2);
+ let mut observer = Client::connect(&fixture.address, Some(fixture.host.epoch())).await.unwrap();
+ let mut read = observer.call("business", "read".into(), Value::Null, Default::default()).await.unwrap();
+ read.start().await.unwrap();
+ assert!(matches!(read.receive().await.unwrap(), Reply::Finished { result: Ok(ref value), cleanup: Ok(()) } if value["effects"] == 0));
+ observer.close().await.unwrap();
+ other.close().await.unwrap();
+ desktop.shutdown().await.unwrap();
+ drop(desktop);
+ drop(RootOwner::open(&fixture.directory.path().join("desktop-connections"), &fixture.namespaces).unwrap());
+ assert!(Client::retire(&fixture.address, fixture.host.epoch()).await.unwrap());
+ fixture.serving.await.unwrap().unwrap();
+ registration.remove().unwrap();
+ }).await.unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn quitting_an_externally_stopped_host_requires_its_original_root_release() {
+ use maka_plugin_core::composition::PresentationTarget;
+ use maka_transport::{Configuration, Connections, ExitMode, HostLocation};
+ tokio::time::timeout(Duration::from_secs(30), async {
+ for replacement in [false, true] {
+ let fixture = PeerHost::start().await;
+ let registration = fixture.host.publish_registration(&fixture.address, None).unwrap();
+ let (ui, source, _) = fixture.ui("external-stop-ui", PresentationTarget::Tui).await;
+ let ui_root = fixture.directory.path().join("external-stop-ui");
+ let host_root = fixture.directory.path().join("host");
+ let original_epoch = fixture.host.epoch();
+ let deployment = fixture.host.source().deployment;
+ let configuration = Configuration { hosts: vec![maka_transport::Host {
+ name: "local".into(),
+ location: HostLocation::Public { root: host_root.clone(), state_domain: Some(fixture.host.root_id().into()) },
+ deployment: fixture.host.source().deployment,
+ endpoints: fixture.selections.clone(), native: vec![],
+ }] };
+ let mut connections = Connections::prepare(ui.clone(), source.deployment, PresentationTarget::Tui,
+ configuration, fixture.namespaces.clone(), std::path::PathBuf::from("unused-local-startup-helper"),
+ NativeRegistry::default(), Some(maka_host_js::factory(Default::default()))).await.unwrap();
+ connections.start().await.unwrap();
+ assert_eq!(fixture.host.status().sessions, 1);
+ assert!(Client::retire_with_options(&fixture.address, fixture.host.epoch(), true, true).await.unwrap());
+ fixture.serving.await.unwrap().unwrap();
+ assert!(fixture.host.status().closed);
+ // SQL cleanup and the closed socket do not release retained Root owners.
+ ui.shutdown().await.unwrap();
+ drop(ui);
+ assert_eq!(RootOwner::open(&host_root, &fixture.namespaces).err().unwrap().kind(), std::io::ErrorKind::WouldBlock);
+ assert!(connections.close_with(ExitMode::Quit).await.is_err());
+ assert_eq!(RootOwner::open(&ui_root, &fixture.namespaces).err().unwrap().kind(), std::io::ErrorKind::WouldBlock);
+ // Releasing the actual original owners makes the same close retry safe.
+ registration.remove().unwrap();
+ drop((fixture.host, fixture.store));
+ drop(RootOwner::open(&host_root, &fixture.namespaces).unwrap());
+ let replacement = if replacement {
+ let store = Arc::new(EventLog::for_root(Arc::new(RootOwner::open(&host_root, &fixture.namespaces).unwrap())).await.unwrap());
+ let host = Host::open(store, deployment, NativeRegistry::default(),
+ Some(maka_host_js::factory(Default::default())), fixture.selections.clone(), vec![]).await.unwrap();
+ assert_ne!(host.epoch(), original_epoch);
+ #[cfg(unix)]
+ let address = fixture.address.clone();
+ #[cfg(windows)]
+ let address = std::path::PathBuf::from(format!(r"\\.\pipe\maka-replacement-test-{}", Uuid::new_v4()));
+ let listener = LocalListener::bind(&address).unwrap();
+ let registration = host.publish_registration(&address, None).unwrap();
+ let serving = tokio::spawn({ let host = host.clone(); async move { host.serve_local(listener, CancellationToken::new()).await } });
+ Some((host, address, registration, serving))
+ } else {
+ None
+ };
+ connections.close_with(ExitMode::Quit).await.unwrap();
+ drop(RootOwner::open(&ui_root, &fixture.namespaces).unwrap());
+ if let Some((host, address, registration, serving)) = replacement {
+ // The old UI's quit may confirm its original owner disappeared, but
+ // cannot retire the new incarnation now holding the same Root.
+ let mut client = Client::connect(&address, Some(host.epoch())).await.unwrap();
+ let mut read = client.call("business", "read".into(), Value::Null, Default::default()).await.unwrap();
+ read.start().await.unwrap();
+ assert!(matches!(read.receive().await.unwrap(), Reply::Finished { result: Ok(ref value), cleanup: Ok(()) } if value["effects"] == 0));
+ assert!(!Client::status(&address, Some(host.epoch())).await.unwrap().closing);
+ client.close().await.unwrap();
+ assert!(Client::retire(&address, host.epoch()).await.unwrap());
+ serving.await.unwrap().unwrap();
+ registration.remove().unwrap();
+ drop((read, client, host));
+ drop(RootOwner::open(&host_root, &fixture.namespaces).unwrap());
+ }
+ }
+ }).await.unwrap();
+}
+
+struct NativeCaller;
+impl maka_host_runtime::application::NativeOwner for NativeCaller {
+ fn prepare_retirement(
+ &self,
+ ) -> Option<Box<dyn maka_host_runtime::application::NativeRetirement>> {
+ Some(Box::new(()))
+ }
+ fn close(&self) -> futures_util::future::BoxFuture<'static, Result<(), String>> {
+ Box::pin(async { Ok(()) })
+ }
+}
+fn native_caller() -> (
+ NativeRegistry,
+ tokio::sync::mpsc::UnboundedReceiver<maka_host_runtime::application::NativeServices>,
+) {
+ let (send, receive) = tokio::sync::mpsc::unbounded_channel();
+ let mut native = NativeRegistry::default();
+ native
+ .insert(
+ "test.native-caller".into(),
+ Arc::new(move |context, _| {
+ send.send(context.services()).unwrap();
+ Box::pin(async {
+ maka_host_runtime::application::NativeActivation {
+ exports: Ok(BTreeMap::new()),
+ owner: Arc::new(NativeCaller),
+ }
+ })
+ }),
+ )
+ .unwrap();
+ (native, receive)
+}
+async fn native_read(
+ services: &maka_host_runtime::application::NativeServices,
+) -> Result<Value, String> {
+ let request = services
+ .request(
+ maka_capabilities::call::Identity::Remote {
+ request_id: Uuid::new_v4(),
+ },
+ CancellationToken::new(),
+ )
+ .unwrap();
+ let result = match request.call(
+ "business",
+ 0,
+ "read".into(),
+ Value::Null,
+ Default::default(),
+ ) {
+ Ok(call) => call.await,
+ Err(error) => Err(error),
+ };
+ request.finish().await.unwrap();
+ result
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn configured_native_grants_apply_live_survive_source_update_and_root_restart() {
+ use maka_host_runtime::application::{host::Configuration, transport::NativeGrant};
+ tokio::time::timeout(Duration::from_secs(30), async {
+ let directory = tempfile::tempdir().unwrap();
+ maka_event_log::root::private_directory(directory.path()).unwrap();
+ let namespaces = RootNamespaces {
+ ownership: directory.path().join("owners"),
+ control: directory.path().join("control"),
+ };
+ let root = directory.path().join("host");
+ let store = Arc::new(
+ EventLog::for_root(Arc::new(RootOwner::create(&root, &namespaces).unwrap()))
+ .await
+ .unwrap(),
+ );
+ let deployment = Uuid::new_v4();
+ let (mut input, mut packages) = support::host();
+ let gateway = packages
+ .iter_mut()
+ .find(|package| package.manifest().id == "local.gateway")
+ .unwrap();
+ let mut declaration = gateway.manifest().application.clone().unwrap();
+ declaration
+ .entries
+ .get_mut(&maka_plugin_core::package::EntryTarget::Host)
+ .unwrap()
+ .implementation = maka_plugin_core::package::Implementation::Native {
+ implementation: "test.native-caller".into(),
+ };
+ *gateway = support::counter::package("local.gateway", declaration, None);
+ input.instances.get_mut("gateway").unwrap().package = gateway.application_ref().into();
+ let source = store
+ .commit_application(deployment, 0, input.clone(), vec![], packages.clone())
+ .await
+ .unwrap();
+ let endpoints = BTreeMap::from([(
+ "business".into(),
+ HostEndpoint {
+ consumer: support::identity(13),
+ import: "business".into(),
+ grant: Grant {
+ methods: ["read".into()].into(),
+ recovery: false,
+ },
+ },
+ )]);
+ let previous = Configuration::PublicHost {
+ endpoints: endpoints.clone(),
+ native: vec![],
+ };
+ let desired = Configuration::PublicHost {
+ endpoints: endpoints.clone(),
+ native: vec![NativeGrant {
+ origin: support::identity(13),
+ consumer: support::identity(13),
+ provider: support::identity(11),
+ export: "Counter".into(),
+ methods: ["read".into()].into(),
+ recovery: false,
+ }],
+ };
+ store
+ .select_committed_application_launch(
+ source.clone(),
+ serde_json::to_value(&previous).unwrap(),
+ )
+ .await
+ .unwrap();
+ let (native, mut activations) = native_caller();
+ let host = Host::open(
+ store.clone(),
+ deployment,
+ native,
+ Some(maka_host_js::factory(Default::default())),
+ endpoints,
+ vec![],
+ )
+ .await
+ .unwrap();
+ #[cfg(unix)]
+ let address = directory.path().join("before.sock");
+ #[cfg(windows)]
+ let address =
+ std::path::PathBuf::from(format!(r"\\.\pipe\maka-native-before-{}", Uuid::new_v4()));
+ let listener = LocalListener::bind(&address).unwrap();
+ let serving = tokio::spawn({
+ let host = host.clone();
+ async move { host.serve_local(listener, CancellationToken::new()).await }
+ });
+ let mut client = Client::connect(&address, Some(host.epoch())).await.unwrap();
+ let services = activations.recv().await.unwrap();
+ assert_eq!(
+ native_read(&services).await.unwrap_err(),
+ "native service invocation is not granted"
+ );
+ // A stale source or unrelated previous selection must not publish grants.
+ let mut future = source.clone();
+ future.revision += 1;
+ assert!(
+ client
+ .configure(future, previous.clone(), desired.clone())
+ .await
+ .is_err()
+ );
+ assert!(
+ client
+ .configure(source.clone(), desired.clone(), desired.clone())
+ .await
+ .is_err()
+ );
+ assert_eq!(
+ store
+ .application_launch()
+ .await
+ .unwrap()
+ .unwrap()
+ .configuration,
+ serde_json::to_value(&previous).unwrap()
+ );
+ assert_eq!(
+ native_read(&services).await.unwrap_err(),
+ "native service invocation is not granted"
+ );
+ client
+ .configure(source.clone(), previous.clone(), desired.clone())
+ .await
+ .unwrap();
+ // Retry the exact command after losing its acknowledgement.
+ client
+ .configure(source.clone(), previous.clone(), desired.clone())
+ .await
+ .unwrap();
+ assert_eq!(
+ native_read(&services).await.unwrap(),
+ serde_json::json!({"effects":0})
+ );
+ let next = store
+ .commit_application(deployment, 1, input, vec![], packages)
+ .await
+ .unwrap();
+ assert_eq!(next.revision, 2);
+ assert!(
+ client
+ .configure(source, desired.clone(), previous)
+ .await
+ .is_err()
+ );
+ assert_eq!(
+ native_read(&services).await.unwrap(),
+ serde_json::json!({"effects":0})
+ );
+ let saved = store.application_launch().await.unwrap().unwrap();
+ assert_eq!(saved.configuration, serde_json::to_value(&desired).unwrap());
+ assert!(saved.pending.is_none());
+ client.close().await.unwrap();
+ assert!(Client::retire(&address, host.epoch()).await.unwrap());
+ serving.await.unwrap().unwrap();
+ drop((client, services, host, store));
+ // Reopen from the durable launcher selection, as a new Host process does.
+ let store = Arc::new(
+ EventLog::for_root(Arc::new(RootOwner::open(&root, &namespaces).unwrap()))
+ .await
+ .unwrap(),
+ );
+ let saved = store.application_launch().await.unwrap().unwrap();
+ let Configuration::PublicHost {
+ endpoints,
+ native: grants,
+ } = serde_json::from_value(saved.configuration).unwrap();
+ let (native, mut activations) = native_caller();
+ let host = Host::open(
+ store.clone(),
+ deployment,
+ native,
+ Some(maka_host_js::factory(Default::default())),
+ endpoints,
+ grants,
+ )
+ .await
+ .unwrap();
+ #[cfg(unix)]
+ let address = directory.path().join("after.sock");
+ #[cfg(windows)]
+ let address =
+ std::path::PathBuf::from(format!(r"\\.\pipe\maka-native-after-{}", Uuid::new_v4()));
+ let listener = LocalListener::bind(&address).unwrap();
+ let serving = tokio::spawn({
+ let host = host.clone();
+ async move { host.serve_local(listener, CancellationToken::new()).await }
+ });
+ let mut client = Client::connect(&address, Some(host.epoch())).await.unwrap();
+ assert_eq!(host.source(), &next);
+ let services = activations.recv().await.unwrap();
+ assert_eq!(
+ native_read(&services).await.unwrap(),
+ serde_json::json!({"effects":0})
+ );
+ client.close().await.unwrap();
+ assert!(Client::retire(&address, host.epoch()).await.unwrap());
+ serving.await.unwrap().unwrap();
+ drop((client, services, host, store));
+ drop(RootOwner::open(&root, &namespaces).unwrap());
+ })
+ .await
+ .unwrap();
+}
diff --git a/plugins/shared/tui-messages.ts b/plugins/shared/tui-messages.ts
index 484afdc053..d8bf50b02b 100644
--- a/plugins/shared/tui-messages.ts
+++ b/plugins/shared/tui-messages.ts
@@ -522,9 +522,9 @@ export const tui_language_auto: Message = [
];
export const tui_help: Message = [
- 'Tab / Shift+Tab Move focus between controls\nUp / Down Select a navigation, page or command item\nEnter Activate the focused item\nEnter Send from composer / queue for next turn\nShift+Enter New line in composer\nCtrl+J New line in composer (all terminals)\nCtrl+Enter Submit answers / run message revision\nCtrl+K / Ctrl+P Open command palette\nCtrl+F Find in conversation\nCtrl+N New session\n [...]
- 'Tab / Shift+Tab 在控件间移动焦点\n上 / 下 选择导航、页面或命令项\nEnter 激活焦点项\nEnter 发送消息 / 运行中排入下一轮\nShift+Enter 输入框内换行\nCtrl+J 消息输入框换行（所有终端）\nCtrl+Enter 提交问答 / 运行消息修订\nCtrl+K / Ctrl+P 打开命令面板\nCtrl+F 查找对话内容\nCtrl+N 新建会话\nCtrl+A / Ctrl+Z 编辑器内全选 / 撤销\nF4 检查产物与执行证据\nCtrl+G 查看待审批请求或问题\nCtrl+B 展开 / 折叠导航\nCtrl+PgUp/PgDn 切换会话页签\nCtrl+W 关闭页签（保留草稿与任务）\nF11 [...]
- 'Tab / Shift+Tab 在控制項間移動焦點\n上 / 下 選擇導覽、頁面或命令項\nEnter 啟用焦點項\nEnter 傳送訊息 / 執行中排入下一輪\nShift+Enter 輸入框內換行\nCtrl+J 訊息輸入框換行（所有終端機）\nCtrl+Enter 提交問答 / 執行訊息修訂\nCtrl+K / Ctrl+P 開啟命令面板\nCtrl+F 尋找對話內容\nCtrl+N 建立對話\nCtrl+A / Ctrl+Z 編輯器內全選 / 復原\nF4 檢查產物與執行證據\nCtrl+G 查看待審批請求或問題\nCtrl+B 展開 / 收合導覽\nCtrl+PgUp/PgDn 切換對話分頁\nCtrl+W 關閉分頁（保留草稿與工作）\nF11 [...]
+ 'Tab / Shift+Tab Move focus between controls\nUp / Down Select a navigation, page or command item\nEnter Activate the focused item\nEnter Send from composer / queue for next turn\nShift+Enter New line in composer\nCtrl+J New line in composer (all terminals)\nCtrl+Enter Submit answers / run message revision\nCtrl+K / Ctrl+P Open command palette\nCtrl+F Find in conversation\nCtrl+N New session\n [...]
+ 'Tab / Shift+Tab 在控件间移动焦点\n上 / 下 选择导航、页面或命令项\nEnter 激活焦点项\nEnter 发送消息 / 运行中排入下一轮\nShift+Enter 输入框内换行\nCtrl+J 消息输入框换行（所有终端）\nCtrl+Enter 提交问答 / 运行消息修订\nCtrl+K / Ctrl+P 打开命令面板\nCtrl+F 查找对话内容\nCtrl+N 新建会话\nCtrl+A / Ctrl+Z 编辑器内全选 / 撤销\nF4 检查产物与执行证据\nCtrl+G 查看待审批请求或问题\nCtrl+B 展开 / 折叠导航\nCtrl+PgUp/PgDn 切换会话页签\nCtrl+W 关闭页签（保留草稿与任务）\nF11 [...]
+ 'Tab / Shift+Tab 在控制項間移動焦點\n上 / 下 選擇導覽、頁面或命令項\nEnter 啟用焦點項\nEnter 傳送訊息 / 執行中排入下一輪\nShift+Enter 輸入框內換行\nCtrl+J 訊息輸入框換行（所有終端機）\nCtrl+Enter 提交問答 / 執行訊息修訂\nCtrl+K / Ctrl+P 開啟命令面板\nCtrl+F 尋找對話內容\nCtrl+N 建立對話\nCtrl+A / Ctrl+Z 編輯器內全選 / 復原\nF4 檢查產物與執行證據\nCtrl+G 查看待審批請求或問題\nCtrl+B 展開 / 收合導覽\nCtrl+PgUp/PgDn 切換對話分頁\nCtrl+W 關閉分頁（保留草稿與工作）\nF11 [...]
];
export const tui_command_agent_screen: Message = [
'Open Agent screen',
diff --git a/products/maka/bundle/src/composition.ts b/products/maka/bundle/src/composition.ts
index ff2f36d87a..3c1e70e75b 100644
--- a/products/maka/bundle/src/composition.ts
+++ b/products/maka/bundle/src/composition.ts
@@ -580,7 +580,7 @@ function settingsComposition(
},
{
name: 'local',
- location: { kind: 'process', root: configuration.hostRoot },
+ location: { kind: 'public', root: configuration.hostRoot },
deployment: configuration.hostDeployment,
endpoints: {
management: expose(
diff --git a/products/maka/cli/Cargo.toml b/products/maka/cli/Cargo.toml
index 3b0d7789ef..1353376f3d 100644
--- a/products/maka/cli/Cargo.toml
+++ b/products/maka/cli/Cargo.toml
@@ -35,6 +35,7 @@ path = "src/service_main.rs"
test = false
[dependencies]
+maka-capabilities.workspace = true
maka-agent-facts.workspace = true
maka-packages.workspace = true
maka-product-native.workspace = true
@@ -74,7 +75,6 @@ tempfile.workspace = true
[dev-dependencies]
unicode-width.workspace = true
-maka-capabilities.workspace = true
schemars.workspace = true
maka-agent-ports.workspace = true
agent-client-protocol.workspace = true
diff --git a/products/maka/cli/README.md b/products/maka/cli/README.md
index 3965e4f149..e23da7c9fc 100644
--- a/products/maka/cli/README.md
+++ b/products/maka/cli/README.md
@@ -29,9 +29,11 @@ From the repository root, run `just run` or `just run --help`. Development build
The first installation asks for workspace trust inside the TUI, before starting the Host or enabling project skills and tools. Tab or the arrow keys select an action; Enter confirms and Esc exits. The initial selection is Exit. Confirmation keeps the same terminal screen while the product initializes. Reopening uses the saved installation without asking again.
+Startup discovers the local Host and connects to it, or starts one automatically. TUI and Desktop can share that Host. By default, quitting a UI also stops the Host when no other client remains. Use Ctrl+Alt+Q in the TUI to close only the UI and keep accepted background tasks running; reconnect to observe their results.
+
## Product recovery and URL installation
-Close the product UI and its local Host before changing an owned Root from the CLI. Both commands default to `~/.maka/application`; select another installation with `--root DIRECTORY`.
+Close the product UI before recovery; restoration updates the independent Host through its management service. Direct package installation requires the selected Root to be inactive. Both commands default to `~/.maka/application`; select another installation with `--root DIRECTORY`.
```sh
maka application restore-defaults
diff --git a/products/maka/cli/README.zh-CN.md b/products/maka/cli/README.zh-CN.md
index 1b7c2f78ab..6e0e2e67b2 100644
--- a/products/maka/cli/README.zh-CN.md
+++ b/products/maka/cli/README.zh-CN.md
@@ -29,9 +29,11 @@
首次安装会在 TUI 内确认工作目录信任，然后才启动 Host、启用项目技能与工具。Tab 或方向键选择，Enter 确认，Esc 退出；默认选中退出。确认后在同一终端画面中初始化并进入产品。再次打开使用已保存的安装，不重复询问。
+启动时自动发现并连接本机 Host；不存在则自动启动。TUI 与 Desktop 可以共享同一个 Host。默认退出 UI 时，若没有其他客户端连接，Host 也会退出。TUI 中可用 Ctrl+Alt+Q 仅关闭界面，让已接受的后台任务继续运行，重连后观察结果。
+
## 产品恢复与 URL 安装
-通过 CLI 修改已有 Root 前，先关闭该产品的 UI 和本地 Host。两个命令默认使用 `~/.maka/application`；可用 `--root DIRECTORY` 选择其他安装。
+恢复前先关闭产品 UI；恢复通过管理服务更新独立 Host。直接安装插件包时，所选 Root 必须处于未运行状态。两个命令默认使用 `~/.maka/application`；可用 `--root DIRECTORY` 选择其他安装。
```sh
maka application restore-defaults
diff --git a/products/maka/cli/src/application.rs b/products/maka/cli/src/application.rs
index 2ce043fc8e..e1fef12a3a 100644
--- a/products/maka/cli/src/application.rs
+++ b/products/maka/cli/src/application.rs
@@ -21,7 +21,7 @@ mod install;
mod product;
mod serve;
-pub(crate) use product::{Launch, Roots, open_host};
+pub(crate) use product::{Launch, Roots, open_host, select_committed_host};
use std::{
io::Read,
@@ -66,7 +66,7 @@ pub(super) enum Command {
Commit(Commit),
/// Read and validate the persisted source and every selected package.
Read(ReadSource),
- /// Own a shared Host with inherited bootstrap/control and private IPC calls.
+ /// Serve an existing deployment as an independent public Host.
Serve(serve::Serve),
}
@@ -210,7 +210,7 @@ impl Command {
.transpose()?
.unwrap_or_default();
let store = Arc::new(open_store(args.root).await?);
- run_tui(store, args.deployment, configuration, None).await?;
+ run_tui(store, args.deployment, configuration, None, Vec::new()).await?;
}
}
Ok(())
@@ -291,20 +291,18 @@ async fn run_tui(
deployment: uuid::Uuid,
configuration: maka_transport::Configuration,
startup: Option<maka_render_terminal::startup::Startup>,
+ anchors: Vec<maka_host_runtime::application::host::Client>,
) -> Result<(), HostError> {
// Keep the foreground VM on one dedicated blocking owner, with
// no deeply nested CLI async stack under V8 initialization.
let scheduler = tokio::runtime::Handle::current();
let setup = (|| -> Result<_, HostError> {
let namespaces = maka_event_log::root::RootNamespaces::for_current_account()?;
- let executable = maka_transport::HostExecutable::new(
- std::env::current_exe()?,
- vec!["application".into(), "serve".into()],
- )?;
let cohost_native = serve::native()?;
- Ok((namespaces, executable, cohost_native))
+ let helper = std::env::current_exe()?;
+ Ok((namespaces, helper, cohost_native))
})();
- let (namespaces, executable, cohost_native) = match setup {
+ let (namespaces, helper, cohost_native) = match setup {
Ok(value) => value,
Err(error) => {
store.drain().await;
@@ -317,7 +315,7 @@ async fn run_tui(
PresentationTarget::Tui,
configuration,
namespaces,
- executable,
+ helper,
cohost_native,
Some(maka_host_js::factory(Default::default())),
)
@@ -334,6 +332,8 @@ async fn run_tui(
store.drain().await;
return Err(error.into());
}
+ // The actual UI peer now keeps the independently started Host resident.
+ drop(anchors);
let startup_errors = connections.startup_errors().to_vec();
let source = connections.source().clone();
let loaded = connections.take_initial().expect("initial UI snapshot");
@@ -354,30 +354,35 @@ async fn run_tui(
}
};
let terminal_errors = startup_errors.clone();
- let result = tokio::task::spawn_blocking(move || -> Result<(), HostError> {
- maka_render_terminal::application::run(
- foreground_store,
- loaded,
- registrations,
- scheduler,
- updates,
- maka_ui_js::execution(Default::default()),
- startup,
- terminal_errors,
- )
- .map_err(Into::into)
- })
- .await;
+ let result =
+ tokio::task::spawn_blocking(move || -> Result<maka_transport::ExitMode, HostError> {
+ maka_render_terminal::application::run(
+ foreground_store,
+ loaded,
+ registrations,
+ scheduler,
+ updates,
+ maka_ui_js::execution(Default::default()),
+ startup,
+ terminal_errors,
+ )
+ .map_err(Into::into)
+ })
+ .await;
// A panic is not a terminal/module cleanup receipt. Preserve
// both actual owners if the foreground execution was lost.
if let Err(error) = &result {
eprintln!("terminal execution cleanup unconfirmed: {error}");
std::future::pending::<()>().await;
}
- connections.lock().await.drain().await;
+ let result = result?;
+ match &result {
+ Ok(mode) => connections.lock().await.drain_with(*mode).await,
+ Err(_) => connections.lock().await.drain().await,
+ }
store.drain().await;
- match result? {
- Ok(()) => Ok(()),
+ match result {
+ Ok(_) => Ok(()),
Err(error)
if startup_errors.is_empty()
|| startup_errors
diff --git a/products/maka/cli/src/application/product.rs b/products/maka/cli/src/application/product.rs
index 68fe83fb7a..0684fe19b2 100644
--- a/products/maka/cli/src/application/product.rs
+++ b/products/maka/cli/src/application/product.rs
@@ -20,15 +20,19 @@
use super::{HostError, Input, Package, RemoteObservation, read_json};
use clap::{Args, ValueEnum};
use maka_js_runtime::plugin::Vm;
+use maka_plugin_core::plan::Plan;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{collections::BTreeMap, io::Write, path::PathBuf};
mod headless;
mod launch;
+mod public;
mod restore;
+#[cfg(test)]
+mod test_public;
mod update;
-pub(crate) use headless::{Roots, open_host};
+pub(crate) use headless::{Roots, open_host, select_committed_host};
pub(crate) use launch::Launch;
pub(super) use restore::Restore;
pub(super) use update::Update;
@@ -86,7 +90,7 @@ impl Compose {
let packages = tokio::task::spawn_blocking(packages).await??;
let authoring = authoring()?;
verified_fingerprint(&authoring, &packages, &version)?;
- let (sources, installation) =
+ let (sources, installation, _) =
author_with_source(self.product, configuration, &packages, &authoring).await?;
tokio::task::spawn_blocking(move || export(self.output, sources, installation, packages))
.await??;
@@ -229,7 +233,7 @@ async fn author(
product: Product,
configuration: Value,
packages: &BTreeMap<String, Package>,
-) -> Result<(Sources, Value), HostError> {
+) -> Result<(Sources, Value, Plan), HostError> {
let source = authoring()?;
verified_fingerprint(&source, packages, &bundle_fingerprint()?)?;
author_with_source(product, configuration, packages, &source).await
@@ -239,7 +243,7 @@ async fn author_with_source(
mut configuration: Value,
packages: &BTreeMap<String, Package>,
source: &str,
-) -> Result<(Sources, Value), HostError> {
+) -> Result<(Sources, Value, Plan), HostError> {
let values = configuration
.as_object_mut()
.ok_or("installation configuration must be an object")?;
@@ -298,18 +302,22 @@ async fn author_with_source(
vm.shutdown().await;
let sources: Sources = serde_json::from_value(result?)?;
let checked: Vec<_> = packages.values().cloned().collect();
- for (input, remote) in [
- (&sources.host, &[][..]),
- (&sources.management_host, &[][..]),
- (&sources.ui, sources.remote_exports.as_slice()),
- ] {
- maka_event_log::applications::validate_application(input, remote, &checked)?;
- }
+ let host_plan = maka_event_log::applications::validate_applications(
+ &[
+ (&sources.host, &[][..]),
+ (&sources.management_host, &[][..]),
+ (&sources.ui, sources.remote_exports.as_slice()),
+ ],
+ &checked,
+ )?
+ .into_iter()
+ .next()
+ .expect("validated Host plan");
let connections: maka_transport::Configuration =
serde_json::from_value(sources.connections.clone())?;
connections.validate().map_err(HostError::from)?;
values.remove("packages");
- Ok((sources, configuration))
+ Ok((sources, configuration, host_plan))
}
fn export(
diff --git a/products/maka/cli/src/application/product/headless.rs b/products/maka/cli/src/application/product/headless.rs
index 76ac989089..50ead7b80c 100644
--- a/products/maka/cli/src/application/product/headless.rs
+++ b/products/maka/cli/src/application/product/headless.rs
@@ -17,7 +17,7 @@
* under the License.
*/
//! A public Host opens the same persisted product source across process restarts.
-use super::{HostError, Input, Product, author, installation, launch, packages, selected};
+use super::{HostError, Input, Product, author, installation, launch, packages};
use base64::{Engine, engine::general_purpose::STANDARD};
use maka_event_log::{EventLog, applications::launch::Launch, root::RootOwner};
use maka_host_runtime::{
@@ -48,14 +48,7 @@ fn directories(roots: Option<Vec<DirectoryRootSpec>>) -> Result<Vec<DirectoryRoo
})
}
-#[derive(Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
-enum Selection {
- PublicHost {
- endpoints: BTreeMap<String, HostEndpoint>,
- native: Vec<NativeGrant>,
- },
-}
+use maka_host_runtime::application::host::Configuration as Selection;
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Preparation {
@@ -63,6 +56,86 @@ struct Preparation {
packages: Vec<BTreeMap<String, String>>,
}
+/// Persist the exact UI-authored business Host selection before any public
+/// process starts. Resumption checks the original selection and revision one.
+pub(super) async fn save_product(
+ store: &EventLog,
+ deployment: Uuid,
+ input: &Input,
+ packages: &[super::Package],
+ endpoints: BTreeMap<String, HostEndpoint>,
+ native: Vec<NativeGrant>,
+) -> Result<(), HostError> {
+ let selection = serde_json::to_value(Selection::PublicHost { endpoints, native })?;
+ match store.application_launch().await? {
+ Some(record) => {
+ if record.deployment != deployment || record.configuration != selection {
+ return Err("original public Host selection changed".into());
+ }
+ }
+ None => {
+ let preparation = Preparation {
+ input: input.clone(),
+ packages: packages
+ .iter()
+ .map(|package| {
+ package
+ .files()
+ .iter()
+ .map(|(path, bytes)| (path.clone(), STANDARD.encode(bytes)))
+ .collect()
+ })
+ .collect(),
+ };
+ store
+ .begin_application_launch(Launch {
+ deployment,
+ configuration: selection,
+ pending: Some(serde_json::to_vec(&preparation)?),
+ })
+ .await?;
+ }
+ }
+ prepare(store, Roots::Saved).await?;
+ launch::ensure(store, deployment, input, &[], packages).await
+}
+
+/// Bind an explicitly committed generic Host without reauthoring its source.
+pub(crate) async fn select_committed_host(
+ store: &EventLog,
+ request: &maka_transport::startup::Request,
+) -> Result<(), HostError> {
+ if let Some(record) = store.application_launch().await? {
+ let _: Selection = serde_json::from_value(record.configuration)?;
+ if record.deployment != request.deployment {
+ return Err("saved public Host selects another application deployment".into());
+ }
+ // Existing publication/grants remain authoritative, even when this UI
+ // selects a narrower subset. The peer handshake validates that subset.
+ return Ok(());
+ }
+ let source = store
+ .read_application_metadata(request.deployment)
+ .await?
+ .ok_or("commit the selected Host deployment before opening its UI")?;
+ if source.source.input.environment.location
+ != maka_plugin_core::composition::ExecutionLocation::Host
+ || !source.source.remote_exports.is_empty()
+ {
+ return Err("selected deployment is not a local Host source".into());
+ }
+ store
+ .select_committed_application_launch(
+ source.source.reference,
+ serde_json::to_value(Selection::PublicHost {
+ endpoints: request.endpoints.clone(),
+ native: request.native.clone(),
+ })?,
+ )
+ .await?;
+ Ok(())
+}
+
pub(crate) async fn open_host(
owner: RootOwner,
options: Options,
@@ -133,18 +206,22 @@ async fn begin(store: &EventLog, roots: Option<Vec<DirectoryRootSpec>>) -> Resul
.collect(),
);
let packages = tokio::task::spawn_blocking(packages).await??;
- let (sources, _) = author(Product::Agent, configuration, &packages).await?;
+ let (sources, _, host_plan) = author(Product::Agent, configuration, &packages).await?;
let connections: maka_transport::Configuration = serde_json::from_value(sources.connections)?;
let connection = connections
.hosts
.into_iter()
.find(|host| host.name == "local")
.ok_or("product public Host missing")?;
- let packages = selected(
- &sources.host,
- &[],
- &packages.into_values().collect::<Vec<_>>(),
- )?;
+ let required: std::collections::BTreeSet<_> = host_plan
+ .instances()
+ .values()
+ .map(|instance| &instance.package)
+ .collect();
+ let packages: Vec<_> = packages
+ .into_values()
+ .filter(|package| required.contains(&package.application_ref()))
+ .collect();
let selection = Selection::PublicHost {
endpoints: connection.endpoints,
native: connection.native,
diff --git a/products/maka/cli/src/application/product/launch.rs b/products/maka/cli/src/application/product/launch.rs
index 161fb3cb63..5b4fdff65b 100644
--- a/products/maka/cli/src/application/product/launch.rs
+++ b/products/maka/cli/src/application/product/launch.rs
@@ -117,7 +117,7 @@ impl Launch {
Ok(Some(selected))
}
}.await;
- let (deployment, configuration) = match result {
+ let (deployment, configuration, anchors) = match result {
Ok(Some(value)) => value,
Ok(None) => {
store.drain().await;
@@ -128,7 +128,7 @@ impl Launch {
return Err(error);
}
};
- run_tui(store, deployment, configuration, startup).await
+ run_tui(store, deployment, configuration, startup, anchors).await
}
}
async fn standard_configuration(
@@ -195,12 +195,12 @@ pub(super) fn targets(
configuration: &Configuration,
) -> Result<(PathBuf, String, Uuid, Uuid), HostError> {
configuration.validate().map_err(HostError::from)?;
- let mut process = None;
+ let mut public = None;
let mut management = None;
for host in &configuration.hosts {
match &host.location {
- HostLocation::Process { root, state_domain } if process.is_none() => {
- process = Some((
+ HostLocation::Public { root, state_domain } if public.is_none() => {
+ public = Some((
root.clone(),
state_domain
.clone()
@@ -209,10 +209,10 @@ pub(super) fn targets(
))
}
HostLocation::Cohost {} if management.is_none() => management = Some(host.deployment),
- _ => return Err("product launch requires one process Host and one cohost".into()),
+ _ => return Err("product launch requires one public Host and one cohost".into()),
}
}
- let (path, domain, host) = process.ok_or("product Host is missing")?;
+ let (path, domain, host) = public.ok_or("product Host is missing")?;
Ok((
path,
domain,
@@ -224,7 +224,14 @@ pub(super) async fn prepare(
store: &EventLog,
configuration: Option<Installation>,
product: Product,
-) -> Result<(Uuid, Configuration), HostError> {
+) -> Result<
+ (
+ Uuid,
+ Configuration,
+ Vec<maka_host_runtime::application::host::Client>,
+ ),
+ HostError,
+> {
let record = match store.application_launch().await? {
Some(record) => {
if configuration.is_some() {
@@ -262,25 +269,55 @@ pub(super) async fn prepare(
&preparation.sources.remote_exports,
&packages,
)?;
- let host_store = open(root, false, Some(domain)).await?;
+ if preparation.restore.is_empty() {
+ bootstrap_host(
+ &root,
+ &domain,
+ host,
+ &connections,
+ &preparation.sources.host,
+ &host_packages,
+ )
+ .await?;
+ }
+ let public = super::public::Host::connect(root.clone(), domain.clone(), host).await?;
let result = async {
- for (target, deployment, input, observations, packages) in [
- (
- &host_store,
- host,
- &preparation.sources.host,
- &[][..],
- host_packages.as_slice(),
- ),
+ if preparation.restore.is_empty() {
+ public
+ .ensure_initial(&preparation.sources.host, &[], &host_packages)
+ .await?;
+ } else {
+ let previous = host_configuration(&connections)?;
+ let desired = host_configuration(&serde_json::from_value(
+ preparation.sources.connections.clone(),
+ )?)?;
+ let configuration =
+ if serde_json::to_value(&previous)? == serde_json::to_value(&desired)? {
+ None
+ } else {
+ Some((previous, desired))
+ };
+ public
+ .ensure_restore(
+ *preparation
+ .restore
+ .get(&host)
+ .ok_or("restore revision missing")?,
+ &preparation.sources.host,
+ &[],
+ &host_packages,
+ configuration,
+ )
+ .await?;
+ }
+ for (deployment, input, observations, packages) in [
(
- store,
management,
&preparation.sources.management_host,
&[][..],
management_packages.as_slice(),
),
(
- store,
record.deployment,
&preparation.sources.ui,
preparation.sources.remote_exports.as_slice(),
@@ -288,10 +325,10 @@ pub(super) async fn prepare(
),
] {
if preparation.restore.is_empty() {
- ensure(target, deployment, input, observations, packages).await?;
+ ensure(store, deployment, input, observations, packages).await?;
} else {
super::restore::ensure(
- target,
+ store,
deployment,
*preparation
.restore
@@ -319,13 +356,67 @@ pub(super) async fn prepare(
Ok::<_, HostError>(())
}
.await;
- host_store.drain().await;
result?;
- } else if maka_event_log::root::resolve(&root)?.root_id() != domain {
- return Err("original product Host Root changed".into());
+ return Ok((record.deployment, connections, vec![public.into_client()]));
}
- Ok((record.deployment, connections))
+ let public = super::public::Host::connect(root, domain, host).await?;
+ Ok((record.deployment, connections, vec![public.into_client()]))
+}
+fn host_configuration(
+ configuration: &Configuration,
+) -> Result<maka_host_runtime::application::host::Configuration, HostError> {
+ let connection = configuration
+ .hosts
+ .iter()
+ .find(|connection| matches!(&connection.location, HostLocation::Public { .. }))
+ .ok_or("product public Host missing")?;
+ let configuration = maka_host_runtime::application::host::Configuration::PublicHost {
+ endpoints: connection.endpoints.clone(),
+ native: connection.native.clone(),
+ };
+ configuration.validate()?;
+ Ok(configuration)
}
+
+async fn bootstrap_host(
+ root: &std::path::Path,
+ domain: &str,
+ deployment: Uuid,
+ connections: &Configuration,
+ input: &Input,
+ packages: &[Package],
+) -> Result<(), HostError> {
+ let host_store = match open(root.into(), false, Some(domain.into())).await {
+ Ok(store) => store,
+ Err(error)
+ if error
+ .downcast_ref::<std::io::Error>()
+ .is_some_and(|error| error.kind() == std::io::ErrorKind::WouldBlock) =>
+ {
+ return Ok(());
+ }
+ Err(error) => return Err(error),
+ };
+ let connection = connections
+ .hosts
+ .iter()
+ .find(|connection| matches!(&connection.location, HostLocation::Public { .. }))
+ .ok_or("product public Host missing")?;
+ let result = super::headless::save_product(
+ &host_store,
+ deployment,
+ input,
+ packages,
+ connection.endpoints.clone(),
+ connection.native.clone(),
+ )
+ .await;
+ // Public startup can acquire Root only after bootstrap has actually drained.
+ host_store.drain().await;
+ drop(host_store);
+ result
+}
+
async fn begin(store: &EventLog, input: Installation, product: Product) -> Result<(), HostError> {
let version = super::bundle_fingerprint()?;
let authoring = super::authoring()?;
@@ -363,9 +454,9 @@ async fn begin(store: &EventLog, input: Installation, product: Product) -> Resul
}
let packages = tokio::task::spawn_blocking(packages).await??;
let version = super::verified_fingerprint(&authoring, &packages, &version)?;
- let (mut sources, installation) = super::author_with_source(product, configuration, &packages, &authoring).await?;
+ let (mut sources, installation, _) = super::author_with_source(product, configuration, &packages, &authoring).await?;
for host in sources.connections["hosts"].as_array_mut().ok_or("product connections missing")? {
- if host["location"]["kind"] == "process" {
+ if host["location"]["kind"] == "public" {
host["location"]["stateDomain"] = Value::String(owner.root_id().into());
}
}
diff --git a/products/maka/cli/src/application/product/launch/tests.rs b/products/maka/cli/src/application/product/launch/tests.rs
index e10dfc4b29..22152ed3d8 100644
--- a/products/maka/cli/src/application/product/launch/tests.rs
+++ b/products/maka/cli/src/application/product/launch/tests.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use super::super::test_public::PublishedHost;
use super::*;
use maka_plugin_core::composition::{Environment, ExecutionLocation, PresentationTarget};
use serde_json::json;
@@ -39,7 +40,7 @@ async fn pending(store: &EventLog, host: &EventLog) -> (Uuid, Uuid, Uuid) {
let endpoint = json!({"service":{"consumer":[Uuid::new_v4()],"import":"service","grant":{"methods":["read"],"recovery":true}}});
let owner = host.owned_root().unwrap();
let connections = json!({"hosts":[
- {"name":"local","location":{"kind":"process","root":owner.canonical_path(),"stateDomain":owner.root_id()},"deployment":host_id,"endpoints":endpoint},
+ {"name":"local","location":{"kind":"public","root":owner.canonical_path(),"stateDomain":owner.root_id()},"deployment":host_id,"endpoints":endpoint},
{"name":"manager","location":{"kind":"cohost"},"deployment":management,"endpoints":endpoint}
]});
let selection = connections.clone();
@@ -73,7 +74,7 @@ async fn partial_initialization_resumes_original_inputs_and_preserves_new_heads(
let ui_path = directory.path().join("ui");
let host_path = directory.path().join("host");
let store = open(ui_path.clone(), true, None).await.unwrap();
- let host = open(host_path.clone(), true, None).await.unwrap();
+ let host = Arc::new(open(host_path.clone(), true, None).await.unwrap());
let (ui, host_id, management) = pending(&store, &host).await;
ensure(&host, host_id, &input(), &[], &[]).await.unwrap();
let mut edited = input();
@@ -87,12 +88,11 @@ async fn partial_initialization_resumes_original_inputs_and_preserves_new_heads(
ensure(&store, management, &input(), &[], &[])
.await
.unwrap();
- host.drain().await;
+ let serving = PublishedHost::start(host.clone(), host_id).await;
store.drain().await;
- drop(host);
drop(store);
let store = open(ui_path.clone(), false, None).await.unwrap();
- let (selected, _) = prepare(&store, None, Product::Studio).await.unwrap();
+ let (selected, _, clients) = prepare(&store, None, Product::Studio).await.unwrap();
assert_eq!(selected, ui);
assert!(
store
@@ -125,7 +125,6 @@ async fn partial_initialization_resumes_original_inputs_and_preserves_new_heads(
.revision,
1
);
- let host = open(host_path, false, None).await.unwrap();
assert_eq!(
host.load_application(host_id)
.await
@@ -136,7 +135,7 @@ async fn partial_initialization_resumes_original_inputs_and_preserves_new_heads(
.revision,
2
);
- host.drain().await;
+ drop(clients);
store
.commit_application(ui, 1, input(), vec![], vec![])
.await
@@ -162,6 +161,8 @@ async fn partial_initialization_resumes_original_inputs_and_preserves_new_heads(
.revision,
2
);
+ serving.stop().await;
+ drop(host);
store.drain().await;
}
@@ -169,9 +170,11 @@ async fn partial_initialization_resumes_original_inputs_and_preserves_new_heads(
async fn conflicting_original_commit_keeps_pending_record_and_does_not_create_ui() {
let directory = tempfile::tempdir().unwrap();
let store = open(directory.path().join("ui"), true, None).await.unwrap();
- let host = open(directory.path().join("host"), true, None)
- .await
- .unwrap();
+ let host = Arc::new(
+ open(directory.path().join("host"), true, None)
+ .await
+ .unwrap(),
+ );
let (ui, host_id, _) = pending(&store, &host).await;
let mut conflicting = input();
conflicting.environment = Environment {
@@ -181,8 +184,7 @@ async fn conflicting_original_commit_keeps_pending_record_and_does_not_create_ui
host.commit_application(host_id, 0, conflicting, vec![], vec![])
.await
.unwrap();
- host.drain().await;
- drop(host);
+ let serving = PublishedHost::start(host.clone(), host_id).await;
let error = prepare(&store, None, Product::Agent).await.err().unwrap();
assert!(error.to_string().contains("conflicts"), "{error}");
assert!(
@@ -195,6 +197,8 @@ async fn conflicting_original_commit_keeps_pending_record_and_does_not_create_ui
.is_some()
);
assert!(store.load_application(ui).await.unwrap().is_none());
+ serving.stop().await;
+ drop(host);
store.drain().await;
}
diff --git a/products/maka/cli/src/application/product/public.rs b/products/maka/cli/src/application/product/public.rs
new file mode 100644
index 0000000000..169e445d24
--- /dev/null
+++ b/products/maka/cli/src/application/product/public.rs
@@ -0,0 +1,406 @@
+/*
+ * 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.
+ */
+//! Product lifecycle operations admitted by the independently owned public Host.
+use super::{HostError, Input, Package, RemoteObservation};
+use maka_capabilities::call::{Attachments, ReadResource, ResourceEncoding};
+use maka_host_runtime::application::{
+ host::{Client, LocalCall},
+ transport::wire::Reply,
+};
+use maka_plugin_core::{
+ contract::{ResourceType, binary_resource_type, text_resource_type},
+ deployment::{
+ DeploymentRef,
+ management::{
+ DeploymentDocument, DeploymentDocumentChunk, ManagedDeployment, ManagementResult,
+ PackageBundleExport, deployment_manager_contract,
+ },
+ },
+ operation::{OperationCleanup, OperationOutcome, Receipt},
+ package::PackageRef,
+};
+use serde_json::{Value, json};
+use std::{collections::BTreeMap, ops::Range, path::PathBuf, sync::Arc};
+use uuid::Uuid;
+
+pub(super) struct Host {
+ client: Client,
+ domain: String,
+ deployment: Uuid,
+}
+impl Host {
+ pub(super) async fn connect(
+ path: PathBuf,
+ domain: String,
+ deployment: Uuid,
+ ) -> Result<Self, HostError> {
+ if maka_event_log::root::resolve(&path)?.root_id() != domain {
+ return Err("original product Host Root changed".into());
+ }
+ let client = crate::deployment::connect_local(path).await?;
+ if client
+ .catalogue()
+ .select("management")?
+ .description()
+ .contract
+ != deployment_manager_contract()
+ {
+ return Err("product Host management contract changed".into());
+ }
+ let host = Self {
+ client,
+ domain,
+ deployment,
+ };
+ host.read().await?;
+ Ok(host)
+ }
+ pub(super) fn into_client(self) -> Client {
+ self.client
+ }
+ fn check_source(&self, source: &DeploymentRef) -> Result<(), HostError> {
+ source.validate()?;
+ if source.state_domain != self.domain || source.deployment != self.deployment {
+ return Err("product management names a foreign Host deployment".into());
+ }
+ Ok(())
+ }
+ pub(super) async fn read(&self) -> Result<ManagedDeployment, HostError> {
+ let head: ManagedDeployment = serde_json::from_value(self.query("read", json!({})).await?)?;
+ self.check_source(&head.source.reference)?;
+ Ok(head)
+ }
+ async fn query(&self, method: &str, input: Value) -> Result<Value, HostError> {
+ let mut call = self
+ .client
+ .call("management", method.into(), input, BTreeMap::new())
+ .await?;
+ let result: Result<Value, HostError> = async {
+ call.start().await?;
+ match call.receive().await? {
+ Reply::Finished { result, cleanup } => {
+ cleanup?;
+ Ok(result?)
+ }
+ Reply::Aborted { reason, cleanup } => {
+ cleanup?;
+ Err(reason.into())
+ }
+ _ => Err("product management query did not finish".into()),
+ }
+ }
+ .await;
+ settle(&mut call).await;
+ result
+ }
+ pub(super) async fn document(
+ &self,
+ source: &DeploymentRef,
+ ) -> Result<DeploymentDocument, HostError> {
+ self.check_source(source)?;
+ let mut text = String::new();
+ loop {
+ let chunk: DeploymentDocumentChunk = serde_json::from_value(
+ self.query("readDocument", json!({"source":source,"offset":text.len()}))
+ .await?,
+ )?;
+ if chunk.source != *source
+ || chunk.offset != text.len()
+ || chunk.total_bytes > maka_plugin_core::deployment::management::MAX_DOCUMENT_BYTES
+ || chunk.text.len() > chunk.total_bytes.saturating_sub(text.len())
+ {
+ return Err("product Host returned an invalid document continuation".into());
+ }
+ text.push_str(&chunk.text);
+ match chunk.next {
+ Some(next) if next == text.len() && !chunk.text.is_empty() => {}
+ None if text.len() == chunk.total_bytes => break,
+ _ => return Err("product Host returned an incomplete document".into()),
+ }
+ }
+ Ok(serde_json::from_str(&text)?)
+ }
+ pub(super) async fn packages(
+ &self,
+ source: &DeploymentRef,
+ references: &[PackageRef],
+ ) -> Result<Vec<Package>, HostError> {
+ self.check_source(source)?;
+ let mut packages = Vec::new();
+ for reference in references {
+ let mut call = self
+ .client
+ .call(
+ "management",
+ "exportPackage".into(),
+ json!({"source":source,"package":reference}),
+ BTreeMap::new(),
+ )
+ .await?;
+ let result = async {
+ call.start().await?;
+ if !matches!(call.receive().await?, Reply::Subscribed) {
+ return Err(HostError::from("product package export did not subscribe"));
+ }
+ let mut bytes = Vec::new();
+ let mut total = None;
+ loop {
+ match call.pull().await? {
+ Reply::Item { value } => {
+ let PackageBundleExport::Chunk {
+ offset,
+ total_bytes,
+ text,
+ } = serde_json::from_value(value)?
+ else {
+ return Err("product package export ended without cleanup".into());
+ };
+ if offset != bytes.len()
+ || total_bytes > maka_packages::MAX_PACKAGE_BYTES * 2
+ || total.is_some_and(|total| total != total_bytes)
+ || text.is_empty()
+ || text.len() > total_bytes.saturating_sub(bytes.len())
+ {
+ return Err("product package export continuation changed".into());
+ }
+ total = Some(total_bytes);
+ bytes.extend_from_slice(text.as_bytes());
+ }
+ Reply::Finished { result, cleanup } => {
+ cleanup?;
+ if !matches!(
+ serde_json::from_value::<PackageBundleExport>(result?)?,
+ PackageBundleExport::Complete {}
+ ) || total != Some(bytes.len())
+ {
+ return Err("product package export is incomplete".into());
+ }
+ let package = Package::from_bundle(&bytes)?;
+ if package.application_ref() != *reference {
+ return Err("product package export identity changed".into());
+ }
+ break Ok(package);
+ }
+ _ => return Err("product package export did not finish".into()),
+ }
+ }
+ }
+ .await;
+ settle(&mut call).await;
+ packages.push(result?);
+ }
+ Ok(packages)
+ }
+ pub(super) async fn replace(
+ &self,
+ expected: u64,
+ input: &Input,
+ observations: &[RemoteObservation],
+ packages: &[Package],
+ ) -> Result<(), HostError> {
+ let document = DeploymentDocument {
+ input: input.clone(),
+ remote_exports: observations.to_vec(),
+ packages: packages.iter().map(Package::application_ref).collect(),
+ };
+ let bundles: Vec<Value> = packages
+ .iter()
+ .map(|package| serde_json::from_slice(&package.to_bundle()))
+ .collect::<Result<_, _>>()?;
+ let resources: Attachments = BTreeMap::from([
+ (
+ "document".into(),
+ Arc::new(Bytes {
+ bytes: serde_json::to_vec(&document)?,
+ encoding: ResourceEncoding::Utf8,
+ }) as Arc<dyn ReadResource>,
+ ),
+ (
+ "packages".into(),
+ Arc::new(Bytes {
+ bytes: serde_json::to_vec(&bundles)?,
+ encoding: ResourceEncoding::Bytes,
+ }) as Arc<dyn ReadResource>,
+ ),
+ ]);
+ let mut call = self
+ .client
+ .call(
+ "management",
+ "replace".into(),
+ json!({"expectedRevision":expected}),
+ resources,
+ )
+ .await?;
+ let result = async {
+ call.start().await?;
+ if !matches!(call.receive().await?, Reply::Prepared { .. }) {
+ return Err(HostError::from("product replacement was not prepared"));
+ }
+ let Reply::Mutation { reply, cleanup } = call.confirm().await? else {
+ return Err("product replacement did not return a receipt".into());
+ };
+ cleanup?;
+ let Receipt::Recorded {
+ outcome: OperationOutcome::Returned { value },
+ cleanup: OperationCleanup::Confirmed,
+ } = reply.receipt
+ else {
+ return Err("product replacement outcome remains unconfirmed".into());
+ };
+ match serde_json::from_value::<ManagementResult<DeploymentRef>>(value)? {
+ ManagementResult::Committed { value } => {
+ self.check_source(&value)?;
+ if value.revision != expected + 1 {
+ return Err("product replacement revision changed".into());
+ }
+ Ok(())
+ }
+ ManagementResult::Conflict { current_revision } => Err(format!(
+ "product replacement conflicts with revision {current_revision}"
+ )
+ .into()),
+ ManagementResult::Rejected { reason, diagnostic } => Err(format!(
+ "product replacement rejected: {reason:?}: {}",
+ diagnostic.unwrap_or_default()
+ )
+ .into()),
+ }
+ }
+ .await;
+ settle(&mut call).await;
+ result
+ }
+ pub(super) async fn ensure_initial(
+ &self,
+ input: &Input,
+ observations: &[RemoteObservation],
+ packages: &[Package],
+ ) -> Result<(), HostError> {
+ let source = DeploymentRef {
+ state_domain: self.domain.clone(),
+ deployment: self.deployment,
+ revision: 1,
+ };
+ let document = self.document(&source).await?;
+ if !same_document(&document, input, observations, packages)? {
+ return Err(
+ "original product initialization source conflicts with stored revision 1".into(),
+ );
+ }
+ // Read the original immutable bytes as well as their declared digests.
+ self.packages(&source, &document.packages).await?;
+ Ok(())
+ }
+ pub(super) async fn ensure_restore(
+ &self,
+ expected: u64,
+ input: &Input,
+ observations: &[RemoteObservation],
+ packages: &[Package],
+ configuration: Option<(
+ maka_host_runtime::application::host::Configuration,
+ maka_host_runtime::application::host::Configuration,
+ )>,
+ ) -> Result<(), HostError> {
+ let head = self.read().await?;
+ let source = head.source.reference.clone();
+ let document = DeploymentDocument {
+ input: head.source.input,
+ remote_exports: head.source.remote_exports,
+ packages: head.packages,
+ };
+ let revision = head.source.reference.revision;
+ let matches = same_document(&document, input, observations, packages)?;
+ if revision != expected && !(matches && revision == expected + 1) {
+ return Err("restoration conflicts with a later source revision".into());
+ }
+ if let Some((previous, configuration)) = configuration {
+ // Authority is a separate durable step coordinated by the original
+ // UI preparation. It must precede admitting the new source.
+ self.client
+ .configure(source, previous, configuration)
+ .await?;
+ }
+ if matches {
+ return Ok(());
+ }
+ self.replace(expected, input, observations, packages).await
+ }
+}
+fn same_document(
+ document: &DeploymentDocument,
+ input: &Input,
+ observations: &[RemoteObservation],
+ packages: &[Package],
+) -> Result<bool, HostError> {
+ let mut actual = document.packages.clone();
+ let mut desired: Vec<_> = packages.iter().map(Package::application_ref).collect();
+ actual.sort();
+ desired.sort();
+ Ok(actual == desired
+ && serde_json::to_value(&document.input)? == serde_json::to_value(input)?
+ && serde_json::to_value(&document.remote_exports)? == serde_json::to_value(observations)?)
+}
+async fn settle(call: &mut LocalCall) {
+ let mut reported = false;
+ while !call.cleanup_confirmed() {
+ if let Err(error) = call.close().await {
+ if !reported {
+ eprintln!("product Host call cleanup unconfirmed: {error}");
+ reported = true;
+ }
+ tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+ }
+ }
+}
+struct Bytes {
+ bytes: Vec<u8>,
+ encoding: ResourceEncoding,
+}
+impl ReadResource for Bytes {
+ fn encoding(&self) -> ResourceEncoding {
+ self.encoding
+ }
+ fn resource_type(&self) -> ResourceType {
+ match self.encoding {
+ ResourceEncoding::Utf8 => text_resource_type(),
+ ResourceEncoding::Bytes => binary_resource_type(),
+ }
+ }
+ fn bytes(&self) -> usize {
+ self.bytes.len()
+ }
+ fn metadata(&self) -> Value {
+ json!({})
+ }
+ fn read(&self, range: Range<usize>) -> Result<Vec<u8>, String> {
+ if self.encoding == ResourceEncoding::Utf8 {
+ let text = std::str::from_utf8(&self.bytes).map_err(|error| error.to_string())?;
+ text.get(range)
+ .map(|text| text.as_bytes().to_vec())
+ .ok_or("invalid UTF-8 resource range".into())
+ } else {
+ self.bytes
+ .get(range)
+ .map(<[u8]>::to_vec)
+ .ok_or("invalid binary resource range".into())
+ }
+ }
+}
diff --git a/products/maka/cli/src/application/product/restore.rs b/products/maka/cli/src/application/product/restore.rs
index 5cea59a3bb..f6f83cd949 100644
--- a/products/maka/cli/src/application/product/restore.rs
+++ b/products/maka/cli/src/application/product/restore.rs
@@ -28,7 +28,7 @@ use uuid::Uuid;
#[derive(Args)]
pub(crate) struct Restore {
- /// UI State Root (default: ~/.maka/application). Close its UI and local Host first.
+ /// UI State Root (default: ~/.maka/application). Close its UI first.
#[arg(long)]
root: Option<PathBuf>,
}
@@ -65,16 +65,21 @@ async fn restore(store: &EventLog) -> Result<DeploymentRef, HostError> {
}
let connections = serde_json::from_value(record.configuration.clone())?;
let (path, domain, host_id, management) = launch::targets(&connections)?;
- let host = launch::open(path, false, Some(domain)).await?;
+ let host = super::public::Host::connect(path, domain, host_id).await?;
let result = async {
let mut revisions = BTreeMap::new();
- let mut originals = Vec::new();
+ let current = host.read().await?;
+ revisions.insert(host_id, current.source.reference.revision);
+ let mut baseline = current.source.reference;
+ baseline.revision = 1;
+ let original = host.document(&baseline).await?;
let mut packages = BTreeMap::new();
- for (target, deployment) in [
- (&host, host_id),
- (store, management),
- (store, record.deployment),
- ] {
+ for package in host.packages(&baseline, &original.packages).await? {
+ packages.insert(package.application_ref(), package);
+ }
+ let mut originals = vec![original];
+ for deployment in [management, record.deployment] {
+ let target = store;
let current = target
.read_application_metadata(deployment)
.await?
@@ -89,11 +94,21 @@ async fn restore(store: &EventLog) -> Result<DeploymentRef, HostError> {
for package in original.packages() {
packages.insert(package.application_ref(), package.clone());
}
- originals.push(original);
+ originals.push(
+ maka_plugin_core::deployment::management::DeploymentDocument {
+ input: original.source().input.clone(),
+ remote_exports: original.source().remote_exports.clone(),
+ packages: original
+ .packages()
+ .iter()
+ .map(Package::application_ref)
+ .collect(),
+ },
+ );
}
- let ui = originals.pop().unwrap().into_parts().0;
- let management = originals.pop().unwrap().into_parts().0;
- let host = originals.pop().unwrap().into_parts().0;
+ let ui = originals.pop().unwrap();
+ let management = originals.pop().unwrap();
+ let host = originals.pop().unwrap();
let preparation = launch::Preparation {
product: None,
previous_connections: None,
@@ -122,8 +137,6 @@ async fn restore(store: &EventLog) -> Result<DeploymentRef, HostError> {
Ok::<_, HostError>(())
}
.await;
- host.drain().await;
- drop(host);
result?;
launch::prepare(store, None, Product::Agent).await?;
Ok(store
@@ -181,10 +194,12 @@ pub(super) async fn ensure(
#[cfg(test)]
mod tests {
+ use super::super::test_public::PublishedHost;
use super::*;
use maka_event_log::applications::launch::Launch;
use maka_plugin_core::composition::{Environment, ExecutionLocation};
use serde_json::json;
+ use std::sync::Arc;
fn input() -> Input {
Input {
@@ -204,18 +219,18 @@ mod tests {
.await
.unwrap();
let host_path = directory.path().join("host");
- let host = launch::open(host_path.clone(), true, None).await.unwrap();
+ let host = Arc::new(launch::open(host_path.clone(), true, None).await.unwrap());
let ui = Uuid::new_v4();
let host_id = Uuid::new_v4();
let manager = Uuid::new_v4();
let owner = host.owned_root().unwrap();
let endpoint = json!({"service":{"consumer":[Uuid::new_v4()],"import":"service","grant":{"methods":["read"],"recovery":true}}});
let configuration = json!({"hosts":[
- {"name":"local","location":{"kind":"process","root":owner.canonical_path(),"stateDomain":owner.root_id()},"deployment":host_id,"endpoints":endpoint},
+ {"name":"local","location":{"kind":"public","root":owner.canonical_path(),"stateDomain":owner.root_id()},"deployment":host_id,"endpoints":endpoint},
{"name":"manager","location":{"kind":"cohost"},"deployment":manager,"endpoints":endpoint}
]});
drop(owner);
- for (target, id) in [(&host, host_id), (&store, manager), (&store, ui)] {
+ for (target, id) in [(host.as_ref(), host_id), (&store, manager), (&store, ui)] {
target
.commit_application(id, 0, input(), vec![], vec![])
.await
@@ -253,8 +268,7 @@ mod tests {
})
.await
.unwrap();
- host.drain().await;
- drop(host);
+ let serving = PublishedHost::start(host.clone(), host_id).await;
let restored = restore(&store).await.unwrap();
assert_eq!(restored.deployment, ui);
assert_eq!(restored.revision, 3);
@@ -275,7 +289,6 @@ mod tests {
);
assert_eq!(restore(&store).await.unwrap(), restored); // Already restored: no new revision.
- let host = launch::open(host_path.clone(), false, None).await.unwrap();
let preparation = launch::Preparation {
product: None,
previous_connections: None,
@@ -297,10 +310,7 @@ mod tests {
host.commit_application(host_id, 3, input(), vec![], vec![])
.await
.unwrap();
- host.drain().await;
- drop(host);
assert_eq!(restore(&store).await.unwrap(), restored);
- let host = launch::open(host_path, false, None).await.unwrap();
assert_eq!(
host.read_application_metadata(host_id)
.await
@@ -311,8 +321,60 @@ mod tests {
.revision,
4
);
- assert!(ensure(&host, host_id, 2, &input(), &[], &[]).await.is_err());
- host.drain().await;
+ let public = super::super::public::Host::connect(
+ host_path,
+ host.owned_root().unwrap().root_id().into(),
+ host_id,
+ )
+ .await
+ .unwrap();
+ assert!(
+ public
+ .ensure_restore(2, &input(), &[], &[], None)
+ .await
+ .is_err()
+ );
+ drop(public);
+ // A pending launcher restore must survive rejection of its first source.
+ let mut conflicting = preparation;
+ conflicting.restore.insert(host_id, 2);
+ store
+ .begin_application_restore(ui, serde_json::to_vec(&conflicting).unwrap())
+ .await
+ .unwrap();
+ let error = restore(&store).await.unwrap_err();
+ assert!(error.to_string().contains("conflicts"), "{error}");
+ assert!(
+ store
+ .application_launch()
+ .await
+ .unwrap()
+ .unwrap()
+ .pending
+ .is_some()
+ );
+ assert_eq!(
+ store
+ .read_application_metadata(ui)
+ .await
+ .unwrap()
+ .unwrap()
+ .source
+ .reference,
+ restored
+ );
+ assert_eq!(
+ host.read_application_metadata(host_id)
+ .await
+ .unwrap()
+ .unwrap()
+ .source
+ .reference
+ .revision,
+ 4
+ );
+ serving.stop().await;
+ drop(host);
store.drain().await;
}
}
diff --git a/products/maka/cli/src/application/product/test_public.rs b/products/maka/cli/src/application/product/test_public.rs
new file mode 100644
index 0000000000..136c1f4aed
--- /dev/null
+++ b/products/maka/cli/src/application/product/test_public.rs
@@ -0,0 +1,175 @@
+/*
+ * 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.
+ */
+
+//! Real public transport fixture; management operations retain their production implementation.
+use maka_event_log::EventLog;
+use maka_host_runtime::application::{
+ host::{Host, LifecycleMode, Options, discovery::Registration},
+ management::{ManagementTarget, provide_target},
+ transport::HostEndpoint,
+};
+use maka_packages::Package;
+use maka_plugin_core::{composition::Input, deployment::management::deployment_manager_contract};
+use serde_json::json;
+use std::{collections::BTreeMap, sync::Arc};
+use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
+
+pub(super) struct PublishedHost {
+ registration: Registration,
+ cancel: CancellationToken,
+ _cleanup: tokio_util::sync::DropGuard,
+ server: tokio::task::JoinHandle<Result<(), String>>,
+}
+impl PublishedHost {
+ /// A separate controller uses the trusted fixed-target manager capability,
+ /// so tests can retain deliberately empty or otherwise unlaunchable sources.
+ pub(super) async fn start(store: Arc<EventLog>, deployment: Uuid) -> Self {
+ let mut source = store
+ .read_application_metadata(deployment)
+ .await
+ .unwrap()
+ .unwrap()
+ .source
+ .reference;
+ source.revision = 1;
+ let target = ManagementTarget::from_source(store.clone(), source)
+ .await
+ .unwrap();
+ let contract = deployment_manager_contract();
+ let implementation = "maka.test.product-manager";
+ let package = |id: &str, implementation: &str, imports, exports| {
+ Package::new(BTreeMap::from([(
+ "maka.extension.json".into(),
+ serde_json::to_vec(&json!({
+ "schemaVersion":1,"id":id,"application":{"entries":{"host":{
+ "implementation":{"kind":"native","implementation":implementation},
+ "configuration":{"type":"object","additionalProperties":false},
+ "imports":imports,"exports":exports
+ }}}
+ }))
+ .unwrap(),
+ )]))
+ .unwrap()
+ };
+ let manager = package(
+ "maka.test.manager",
+ implementation,
+ json!({}),
+ json!({"Management":{"contract":contract,"availability":"required"}}),
+ );
+ let gateway = package(
+ "maka.test.gateway",
+ "maka.application.gateway",
+ json!({"management":{"contract":contract,"cardinality":{"kind":"one"}}}),
+ json!({}),
+ );
+ let consumer = Uuid::new_v4();
+ let input: Input = serde_json::from_value(json!({
+ "environment":{"location":"host"},"instances":{
+ "manager":{"key":Uuid::new_v4(),"package":{"source":"pinned","package":manager.application_ref()},"usePackage":"entry"},
+ "gateway":{"key":consumer,"package":{"source":"pinned","package":gateway.application_ref()},"usePackage":"entry","bindings":{"management":[{"kind":"instance","path":["manager"],"export":"Management"}]}}
+ }
+ })).unwrap();
+ let controller = Uuid::new_v4();
+ store
+ .commit_application(controller, 0, input, vec![], vec![manager, gateway])
+ .await
+ .unwrap();
+ let mut native = crate::application::serve::native().unwrap();
+ provide_target(&mut native, implementation.into(), target).unwrap();
+ let endpoints: BTreeMap<String, HostEndpoint> = serde_json::from_value(json!({"management":{
+ "consumer":[consumer],"import":"management","grant":{"methods":["read","readDocument","exportPackage","replace"],"recovery":false}
+ }})).unwrap();
+ let host = Host::open_with_options(
+ store,
+ controller,
+ native,
+ None,
+ endpoints,
+ vec![],
+ Options {
+ lifecycle_mode: LifecycleMode::Service,
+ ..Default::default()
+ },
+ )
+ .await
+ .unwrap();
+ Self::publish(host)
+ }
+ pub(super) async fn start_product(
+ store: Arc<EventLog>,
+ deployment: Uuid,
+ endpoints: BTreeMap<String, HostEndpoint>,
+ native: Vec<maka_host_runtime::application::transport::NativeGrant>,
+ ) -> Self {
+ let original = store.load_application(deployment).await.unwrap().unwrap();
+ super::headless::save_product(
+ &store,
+ deployment,
+ &original.source().input,
+ original.packages(),
+ endpoints.clone(),
+ native.clone(),
+ )
+ .await
+ .unwrap();
+ let host = Host::open_with_options(
+ store,
+ deployment,
+ crate::application::serve::native().unwrap(),
+ Some(maka_host_js::factory(Default::default())),
+ endpoints,
+ native,
+ Options {
+ lifecycle_mode: LifecycleMode::Service,
+ ..Default::default()
+ },
+ )
+ .await
+ .unwrap();
+ Self::publish(host)
+ }
+ fn publish(host: Arc<Host>) -> Self {
+ let endpoint = crate::endpoint::LocalEndpoint::bind().unwrap();
+ let registration = host.publish_registration(&endpoint.path, None).unwrap();
+ let cancel = CancellationToken::new();
+ let server = tokio::spawn(serve(endpoint, host, cancel.clone()));
+ Self {
+ registration,
+ _cleanup: cancel.clone().drop_guard(),
+ cancel,
+ server,
+ }
+ }
+ pub(super) async fn stop(self) {
+ self.cancel.cancel();
+ self.server.await.unwrap().unwrap();
+ self.registration.remove().unwrap();
+ }
+}
+
+// Owning the whole endpoint keeps its Unix socket directory alive while serving.
+async fn serve(
+ endpoint: crate::endpoint::LocalEndpoint,
+ host: Arc<Host>,
+ cancel: CancellationToken,
+) -> Result<(), String> {
+ host.serve_local(endpoint.listener, cancel).await
+}
diff --git a/products/maka/cli/src/application/product/update.rs b/products/maka/cli/src/application/product/update.rs
index a5727100f2..f41e726f22 100644
--- a/products/maka/cli/src/application/product/update.rs
+++ b/products/maka/cli/src/application/product/update.rs
@@ -44,7 +44,7 @@ pub(super) struct Baseline {
#[derive(Args)]
pub(crate) struct Update {
- /// UI State Root (default: ~/.maka/application). Close its UI and local Host first.
+ /// UI State Root (default: ~/.maka/application). Close its UI first.
#[arg(long)]
root: Option<PathBuf>,
/// Apply the previewed merge only when there are no conflicts.
@@ -112,15 +112,14 @@ async fn synchronize_version(
let mut baseline: Baseline = serde_json::from_value(original.clone())?;
let connections = serde_json::from_value(record.configuration.clone())?;
let (path, domain, host_id, management) = launch::targets(&connections)?;
- let host = launch::open(path, false, Some(domain.clone())).await?;
- let result = async {
+ let host = super::public::Host::connect(path, domain.clone(), host_id).await?;
+ async {
let mut revisions = BTreeMap::new();
- let mut current = Vec::new();
- for (target, deployment) in [
- (&host, host_id),
- (store, management),
- (store, record.deployment),
- ] {
+ let head = host.read().await?;
+ revisions.insert(host_id, head.source.reference.revision);
+ let mut current = vec![head];
+ for deployment in [management, record.deployment] {
+ let target = store;
let head = target
.read_application_metadata(deployment)
.await?
@@ -163,14 +162,14 @@ async fn synchronize_version(
}
None => baseline.installation.clone(),
};
- let (mut next, installation) =
+ let (mut next, installation, _) =
super::author_with_source(baseline.product, configuration, &packages, &authoring)
.await?;
for endpoint in next.connections["hosts"]
.as_array_mut()
.ok_or("product connections missing")?
{
- if endpoint["location"]["kind"] == "process" {
+ if endpoint["location"]["kind"] == "public" {
endpoint["location"]["stateDomain"] = Value::String(domain.clone());
}
}
@@ -206,11 +205,14 @@ async fn synchronize_version(
// Unchanged defaults only need source metadata. Read and validate old
// package bytes when an actual merge needs to retain user-selected code.
let mut available = BTreeMap::new();
- for (target, head) in [
- (&host, ¤t[0]),
- (store, ¤t[1]),
- (store, ¤t[2]),
- ] {
+ for package in host
+ .packages(¤t[0].source.reference, ¤t[0].packages)
+ .await?
+ {
+ available.insert(package.application_ref(), package);
+ }
+ for head in [¤t[1], ¤t[2]] {
+ let target = store;
let loaded = target
.load_application_at(&head.source.reference)
.await?
@@ -268,9 +270,7 @@ async fn synchronize_version(
}
Ok::<_, HostError>(outcome)
}
- .await;
- host.drain().await;
- result
+ .await
}
/// Objects merge by field. Ordered lists and scalar values remain atomic; an
diff --git a/products/maka/cli/src/application/product/update/tests.rs b/products/maka/cli/src/application/product/update/tests.rs
index 64a69706a2..f660332f82 100644
--- a/products/maka/cli/src/application/product/update/tests.rs
+++ b/products/maka/cli/src/application/product/update/tests.rs
@@ -17,10 +17,12 @@
* under the License.
*/
+use super::super::test_public::PublishedHost;
use super::*;
use maka_event_log::applications::launch::Launch;
use maka_packages::Package;
use serde_json::json;
+use std::sync::Arc;
use uuid::Uuid;
#[test]
@@ -50,9 +52,12 @@ fn three_way_merge_preserves_customizations_and_reports_ambiguous_changes() {
assert_eq!(outcome.conflicts, ["/deleted"]);
}
-async fn old_installation(path: &std::path::Path, standard: bool) -> (EventLog, Uuid) {
+async fn old_installation(
+ path: &std::path::Path,
+ standard: bool,
+) -> (EventLog, Uuid, PublishedHost) {
let store = launch::open(path.join("ui"), true, None).await.unwrap();
- let host = launch::open(path.join("host"), true, None).await.unwrap();
+ let host = Arc::new(launch::open(path.join("host"), true, None).await.unwrap());
let options = json!({"workspace": path, "hostRoot": path.join("host"), "themeFile": path.join("theme.json"), "helper": std::env::current_exe().unwrap(), "locale":"zh-CN"});
let mut config = super::super::installation(options.clone()).await.unwrap();
// Previous ceilings blocked network and external-write approval requests.
@@ -78,7 +83,7 @@ async fn old_installation(path: &std::path::Path, standard: bool) -> (EventLog,
.unwrap()
.extend_from_slice(b"\n// previous product release\n");
packages.insert("agent-screen".into(), Package::new(files).unwrap());
- let (mut sources, installation) = super::super::author_with_source(
+ let (mut sources, installation, _) = super::super::author_with_source(
Product::Agent,
config,
&packages,
@@ -87,7 +92,7 @@ async fn old_installation(path: &std::path::Path, standard: bool) -> (EventLog,
.await
.unwrap();
for endpoint in sources.connections["hosts"].as_array_mut().unwrap() {
- if endpoint["location"]["kind"] == "process" {
+ if endpoint["location"]["kind"] == "public" {
endpoint["location"]["stateDomain"] = json!(host.owned_root().unwrap().root_id());
}
}
@@ -122,17 +127,29 @@ async fn old_installation(path: &std::path::Path, standard: bool) -> (EventLog,
})
.await
.unwrap();
- host.drain().await;
- drop(host);
+ let configuration = serde_json::from_value(preparation.sources.connections.clone()).unwrap();
+ let (_, _, host_id, _) = launch::targets(&configuration).unwrap();
+ let available = launch::decode_packages(preparation.packages).unwrap();
+ let selected = super::super::selected(&preparation.sources.host, &[], &available).unwrap();
+ launch::ensure(&host, host_id, &preparation.sources.host, &[], &selected)
+ .await
+ .unwrap();
+ let connection = configuration
+ .hosts
+ .into_iter()
+ .find(|connection| connection.deployment == host_id)
+ .unwrap();
+ let serving =
+ PublishedHost::start_product(host, host_id, connection.endpoints, connection.native).await;
launch::prepare(&store, None, Product::Agent).await.unwrap();
- (store, ui)
+ (store, ui, serving)
}
#[tokio::test]
#[ignore = "requires the built bundled product"]
async fn explicit_installation_keeps_its_selected_tools_when_packages_upgrade() {
let directory = tempfile::tempdir().unwrap();
- let (store, _) = old_installation(directory.path(), false).await;
+ let (store, _, serving) = old_installation(directory.path(), false).await;
assert!(begin(&store).await.unwrap());
let record = store.application_launch().await.unwrap().unwrap();
let preparation: launch::Preparation =
@@ -166,6 +183,7 @@ async fn explicit_installation_keeps_its_selected_tools_when_packages_upgrade()
assert!(!tools.iter().any(|tool| tool["import"] == "computer"));
launch::prepare(&store, None, Product::Agent).await.unwrap();
assert!(!begin(&store).await.unwrap());
+ serving.stop().await;
store.drain().await;
}
@@ -173,7 +191,7 @@ async fn explicit_installation_keeps_its_selected_tools_when_packages_upgrade()
#[ignore = "requires the built bundled product"]
async fn defaults_update_once_and_resume_an_acknowledgement_lost_after_commit() {
let directory = tempfile::tempdir().unwrap();
- let (store, ui) = old_installation(directory.path(), true).await;
+ let (store, ui, serving) = old_installation(directory.path(), true).await;
let original = store.load_application(ui).await.unwrap().unwrap();
let (original_source, _, _, _) = original.into_parts();
let preview = synchronize(&store, true, false).await.unwrap();
@@ -273,6 +291,7 @@ async fn defaults_update_once_and_resume_an_acknowledgement_lost_after_commit()
.pending
.is_none()
);
+ serving.stop().await;
store.drain().await;
}
@@ -280,7 +299,7 @@ async fn defaults_update_once_and_resume_an_acknowledgement_lost_after_commit()
#[ignore = "requires the built bundled product"]
async fn explicit_sync_keeps_custom_configuration_and_advances_the_default_baseline() {
let directory = tempfile::tempdir().unwrap();
- let (store, ui) = old_installation(directory.path(), true).await;
+ let (store, ui, serving) = old_installation(directory.path(), true).await;
let current = store.load_application(ui).await.unwrap().unwrap();
let (current_source, current_packages, _, _) = current.into_parts();
let mut customized = current_source.input;
@@ -327,6 +346,7 @@ async fn explicit_sync_keeps_custom_configuration_and_advances_the_default_basel
let preview = synchronize(&store, true, false).await.unwrap();
assert!(preview.customized && !preview.changed);
assert!(!begin(&store).await.unwrap());
+ serving.stop().await;
store.drain().await;
}
@@ -334,7 +354,7 @@ async fn explicit_sync_keeps_custom_configuration_and_advances_the_default_basel
#[ignore = "requires the built bundled product"]
async fn conflicting_custom_plugin_does_not_prepare_or_change_any_head() {
let directory = tempfile::tempdir().unwrap();
- let (store, ui) = old_installation(directory.path(), true).await;
+ let (store, ui, serving) = old_installation(directory.path(), true).await;
let current = store.load_application(ui).await.unwrap().unwrap();
let (current_source, current_packages, _, _) = current.into_parts();
let old_ref = match ¤t_source.input.instances["agent-screen"].package {
@@ -392,6 +412,7 @@ async fn conflicting_custom_plugin_does_not_prepare_or_change_any_head() {
.revision,
2
);
+ serving.stop().await;
store.drain().await;
}
@@ -423,7 +444,7 @@ fn product_version_covers_authoring_and_each_named_package() {
#[ignore = "requires the built bundled product"]
async fn unchanged_product_records_version_without_rewriting_deployments() {
let directory = tempfile::tempdir().unwrap();
- let (store, ui) = old_installation(directory.path(), true).await;
+ let (store, ui, serving) = old_installation(directory.path(), true).await;
assert!(begin(&store).await.unwrap());
launch::prepare(&store, None, Product::Agent).await.unwrap();
let original = store.application_product().await.unwrap().unwrap();
@@ -469,6 +490,7 @@ async fn unchanged_product_records_version_without_rewriting_deployments() {
.pending
.is_none()
);
+ serving.stop().await;
store.drain().await;
}
diff --git a/products/maka/cli/src/application/serve.rs b/products/maka/cli/src/application/serve.rs
index 9b982e7f26..c80175e4fb 100644
--- a/products/maka/cli/src/application/serve.rs
+++ b/products/maka/cli/src/application/serve.rs
@@ -17,21 +17,70 @@
* under the License.
*/
-//! Compiled product availability for the shared private Host process.
+//! Explicit persisted deployment served by the public Host owner.
use clap::Args;
-use maka_host_runtime::{HostError, application::NativeRegistry};
+use maka_host_runtime::{
+ HostError,
+ application::{
+ NativeRegistry,
+ host::{Host, LifecycleMode, Options},
+ transport::{HostEndpoint, NativeGrant},
+ },
+};
+use serde_json::json;
+use std::{collections::BTreeMap, path::PathBuf, sync::Arc};
+use tokio_util::sync::CancellationToken;
+
#[derive(Args)]
-pub(crate) struct Serve {}
+pub(crate) struct Serve {
+ /// Existing State Root containing the committed Host deployment.
+ #[arg(long)]
+ root: PathBuf,
+ #[arg(long)]
+ deployment: uuid::Uuid,
+ /// Public endpoint selections and their explicit method grants in JSON.
+ #[arg(long, value_name = "FILE")]
+ endpoints: PathBuf,
+ /// Explicit grants for native background calls in JSON.
+ #[arg(long, value_name = "FILE")]
+ native: Option<PathBuf>,
+}
pub(super) fn native() -> Result<NativeRegistry, HostError> {
let mut native = NativeRegistry::default();
maka_product_native::provide(&mut native)?;
Ok(native)
}
-pub(super) async fn run(_: Serve) -> Result<(), HostError> {
- maka_transport::transport::process::serve(
+pub(super) async fn run(args: Serve) -> Result<(), HostError> {
+ let endpoints: BTreeMap<String, HostEndpoint> = super::read_json(&args.endpoints)?;
+ let grants: Vec<NativeGrant> = args
+ .native
+ .as_deref()
+ .map(super::read_json)
+ .transpose()?
+ .unwrap_or_default();
+ let store = Arc::new(super::open_store(args.root).await?);
+ let host = Host::open_with_options(
+ store,
+ args.deployment,
native()?,
Some(maka_host_js::factory(Default::default())),
+ endpoints,
+ grants,
+ Options {
+ lifecycle_mode: LifecycleMode::Service,
+ ..Default::default()
+ },
)
.await
- .map_err(Into::into)
+ .map_err(HostError::from)?;
+ let endpoint = crate::endpoint::LocalEndpoint::bind()?;
+ let cancellation = CancellationToken::new();
+ let _signals = crate::signals::watch(cancellation.clone())?;
+ let registration = host.publish_registration(&endpoint.path, None)?;
+ println!(
+ "{}",
+ json!({"kind":"ready","rootId":host.root_id(),"socketPath":endpoint.path})
+ );
+ let result = crate::serve::listen(endpoint, None, host, cancellation).await;
+ result.and(registration.remove().map_err(Into::into))
}
diff --git a/products/maka/cli/src/args.rs b/products/maka/cli/src/args.rs
index 6bf550b189..d45cb93e99 100644
--- a/products/maka/cli/src/args.rs
+++ b/products/maka/cli/src/args.rs
@@ -107,6 +107,9 @@ enum HostCommand {
#[arg(long)]
allow_interrupt_active_tasks: bool,
},
+ /// Coordinate automatic local startup for a trusted UI launcher.
+ #[command(hide = true)]
+ Ensure(crate::startup::Ensure),
/// Run a discoverable ephemeral Host under its launcher.
Candidate(candidate::Candidate),
/// Run only the currently admitted supervised deployment.
@@ -148,7 +151,7 @@ impl Cli {
}
let finite = matches!(&self.command, Some(Command::Host(command)) if !matches!(command,
HostCommand::Candidate(_) | HostCommand::Serve { .. } | HostCommand::ServiceRun(_)
- | HostCommand::Connect(_)));
+ | HostCommand::Connect(_) | HostCommand::Ensure(_)));
if finite {
let default = if matches!(
&self.command,
@@ -191,6 +194,7 @@ impl Cli {
Command::Host(HostCommand::Fetch(args)) => args.run().await,
Command::Host(HostCommand::Access(args)) => args.run().await,
Command::Host(HostCommand::Candidate(args)) => args.run().await,
+ Command::Host(HostCommand::Ensure(args)) => args.run().await,
Command::Host(HostCommand::Install(args)) => args.run().await,
Command::Host(HostCommand::Setup(args)) => args.run().await,
Command::Host(HostCommand::Activate(args)) => args.run().await,
diff --git a/products/maka/cli/src/deployment.rs b/products/maka/cli/src/deployment.rs
index 7bf1279e19..e9acc09dfa 100644
--- a/products/maka/cli/src/deployment.rs
+++ b/products/maka/cli/src/deployment.rs
@@ -37,7 +37,7 @@ pub(super) use activation::Activate;
pub(super) use connect::Connect;
pub(super) use control::{Control, ControlAction};
pub(super) use entry::ServiceRun;
-pub(super) use local::connect as connect_local;
+pub(super) use local::{connect as connect_local, connect_application};
pub(super) use logs::Logs;
pub(super) use policy::{
Configure as UpdatePolicy,
diff --git a/products/maka/cli/src/deployment/local.rs b/products/maka/cli/src/deployment/local.rs
index 122cfb810a..8589243456 100644
--- a/products/maka/cli/src/deployment/local.rs
+++ b/products/maka/cli/src/deployment/local.rs
@@ -34,10 +34,20 @@ type Connection = Client;
/// Interactive startup does not install, update, repair or take over a Host.
/// The UI calls this only after terminal validation, in a cancellable background job.
pub async fn connect(path: PathBuf) -> Result<Connection, HostError> {
- crate::operation::without_progress(connect_inner(path)).await
+ crate::operation::without_progress(connect_inner(path, None)).await
}
-async fn connect_inner(path: PathBuf) -> Result<Connection, HostError> {
+pub async fn connect_application(
+ request: &maka_transport::startup::Request,
+) -> Result<Connection, HostError> {
+ request.validate()?;
+ crate::operation::without_progress(connect_inner(request.root.clone(), Some(request))).await
+}
+
+async fn connect_inner(
+ path: PathBuf,
+ selection: Option<&maka_transport::startup::Request>,
+) -> Result<Connection, HostError> {
let mut child = None;
let result = tokio::time::timeout(Duration::from_secs(30), async {
let (_launch, root_id) = loop {
@@ -68,7 +78,13 @@ async fn connect_inner(path: PathBuf) -> Result<Connection, HostError> {
Err(error) => return Err(error.into()),
}
};
- if let Some(connection) = live(&path, &root_id).await? {
+ if selection
+ .and_then(|selection| selection.state_domain.as_ref())
+ .is_some_and(|expected| *expected != root_id)
+ {
+ return Err("State Root differs from the selected application Host".into());
+ }
+ if let Some(connection) = live(&path, &root_id, selection).await? {
return Ok(connection);
}
let directory = directory(&root_id)?;
@@ -77,7 +93,7 @@ async fn connect_inner(path: PathBuf) -> Result<Connection, HostError> {
match FileLease::acquire(&directory.join("executor.lock")) {
Ok(lease) => break Arc::new(lease),
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
- if let Some(connection) = live(&path, &root_id).await? {
+ if let Some(connection) = live(&path, &root_id, selection).await? {
return Ok(connection);
}
pause().await;
@@ -86,7 +102,7 @@ async fn connect_inner(path: PathBuf) -> Result<Connection, HostError> {
}
};
lease.validate()?;
- if let Some(connection) = live(&path, &root_id).await? {
+ if let Some(connection) = live(&path, &root_id, selection).await? {
return Ok(connection);
}
match store::read(&directory).await? {
@@ -94,10 +110,11 @@ async fn connect_inner(path: PathBuf) -> Result<Connection, HostError> {
let root = root::resolve(&path)?;
deployment.validate(&root, &directory)?;
deployment.require_active()?;
+ prepare_application(&path, &root_id, selection).await?;
// Keep its probe resident until the actual interactive handshake.
let (probe, _) = activation::connect_or_launch(&deployment, lease.clone()).await?;
let anchor = probe.anchor().await?;
- let connection = live(&path, &root_id)
+ let connection = live(&path, &root_id, selection)
.await?
.ok_or("Activated Host is unavailable")?;
lease.validate()?;
@@ -109,9 +126,10 @@ async fn connect_inner(path: PathBuf) -> Result<Connection, HostError> {
}
store::Installation::Missing => {}
}
+ prepare_application(&path, &root_id, selection).await?;
loop {
lease.validate()?;
- if let Some(connection) = live(&path, &root_id).await? {
+ if let Some(connection) = live(&path, &root_id, selection).await? {
if let Some(child) = &mut child {
release(child).await?;
}
@@ -169,6 +187,36 @@ async fn connect_inner(path: PathBuf) -> Result<Connection, HostError> {
result
}
+async fn prepare_application(
+ path: &std::path::Path,
+ root_id: &str,
+ selection: Option<&maka_transport::startup::Request>,
+) -> Result<(), HostError> {
+ let Some(selection) = selection else {
+ return Ok(());
+ };
+ loop {
+ if live(path, root_id, Some(selection)).await?.is_some() {
+ return Ok(());
+ }
+ let root = match RootOwner::open(path, &RootNamespaces::for_current_account()?) {
+ Ok(root) => root,
+ Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+ pause().await;
+ continue;
+ }
+ Err(error) => return Err(error.into()),
+ };
+ if root.root_id() != root_id {
+ return Err("State Root changed before application preparation".into());
+ }
+ let store = maka_transport::open_store(Arc::new(root)).await?;
+ let result = crate::application::select_committed_host(&store, selection).await;
+ store.drain().await;
+ return result;
+ }
+}
+
async fn pause() {
tokio::time::sleep(Duration::from_millis(25)).await;
}
@@ -186,6 +234,7 @@ async fn release(child: &mut Child) -> Result<(), HostError> {
async fn live(
path: &std::path::Path,
expected_root: &str,
+ selection: Option<&maka_transport::startup::Request>,
) -> Result<Option<Connection>, HostError> {
// A candidate has not published discovery until database initialization is
// complete. Do not inspect its transient SQLite startup files as a live root.
@@ -206,6 +255,10 @@ async fn live(
if discovery.root_id != expected_root {
return Err("State Root changed during startup".into());
}
+ if selection.is_some_and(|selection| discovery.source.deployment != selection.deployment) {
+ return Err("live Host belongs to another application deployment".into());
+ }
+
let stream = match ipc::connect(&discovery.endpoint).await {
Ok(stream) => stream,
Err(error) if error.downcast_ref::<io::Error>().is_some_and(absent) => return Ok(None),
diff --git a/products/maka/cli/src/main.rs b/products/maka/cli/src/main.rs
index edcd7370f2..6ad939067b 100644
--- a/products/maka/cli/src/main.rs
+++ b/products/maka/cli/src/main.rs
@@ -34,6 +34,7 @@ mod operation;
mod sandbox;
mod serve;
mod signals;
+mod startup;
#[cfg(windows)]
mod windows;
diff --git a/products/maka/cli/src/startup.rs b/products/maka/cli/src/startup.rs
new file mode 100644
index 0000000000..137339096b
--- /dev/null
+++ b/products/maka/cli/src/startup.rs
@@ -0,0 +1,89 @@
+/*
+ * 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 clap::Args;
+use maka_host_runtime::{
+ HostError,
+ application::{host::discovery, transport::wire},
+};
+use maka_transport::startup::{REQUEST_LIMIT, Ready, Request};
+use std::time::Duration;
+use tokio::io::AsyncReadExt;
+use tokio_util::sync::CancellationToken;
+
+/// Internal startup coordination for trusted TUI/Desktop launchers.
+#[derive(Args)]
+pub(crate) struct Ensure {}
+impl Ensure {
+ pub(crate) async fn run(self) -> Result<(), HostError> {
+ let maka_transport::stdio::Stdio {
+ input,
+ mut output,
+ output_done,
+ } = maka_transport::stdio::Stdio::open()?;
+ let mut input = wire::Reader::new(input);
+ let request: Request =
+ tokio::time::timeout(Duration::from_secs(5), input.read_bounded(REQUEST_LIMIT))
+ .await??;
+ request.validate()?;
+ let stop = CancellationToken::new();
+ let _signals = crate::signals::watch(stop.clone())?;
+ let client = tokio::select! {
+ result = crate::deployment::connect_application(&request) => result?,
+ _ = stop.cancelled() => return Ok(()),
+ byte = input.inner.read_u8() => {
+ if byte.is_ok() { return Err("unexpected Host startup owner input".into()); }
+ return Ok(());
+ }
+ };
+ let registered = discovery::read(&request.root)?;
+ if registered.source.deployment != request.deployment
+ || registered.host_epoch != client.epoch()
+ {
+ return Err("application Host changed during startup admission".into());
+ }
+ registered.verify(
+ &maka_host_runtime::application::host::Client::status(
+ ®istered.endpoint,
+ Some(client.epoch()),
+ )
+ .await?,
+ )?;
+ wire::write_bounded(
+ &mut output,
+ &Ready {
+ root_id: registered.root_id,
+ epoch: client.epoch(),
+ },
+ 4096,
+ )
+ .await?;
+ tokio::select! {
+ _ = stop.cancelled() => {},
+ byte = input.inner.read_u8() => {
+ if byte.is_ok() { return Err("unexpected Host startup owner input".into()); }
+ }
+ }
+ // This probe owns no UI operation or Root; EOF releases only residency.
+ drop(client);
+ drop(output);
+ tokio::time::timeout(Duration::from_secs(5), output_done).await???;
+ Ok(())
+ }
+}
diff --git a/products/maka/cli/tests/integration/terminal/application.rs b/products/maka/cli/tests/integration/terminal/application.rs
index 3cbe88c8bc..aa7bdfc23b 100644
--- a/products/maka/cli/tests/integration/terminal/application.rs
+++ b/products/maka/cli/tests/integration/terminal/application.rs
@@ -270,7 +270,89 @@ fn plugin_application_setup_rejection_is_clean_and_diagnostics_cannot_emit_escap
#[allow(dead_code)]
#[path = "../../../../../../platform/transport/tests/support/mod.rs"]
mod local_fixture;
-fn local_arguments(directory: &std::path::Path, unrelated_failure: bool) -> Vec<String> {
+/// Tests retain the business Host independently of every foreground UI. This
+/// guard also retires product auto-started Hosts before temporary Roots vanish.
+pub(super) struct HostRootGuard(std::path::PathBuf);
+impl HostRootGuard {
+ pub(super) fn new(root: &std::path::Path) -> Self {
+ Self(root.to_path_buf())
+ }
+ pub(super) fn stop(&self) {
+ self.retire().unwrap();
+ }
+ fn retire(&self) -> Result<(), String> {
+ use maka_host_runtime::application::host::{Client, discovery};
+ if !self.0.exists() {
+ return Ok(());
+ }
+ let discovery = match discovery::read(&self.0) {
+ Ok(discovery) => Some(discovery),
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+ Err(error) => return Err(error.to_string()),
+ };
+ if let Some(discovery) = discovery {
+ let runtime = tokio::runtime::Runtime::new().map_err(|error| error.to_string())?;
+ runtime.block_on(async {
+ // A startup failure may already be draining. Its disappearing
+ // status channel is not cleanup proof; the Root lease below is.
+ if let Ok(status) =
+ Client::status(&discovery.endpoint, Some(discovery.host_epoch)).await
+ && !status.closing
+ && !status.closed
+ {
+ discovery
+ .verify(&status)
+ .map_err(|error| error.to_string())?;
+ if !Client::retire_with_options(
+ &discovery.endpoint,
+ discovery.host_epoch,
+ true,
+ false,
+ )
+ .await?
+ {
+ return Err("Host did not accept explicit retirement".into());
+ }
+ }
+ Ok::<_, String>(())
+ })?;
+ }
+ let namespaces = maka_event_log::root::RootNamespaces::for_current_account()
+ .map_err(|error| error.to_string())?;
+ let deadline = Instant::now() + Duration::from_secs(15);
+ loop {
+ match maka_event_log::root::RootOwner::open(&self.0, &namespaces) {
+ Ok(owner) => {
+ drop(owner);
+ return Ok(());
+ }
+ Err(error)
+ if error.kind() == std::io::ErrorKind::WouldBlock
+ && Instant::now() < deadline =>
+ {
+ std::thread::park_timeout(Duration::from_millis(5));
+ }
+ Err(error) => return Err(error.to_string()),
+ }
+ }
+ }
+}
+impl Drop for HostRootGuard {
+ fn drop(&mut self) {
+ if let Err(error) = self.retire() {
+ if !std::thread::panicking() {
+ panic!("Host cleanup failed: {error}");
+ }
+ eprintln!("Host cleanup after test failure: {error}");
+ }
+ }
+}
+
+fn local_arguments(
+ directory: &std::path::Path,
+ unrelated_failure: bool,
+ automatic: bool,
+) -> (Vec<String>, Option<std::process::Child>) {
let timer = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_time()
@@ -306,32 +388,112 @@ fn local_arguments(directory: &std::path::Path, unrelated_failure: bool) -> Vec<
}
});
let path = directory.join("connections.json");
+ let endpoints = serde_json::json!({"business":{"consumer":local_fixture::identity(13),"import":"business","grant":{"methods":["read","write"],"recovery":false}},"receipts":{"consumer":local_fixture::identity(13),"import":"receipts","grant":{"methods":["read"],"recovery":true}}});
+ let endpoints_path = directory.join("endpoints.json");
+ std::fs::write(&endpoints_path, serde_json::to_vec(&endpoints).unwrap()).unwrap();
std::fs::write(&path,serde_json::to_vec(&serde_json::json!({"hosts":[{
- "name":"local","location":{"kind":"process","root":directory.join("host")},"deployment":host,"endpoints":{"business":{"consumer":local_fixture::identity(13),"import":"business","grant":{"methods":["read","write"],"recovery":false}},"receipts":{"consumer":local_fixture::identity(13),"import":"receipts","grant":{"methods":["read"],"recovery":true}}}
+ "name":"local","location":{"kind":"public","root":directory.join("host")},"deployment":host,"endpoints":endpoints
}]})).unwrap()).unwrap();
- vec![
- "application".into(),
- "run".into(),
- "--root".into(),
- directory.join("caller").to_str().unwrap().into(),
- "--deployment".into(),
- caller.to_string(),
- "--connections".into(),
- path.to_str().unwrap().into(),
- ]
+ let child = if automatic {
+ None
+ } else {
+ let mut child = Command::new(env!("CARGO_BIN_EXE_maka"))
+ .args(["application", "serve", "--root"])
+ .arg(directory.join("host"))
+ .args(["--deployment", &host.to_string(), "--endpoints"])
+ .arg(endpoints_path)
+ .stdin(std::process::Stdio::null())
+ .stdout(std::process::Stdio::piped())
+ .spawn()
+ .unwrap();
+ use std::io::BufRead;
+ let mut stdout = std::io::BufReader::new(child.stdout.take().unwrap());
+ let mut poll = libc::pollfd {
+ fd: stdout.get_ref().as_raw_fd(),
+ events: libc::POLLIN,
+ revents: 0,
+ };
+ assert!(
+ unsafe { libc::poll(&mut poll, 1, 30_000) } > 0,
+ "public Host did not become ready"
+ );
+ let mut line = String::new();
+ stdout.read_line(&mut line).unwrap();
+ let ready: serde_json::Value = serde_json::from_str(&line).unwrap();
+ assert_eq!(ready["kind"], "ready");
+ Some(child)
+ };
+ (
+ vec![
+ "application".into(),
+ "run".into(),
+ "--root".into(),
+ directory.join("caller").to_str().unwrap().into(),
+ "--deployment".into(),
+ caller.to_string(),
+ "--connections".into(),
+ path.to_str().unwrap().into(),
+ ],
+ child,
+ )
}
#[test]
fn production_terminal_connections_execute_host_mutation_and_read_original_receipt_before_exit() {
- check_terminal_connections(false);
+ check_terminal_connections(false, false);
}
#[test]
fn production_terminal_connections_remain_available_after_independent_host_startup_failure() {
- check_terminal_connections(true);
+ check_terminal_connections(true, false);
+}
+#[test]
+fn generic_terminal_automatically_starts_its_committed_host_and_reuses_it_on_reopen() {
+ check_terminal_connections(false, true);
}
-fn check_terminal_connections(unrelated_failure: bool) {
+#[test]
+fn default_terminal_exit_stops_its_last_client_host_process_and_releases_both_roots() {
let directory = tempfile::tempdir().unwrap();
- let args = local_arguments(directory.path(), unrelated_failure);
- let args = args.iter().map(String::as_str).collect::<Vec<_>>();
+ let host_root = directory.path().join("host");
+ let _guard = HostRootGuard::new(&host_root);
+ // A service Host has no idle retirement, so exit proves the UI requested it.
+ let (arguments, child) = local_arguments(directory.path(), false, false);
+ let mut child = child.unwrap();
+ let args = arguments.iter().map(String::as_str).collect::<Vec<_>>();
+ let mut tui = Pty::spawn(&args);
+ tui.wait_for("Read Host");
+ tui.send(b"\r");
+ tui.wait_for("\"effects\":0");
+ tui.send(b"\x11");
+ tui.finish();
+ let deadline = Instant::now() + Duration::from_secs(10);
+ loop {
+ if let Some(status) = child.try_wait().unwrap() {
+ assert!(status.success(), "last-client Host exit failed: {status}");
+ break;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "default UI exit left its last-client Host running"
+ );
+ std::thread::park_timeout(Duration::from_millis(5));
+ }
+ let namespaces = maka_event_log::root::RootNamespaces::for_current_account().unwrap();
+ drop(
+ maka_event_log::root::RootOwner::open(&directory.path().join("caller"), &namespaces)
+ .unwrap(),
+ );
+ drop(maka_event_log::root::RootOwner::open(&host_root, &namespaces).unwrap());
+}
+
+fn check_terminal_connections(unrelated_failure: bool, automatic: bool) {
+ let directory = tempfile::tempdir().unwrap();
+ let host_root = directory.path().join("host");
+ let guard = HostRootGuard::new(&host_root);
+ let (arguments, mut host_child) =
+ local_arguments(directory.path(), unrelated_failure, automatic);
+ let args = arguments.iter().map(String::as_str).collect::<Vec<_>>();
+ if automatic {
+ assert!(maka_host_runtime::application::host::discovery::read(&host_root).is_err());
+ }
let mut tui = Pty::spawn(&args);
if unrelated_failure {
tui.wait_for("unrelated initializer rejected");
@@ -339,6 +501,15 @@ fn check_terminal_connections(unrelated_failure: bool) {
tui.send(b"\r");
}
tui.wait_for("Read Host");
+ let discovery = maka_host_runtime::application::host::discovery::read(&host_root).unwrap();
+ let runtime = tokio::runtime::Runtime::new().unwrap();
+ // Default UI exit must leave the Host available to this independent client.
+ let keepalive = runtime
+ .block_on(maka_host_runtime::application::host::Client::connect(
+ &discovery.endpoint,
+ Some(discovery.host_epoch),
+ ))
+ .unwrap();
tui.send(b"\r");
tui.wait_for("\"effects\":0");
tui.send(b"\t\r");
@@ -349,7 +520,104 @@ fn check_terminal_connections(unrelated_failure: bool) {
tui.finish();
let ns = maka_event_log::root::RootNamespaces::for_current_account().unwrap();
drop(maka_event_log::root::RootOwner::open(&directory.path().join("caller"), &ns).unwrap());
- drop(maka_event_log::root::RootOwner::open(&directory.path().join("host"), &ns).unwrap());
+ if let Some(child) = &mut host_child {
+ assert!(
+ child.try_wait().unwrap().is_none(),
+ "UI exit stopped the independent Host"
+ );
+ }
+ let error = maka_event_log::root::RootOwner::open(&host_root, &ns)
+ .err()
+ .expect("Host must retain its original Root");
+ assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock);
+ runtime.block_on(async {
+ use maka_host_runtime::application::host::Client;
+ use maka_host_runtime::application::transport::{Call, Reply, Source};
+ let status = Client::status(&discovery.endpoint, Some(discovery.host_epoch))
+ .await
+ .unwrap();
+ discovery.verify(&status).unwrap();
+ assert!(!status.closing && !status.closed);
+ assert_eq!(status.sessions, 1);
+ let caller = std::sync::Arc::new(
+ maka_event_log::root::RootOwner::open(&directory.path().join("caller"), &ns).unwrap(),
+ );
+ let store = maka_event_log::EventLog::for_root(caller).await.unwrap();
+ let caller_id = arguments[5].parse().unwrap();
+ let intents = store
+ .application_intents(caller_id, local_fixture::identity(2), 0, 10)
+ .await
+ .unwrap();
+ assert_eq!(intents.len(), 1, "one original UI write, never replayed");
+ let intent = &intents[0];
+ let configuration: serde_json::Value = serde_json::from_slice(
+ &std::fs::read(directory.path().join("connections.json")).unwrap(),
+ )
+ .unwrap();
+ let endpoints =
+ serde_json::from_value(configuration["hosts"][0]["endpoints"].clone()).unwrap();
+ let mut peer = Client::connect_peer(
+ &discovery.endpoint,
+ Some(discovery.host_epoch),
+ intent.caller.clone(),
+ [intent.consumer.clone()].into(),
+ endpoints,
+ )
+ .await
+ .unwrap();
+ let channel = peer.open_call_channel().await.unwrap();
+ let (reader, writer) = tokio::io::split(channel);
+ let mut read = Call::new(
+ reader,
+ writer,
+ peer.catalogue().select("receipts").unwrap(),
+ Source {
+ deployment: intent.caller.clone(),
+ consumer: intent.consumer.clone(),
+ mount: uuid::Uuid::new_v4(),
+ request: uuid::Uuid::new_v4(),
+ },
+ "read".into(),
+ serde_json::to_value(&intent.reference).unwrap(),
+ Default::default(),
+ );
+ read.start().await.unwrap();
+ let Reply::Finished {
+ result: Ok(value),
+ cleanup: Ok(()),
+ } = read.receive().await.unwrap()
+ else {
+ panic!("original receipt was not readable after UI exit");
+ };
+ assert_eq!(
+ value["reference"],
+ serde_json::to_value(&intent.reference).unwrap()
+ );
+ assert_eq!(value["receipt"]["kind"], "recorded");
+ assert_eq!(value["receipt"]["outcome"]["value"]["effects"], 1);
+ peer.close().await.unwrap();
+ store.close().await.unwrap();
+ });
+ let mut reopened = Pty::spawn(&args);
+ if unrelated_failure {
+ reopened.wait_for("unrelated initializer rejected");
+ reopened.wait_for("open recovery interface");
+ reopened.send(b"\r");
+ }
+ reopened.wait_for("Read Host");
+ reopened.send(b"\r");
+ reopened.wait_for("\"effects\":1");
+ reopened.send(b"\x11");
+ reopened.finish();
+ if let Some(child) = &mut host_child {
+ assert!(child.try_wait().unwrap().is_none());
+ }
+ drop(keepalive);
+ guard.stop();
+ if let Some(mut child) = host_child {
+ assert!(child.wait().unwrap().success());
+ }
+ drop(maka_event_log::root::RootOwner::open(&host_root, &ns).unwrap());
let snapshot = tui.screen.snapshot().unwrap();
assert!(!snapshot.alternate_screen);
assert!(snapshot.cursor.visible);
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product.rs b/products/maka/cli/tests/integration/terminal/bundled_product.rs
index f67b31c00e..3703e3b020 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product.rs
@@ -17,6 +17,7 @@
* under the License.
*/
+use super::application::HostRootGuard;
use super::*;
use maka_event_log::root::{RootNamespaces, RootOwner};
use serde_json::json;
@@ -52,6 +53,7 @@ fn bundled_product_launch_initializes_and_reopens_original_roots_without_node()
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let host = directory.path().join("host");
+ let host_guard = HostRootGuard::new(&host);
let ui = directory.path().join("ui");
let configuration = directory.path().join("installation.json");
std::fs::write(
@@ -70,6 +72,9 @@ fn bundled_product_launch_initializes_and_reopens_original_roots_without_node()
)
.unwrap();
let namespaces = RootNamespaces::for_current_account().unwrap();
+ let runtime = tokio::runtime::Runtime::new().unwrap();
+ let mut anchor = None;
+ let mut host_epoch = None;
for first in [true, false] {
let mut args = vec!["application", "launch", "--root", ui.to_str().unwrap()];
if first {
@@ -92,6 +97,20 @@ fn bundled_product_launch_initializes_and_reopens_original_roots_without_node()
"bundled launch first={first}: first frame in {:?}",
started.elapsed()
);
+ if first {
+ // An independent client keeps the shared Host resident across UI
+ // exits and the offline UI edits between these two launches.
+ let discovery = maka_host_runtime::application::host::discovery::read(&host).unwrap();
+ host_epoch = Some(discovery.host_epoch);
+ anchor = Some(
+ runtime
+ .block_on(maka_host_runtime::application::host::Client::connect(
+ &discovery.endpoint,
+ Some(discovery.host_epoch),
+ ))
+ .unwrap(),
+ );
+ }
tui.send(b"\x0b");
tui.wait_for("Close commands");
// The first character after a newly visible dialog must survive its
@@ -113,7 +132,6 @@ fn bundled_product_launch_initializes_and_reopens_original_roots_without_node()
// No authoring input remains for reopening; the record selects the
// committed deployment, whose user-updated head must remain authoritative.
std::fs::remove_file(&configuration).unwrap();
- let runtime = tokio::runtime::Runtime::new().unwrap();
runtime.block_on(async {
let owner = std::sync::Arc::new(RootOwner::open(&ui, &namespaces).unwrap());
let store = maka_event_log::EventLog::for_root(owner).await.unwrap();
@@ -147,7 +165,6 @@ fn bundled_product_launch_initializes_and_reopens_original_roots_without_node()
store.close().await.unwrap();
});
} else {
- let runtime = tokio::runtime::Runtime::new().unwrap();
runtime.block_on(async {
let owner = std::sync::Arc::new(RootOwner::open(&ui, &namespaces).unwrap());
let store = maka_event_log::EventLog::for_root(owner).await.unwrap();
@@ -166,10 +183,31 @@ fn bundled_product_launch_initializes_and_reopens_original_roots_without_node()
store.close().await.unwrap();
});
}
- for root in [&host, &ui] {
- drop(RootOwner::open(root, &namespaces).unwrap());
+ drop(RootOwner::open(&ui, &namespaces).unwrap());
+ let error = RootOwner::open(&host, &namespaces)
+ .err()
+ .expect("UI exit must leave the business Host resident");
+ assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock);
+ let discovery = maka_host_runtime::application::host::discovery::read(&host).unwrap();
+ if let Some(epoch) = host_epoch {
+ assert_eq!(discovery.host_epoch, epoch);
}
+ host_epoch = Some(discovery.host_epoch);
+ runtime.block_on(async {
+ let status = maka_host_runtime::application::host::Client::status(
+ &discovery.endpoint,
+ Some(discovery.host_epoch),
+ )
+ .await
+ .unwrap();
+ discovery.verify(&status).unwrap();
+ });
}
+ runtime.block_on(async {
+ anchor.take().unwrap().close().await.unwrap();
+ });
+ host_guard.stop();
+ drop(RootOwner::open(&host, &namespaces).unwrap());
}
#[test]
@@ -180,6 +218,7 @@ fn bundled_product_launch_recovers_partial_commits_without_reauthoring() {
use std::sync::Arc;
let directory = tempfile::tempdir().unwrap();
let host = directory.path().join("host");
+ let host_guard = HostRootGuard::new(&host);
let ui = directory.path().join("ui");
let configuration = directory.path().join("installation.json");
std::fs::write(
@@ -227,6 +266,7 @@ fn bundled_product_launch_recovers_partial_commits_without_reauthoring() {
"{}",
String::from_utf8_lossy(&failed.stderr)
);
+ host_guard.stop();
let (ui_id, host_id) = runtime.block_on(async {
let owner = Arc::new(RootOwner::open(&ui, &namespaces).unwrap());
let store = EventLog::for_root(owner.clone()).await.unwrap();
@@ -243,7 +283,7 @@ fn bundled_product_launch_recovers_partial_commits_without_reauthoring() {
.as_array()
.unwrap()
.iter()
- .find(|host| host["location"]["kind"] == "process")
+ .find(|host| host["location"]["kind"] == "public")
.unwrap()["deployment"]
.as_str()
.unwrap()
@@ -309,6 +349,7 @@ fn bundled_product_launch_recovers_partial_commits_without_reauthoring() {
);
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
runtime.block_on(async {
let owner = Arc::new(RootOwner::open(&ui, &namespaces).unwrap());
let store = EventLog::for_root(owner).await.unwrap();
@@ -355,6 +396,7 @@ fn bundled_ui_authoring_installs_and_reloads_original_source_without_replaying()
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let host = directory.path().join("host");
+ let host_guard = HostRootGuard::new(&host);
let ui = directory.path().join("ui");
let configuration = directory.path().join("configuration.json");
std::fs::write(&configuration,serde_json::to_vec(&json!({
@@ -375,6 +417,7 @@ fn bundled_ui_authoring_installs_and_reloads_original_source_without_replaying()
);
first.send(b"\x11");
first.finish();
+ host_guard.stop();
let runtime = tokio::runtime::Runtime::new().unwrap();
let namespaces = RootNamespaces::for_current_account().unwrap();
let namespace = Namespace::new("maka.application.authoring", Scope::Profile).unwrap();
@@ -434,6 +477,7 @@ fn bundled_ui_authoring_installs_and_reloads_original_source_without_replaying()
runtime.block_on(sql.close()).unwrap();
resumed.send(b"\x11");
resumed.finish();
+ host_guard.stop();
runtime.block_on(async {
let store = EventLog::for_root(Arc::new(RootOwner::open(&host, &namespaces).unwrap()))
.await
@@ -493,6 +537,7 @@ fn bundled_ui_authoring_installs_and_reloads_original_source_without_replaying()
);
restored_ui.send(b"\x11");
restored_ui.finish();
+ host_guard.stop();
runtime.block_on(async {
let store = EventLog::for_root(Arc::new(RootOwner::open(&host, &namespaces).unwrap()))
.await
@@ -515,6 +560,7 @@ fn bundled_ui_authoring_installs_and_reloads_original_source_without_replaying()
fn first_install_trust_uses_one_terminal_and_cancel_has_no_product_effects() {
let directory = tempfile::tempdir().unwrap();
let root = directory.path().join("ui");
+ let _host_guard = HostRootGuard::new(&root.join("host"));
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let args = [
@@ -563,6 +609,7 @@ fn first_install_trust_uses_one_terminal_and_cancel_has_no_product_effects() {
fn first_install_failure_keeps_original_host_error_after_terminal_restoration() {
let directory = tempfile::tempdir().unwrap();
let root = directory.path().join("ui");
+ let _host_guard = HostRootGuard::new(&root.join("host"));
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let mut tui = Pty::spawn(&[
@@ -626,6 +673,7 @@ fn first_install_failure_keeps_original_host_error_after_terminal_restoration()
#[ignore = "starts the bundled product and checks connection settings in a native PTY"]
fn bundled_connections_search_scroll_click_and_chatgpt_credentials() {
let directory = tempfile::tempdir().unwrap();
+ let _host_guard = HostRootGuard::new(&directory.path().join("host"));
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let configuration = directory.path().join("installation.json");
@@ -729,6 +777,7 @@ fn bundled_connections_search_scroll_click_and_chatgpt_credentials() {
#[ignore = "checks live language changes and persistence in the real product PTY"]
fn bundled_language_switch_updates_and_remains_responsive() {
let directory = tempfile::tempdir().unwrap();
+ let _host_guard = HostRootGuard::new(&directory.path().join("host"));
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let configuration = directory.path().join("installation.json");
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat.rs
index d6690275f3..e557a3feb9 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat.rs
@@ -272,6 +272,7 @@ fn bundled_chat_sends_follows_renders_markdown_and_scrolls_original_history() {
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let host = directory.path().join("host");
+ let host_guard = HostRootGuard::new(&host);
let ui = directory.path().join("ui");
let configuration = directory.path().join("installation.json");
let namespaces = RootNamespaces::for_current_account().unwrap();
@@ -410,7 +411,169 @@ fn bundled_chat_sends_follows_renders_markdown_and_scrolls_original_history() {
evidence(&tui, "restored-following");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
server.join().unwrap();
assert!(!tui.screen.snapshot().unwrap().alternate_screen);
drop(RootOwner::open(&host, &namespaces).unwrap());
}
+
+#[test]
+#[ignore = "real Agent invocation continues after explicitly closing only its UI"]
+fn ui_only_exit_preserves_accepted_agent_invocation_and_reopens_without_resend() {
+ use maka_host_runtime::application::host::{Client, discovery};
+ use sqlx::{Connection, SqliteConnection, sqlite::SqliteConnectOptions};
+ const PROMPT: &str = "Finish this original invocation after the UI closes.";
+ const ANSWER: &str = "Original background job finished.";
+ let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+ let base = format!("http://{}/v1", listener.local_addr().unwrap());
+ let (requests, received) = mpsc::channel();
+ let server = model_server(listener, requests, 1, 0);
+ let directory = tempfile::tempdir().unwrap();
+ let workspace = directory.path().join("workspace");
+ std::fs::create_dir(&workspace).unwrap();
+ let host = directory.path().join("host");
+ let host_guard = HostRootGuard::new(&host);
+ let ui = directory.path().join("ui");
+ let configuration = directory.path().join("installation.json");
+ let namespaces = RootNamespaces::for_current_account().unwrap();
+ let runtime = tokio::runtime::Runtime::new().unwrap();
+ runtime.block_on(seed_connection(&host, &namespaces, &base, json!({})));
+ std::fs::write(
+ &configuration,
+ serde_json::to_vec(&json!({
+ "presentation":"tui", "locale":"en", "scheme":"dark", "hostRoot":host,
+ "agent":{
+ "workspaceRoots":[{"label":"Workspace","path":workspace}], "tools":[],
+ "automaticTitles":false,
+ "permissions":{"access":"read_only","requestPermissions":true}
+ }
+ }))
+ .unwrap(),
+ )
+ .unwrap();
+ let mut tui = Pty::spawn(&[
+ "application",
+ "launch",
+ "--root",
+ ui.to_str().unwrap(),
+ "--configuration",
+ configuration.to_str().unwrap(),
+ ]);
+ tui.wait_until_with_timeout(|s| s.contains("Settings"), Duration::from_secs(60));
+ tui.send(b"\x0b");
+ tui.wait_for("Close commands");
+ tui.send(b"New conversation");
+ tui.wait_for("› New conversation");
+ tui.send(b"\r");
+ tui.wait_for("Message…");
+ click(&mut tui, "Message…");
+ paste(&mut tui, PROMPT);
+ tui.wait_for(PROMPT);
+ tui.send(b"\r");
+ let (sent, mut socket) = next_request(&mut tui, &received);
+ assert_eq!(sent["model"], "fixture-model");
+ assert!(sent.to_string().contains(PROMPT));
+ assert!(!is_title(&sent));
+ server.join().unwrap();
+
+ // Receiving this HTTP request proves admission reached the real model
+ // executor. Its answer is physically gated until the foreground has exited.
+ let (release, gated) = mpsc::sync_channel::<()>(0);
+ let responder = std::thread::spawn(move || {
+ gated.recv_timeout(Duration::from_secs(60)).unwrap();
+ delta(&mut socket, ANSWER);
+ complete(&mut socket);
+ });
+ let original = discovery::read(&host).unwrap();
+ let mut sql = runtime
+ .block_on(SqliteConnection::connect_with(
+ &SqliteConnectOptions::new()
+ .filename(host.join(maka_event_log::root::ROOT_DATABASE))
+ .read_only(true),
+ ))
+ .unwrap();
+ let opened: Vec<String> = runtime.block_on(
+ sqlx::query_scalar("SELECT invocation_id FROM event_log WHERE kind='invocation_opened' ORDER BY sequence")
+ .fetch_all(&mut sql),
+ ).unwrap();
+ assert_eq!(opened.len(), 1);
+ let invocation = &opened[0];
+ let ended: i64 = runtime
+ .block_on(
+ sqlx::query_scalar(
+ "SELECT COUNT(*) FROM event_log WHERE invocation_id=? AND kind='invocation_ended'",
+ )
+ .bind(invocation)
+ .fetch_one(&mut sql),
+ )
+ .unwrap();
+ assert_eq!(
+ ended, 0,
+ "the explicit model gate must still hold the original invocation"
+ );
+ tui.send(b"\x1b\x11");
+ tui.finish();
+ assert!(!tui.screen.snapshot().unwrap().alternate_screen);
+ drop(RootOwner::open(&ui, &namespaces).unwrap());
+ let error = RootOwner::open(&host, &namespaces)
+ .err()
+ .expect("business Host must remain resident");
+ assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock);
+ runtime.block_on(async {
+ let status = Client::status(&original.endpoint, Some(original.host_epoch))
+ .await
+ .unwrap();
+ original.verify(&status).unwrap();
+ assert_eq!(
+ status.sessions, 0,
+ "foreground detachment must finish before releasing the answer"
+ );
+ });
+ release.send(()).unwrap();
+ responder.join().unwrap();
+
+ // Observe durable completion before starting another UI. Reconnection must
+ // only read this result, never resume or resend the accepted model request.
+ runtime.block_on(async {
+ tokio::time::timeout(Duration::from_secs(30), async {
+ loop {
+ let ended: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM event_log WHERE invocation_id=? AND kind='invocation_ended'")
+ .bind(invocation).fetch_one(&mut sql).await.unwrap();
+ if ended == 1 { break; }
+ tokio::task::yield_now().await;
+ }
+ }).await.expect("original invocation did not complete without its UI");
+ let completed: Vec<String> = sqlx::query_scalar("SELECT event_json FROM event_log WHERE invocation_id=? AND kind='model_completed'")
+ .bind(invocation).fetch_all(&mut sql).await.unwrap();
+ assert_eq!(completed.len(), 1);
+ assert!(completed[0].contains(ANSWER), "persisted output belongs to the original invocation");
+ });
+ let mut reopened = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
+ reopened.wait_until_with_timeout(|screen| screen.contains(ANSWER), Duration::from_secs(45));
+ let current = discovery::read(&host).unwrap();
+ assert_eq!(current.host_epoch, original.host_epoch);
+ assert_eq!(current.source, original.source);
+ let after: Vec<String> = runtime.block_on(
+ sqlx::query_scalar("SELECT invocation_id FROM event_log WHERE kind='invocation_opened' ORDER BY sequence")
+ .fetch_all(&mut sql),
+ ).unwrap();
+ assert_eq!(
+ after, opened,
+ "reopening must preserve the original invocation identity"
+ );
+ let attempts: i64 = runtime
+ .block_on(
+ sqlx::query_scalar("SELECT COUNT(*) FROM event_log WHERE kind='model_requested'")
+ .fetch_one(&mut sql),
+ )
+ .unwrap();
+ assert_eq!(
+ attempts, 1,
+ "UI reconnection must not resend the model request"
+ );
+ reopened.send(b"\x11");
+ reopened.finish();
+ runtime.block_on(sql.close()).unwrap();
+ host_guard.stop();
+ drop(RootOwner::open(&host, &namespaces).unwrap());
+}
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/code_mode.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/code_mode.rs
index f2caaf7c5b..86f330a574 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/code_mode.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/code_mode.rs
@@ -55,6 +55,7 @@ fn default_product_restores_accepted_code_mode_and_original_native_effect() {
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
let host = ui.join("host");
+ let host_guard = HostRootGuard::new(&host);
let namespaces = RootNamespaces::for_current_account().unwrap();
// The standard Host is nested inside an already owned UI Root.
drop(RootOwner::create(&ui, &namespaces).unwrap());
@@ -122,6 +123,7 @@ fn default_product_restores_accepted_code_mode_and_original_native_effect() {
assert!(received.try_recv().is_err());
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
// Change the next-message policy only after the original product has closed.
runtime.block_on(async {
@@ -194,6 +196,7 @@ fn default_product_restores_accepted_code_mode_and_original_native_effect() {
);
restored.send(b"\x11");
restored.finish();
+ host_guard.stop();
server.join().unwrap();
runtime.block_on(async {
sql.close().await.unwrap();
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/computer.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/computer.rs
index 0157245b5b..e1b3380aee 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/computer.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/computer.rs
@@ -31,6 +31,7 @@ fn default_product_computer_approval_and_restart_keep_grant_but_reset_vm() {
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
let host = ui.join("host");
+ let host_guard = HostRootGuard::new(&host);
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -111,6 +112,7 @@ fn default_product_computer_approval_and_restart_keep_grant_but_reset_vm() {
tui.wait_for("Computer access is ready.");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
// Reopen the saved default composition and original Session in a new Host.
let mut tui = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
tui.wait_until_with_timeout(
@@ -155,6 +157,7 @@ fn default_product_computer_approval_and_restart_keep_grant_but_reset_vm() {
evidence(&tui, "computer-restarted");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
drop(received);
server.join().unwrap();
runtime.block_on(async {
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/goal.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/goal.rs
index 98ef746e74..e63613142f 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/goal.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/goal.rs
@@ -63,6 +63,7 @@ fn default_product_goal_preserves_approval_and_resumes_original_objective_after_
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
+ let host_guard = HostRootGuard::new(&ui.join("host"));
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -110,6 +111,7 @@ fn default_product_goal_preserves_approval_and_resumes_original_objective_after_
tui.wait_for("Retain this independent draft.");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
let mut tui = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
tui.wait_until_with_timeout(|s| s.contains("Settings"), Duration::from_secs(60));
open(&mut tui);
@@ -163,5 +165,6 @@ fn default_product_goal_preserves_approval_and_resumes_original_objective_after_
tui.wait_for("Retain this independent draft.");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
server.join().unwrap();
}
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/graph.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/graph.rs
index 4898d734e5..fe54b7fde8 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/graph.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/graph.rs
@@ -49,6 +49,7 @@ fn default_product_graph_delegates_and_keeps_separate_conversation_epochs() {
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
+ let host_guard = HostRootGuard::new(&ui.join("host"));
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -187,6 +188,7 @@ fn default_product_graph_delegates_and_keeps_separate_conversation_epochs() {
assert!(received.try_recv().is_err());
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
let mut tui = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
tui.wait_until_with_timeout(|s| s.contains("Settings"), Duration::from_secs(60));
command(&mut tui, "WorkHub");
@@ -199,5 +201,6 @@ fn default_product_graph_delegates_and_keeps_separate_conversation_epochs() {
assert!(received.try_recv().is_err());
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
server.join().unwrap();
}
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/health.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/health.rs
index 83070e91f9..ea460d6780 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/health.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/health.rs
@@ -73,6 +73,7 @@ fn default_product_health_checks_process_and_independent_endpoint() {
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
let host = ui.join("host");
+ let host_guard = HostRootGuard::new(&host);
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -187,6 +188,7 @@ fn default_product_health_checks_process_and_independent_endpoint() {
evidence(&tui, "health-completed");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
let mut tui = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
tui.wait_until_with_timeout(
|s| s.contains("Process finished; endpoint independently responded."),
@@ -241,5 +243,6 @@ fn default_product_health_checks_process_and_independent_endpoint() {
evidence(&tui, "health-restarted");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
server.join().unwrap();
}
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/performance.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/performance.rs
index c07ef69236..9a5b70305d 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/performance.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/performance.rs
@@ -31,6 +31,7 @@ fn default_product_input_continues_while_the_original_stdout_owner_is_blocked()
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let host = directory.path().join("host");
+ let host_guard = HostRootGuard::new(&host);
let ui = directory.path().join("ui");
let configuration = directory.path().join("installation.json");
let namespaces = RootNamespaces::for_current_account().unwrap();
@@ -111,6 +112,7 @@ fn default_product_input_continues_while_the_original_stdout_owner_is_blocked()
tui.wait_for("Native input after original output completion");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
server.join().unwrap();
assert!(!tui.screen.snapshot().unwrap().alternate_screen);
drop(RootOwner::open(&host, &namespaces).unwrap());
@@ -124,6 +126,7 @@ fn large_history_fixture(model_base: Option<&str>) -> (tempfile::TempDir, std::p
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
let host = ui.join("host");
+ let host_guard = HostRootGuard::new(&host);
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -144,7 +147,7 @@ fn large_history_fixture(model_base: Option<&str>) -> (tempfile::TempDir, std::p
let mut bootstrap = Pty::spawn(&args);
bootstrap.wait_until_with_timeout(|s| s.contains("Settings"), Duration::from_secs(60));
bootstrap.send(b"\x11");
- // Wait for the original Host shutdown as well as terminal restoration.
+ // First restore the terminal, then explicitly retire the business Host.
let closing = Instant::now();
while bootstrap.child.try_wait().unwrap().is_none() {
bootstrap.read();
@@ -154,6 +157,7 @@ fn large_history_fixture(model_base: Option<&str>) -> (tempfile::TempDir, std::p
);
}
bootstrap.finish();
+ host_guard.stop();
runtime.block_on(async {
let identity = maka_fs_tools::workspace::ensure_identity(&workspace).await.unwrap();
let published = maka_fs_tools::workspace::directory::PublishedDirectory::open(&workspace).unwrap();
@@ -210,6 +214,7 @@ fn large_history_fixture(model_base: Option<&str>) -> (tempfile::TempDir, std::p
#[ignore = "manual default product history scroll profiling in a real PTY"]
fn default_product_scrolls_large_imported_history() {
let (_directory, ui) = large_history_fixture(None);
+ let host_guard = HostRootGuard::new(&ui.join("host"));
let mut tui = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
eprintln!("scroll fixture PID {}", tui.child.id());
tui.wait_until_with_timeout(|s| s.contains("Scroll fixture"), Duration::from_secs(60));
@@ -300,6 +305,7 @@ fn default_product_scrolls_large_imported_history() {
evidence(&tui, "scroll-profile");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
}
#[test]
@@ -314,6 +320,7 @@ fn default_product_sends_over_large_imported_history() {
.unwrap_or(2);
let server = model_server(listener, requests, rounds, 0);
let (_directory, ui) = large_history_fixture(Some(&base));
+ let host_guard = HostRootGuard::new(&ui.join("host"));
let mut tui = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
eprintln!("send fixture PID {}", tui.child.id());
tui.wait_until_with_timeout(|s| s.contains("Scroll fixture"), Duration::from_secs(60));
@@ -409,6 +416,7 @@ fn default_product_sends_over_large_imported_history() {
tui.wait_for(draft);
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
server.join().unwrap();
for reopen in 0..2 {
let mut restored = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
@@ -427,6 +435,7 @@ fn default_product_sends_over_large_imported_history() {
evidence(&restored, "large-history-draft-restored");
restored.send(b"\x11");
restored.finish();
+ host_guard.stop();
tokio::runtime::Runtime::new().unwrap().block_on(async {
use sqlx::{Connection, Row};
let mut db = sqlx::SqliteConnection::connect_with(&sqlx::sqlite::SqliteConnectOptions::new()
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/plan.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/plan.rs
index 627f45dba5..285eea21db 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/plan.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/plan.rs
@@ -40,6 +40,7 @@ fn default_product_plan_requires_review_and_restores_frozen_proposal_and_complet
let workspace = directory.path().join("workspace");
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
+ let host_guard = HostRootGuard::new(&ui.join("host"));
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -104,6 +105,7 @@ fn default_product_plan_requires_review_and_restores_frozen_proposal_and_complet
evidence(&tui, "plan-proposed");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
let mut tui = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
tui.wait_until_with_timeout(|s| s.contains("Settings"), Duration::from_secs(60));
@@ -152,6 +154,7 @@ fn default_product_plan_requires_review_and_restores_frozen_proposal_and_complet
tui.wait_for("Keep this independent draft.");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
let mut tui = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
tui.wait_until_with_timeout(|s| s.contains("Settings"), Duration::from_secs(60));
open(&mut tui);
@@ -161,5 +164,6 @@ fn default_product_plan_requires_review_and_restores_frozen_proposal_and_complet
assert!(received.try_recv().is_err());
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
server.join().unwrap();
}
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/recap.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/recap.rs
index aeb71418cf..f249df77de 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/recap.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/recap.rs
@@ -46,6 +46,7 @@ fn default_product_recap_reads_original_result_without_charging_on_refresh_or_re
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
let host = ui.join("host");
+ let host_guard = HostRootGuard::new(&host);
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -118,6 +119,7 @@ fn default_product_recap_reads_original_result_without_charging_on_refresh_or_re
tui.wait_for("Keep this unfinished draft.");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
server.join().unwrap();
let mut restored = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
restored.wait_until_with_timeout(
@@ -131,6 +133,7 @@ fn default_product_recap_reads_original_result_without_charging_on_refresh_or_re
evidence(&restored, "recap-restarted");
restored.send(b"\x11");
restored.finish();
+ host_guard.stop();
runtime.block_on(async {
let store = EventLog::for_root(Arc::new(RootOwner::open(&host, &namespaces).unwrap()))
.await
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/title.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/title.rs
index 34d07704e5..7bee54d472 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/title.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/title.rs
@@ -36,6 +36,7 @@ fn default_product_title_updates_once_and_preserves_manual_name_after_restart()
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
let host = ui.join("host");
+ let host_guard = HostRootGuard::new(&host);
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -71,6 +72,7 @@ fn default_product_title_updates_once_and_preserves_manual_name_after_restart()
evidence(&tui, "automatic-title");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
let mut restored = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
restored.wait_until_with_timeout(
|s| s.contains("Workspace conversation"),
@@ -104,6 +106,7 @@ fn default_product_title_updates_once_and_preserves_manual_name_after_restart()
evidence(&restored, "manual-title");
restored.send(b"\x11");
restored.finish();
+ host_guard.stop();
server.join().unwrap();
runtime.block_on(async {
let owner = Arc::new(RootOwner::open(&host, &namespaces).unwrap());
diff --git a/products/maka/cli/tests/integration/terminal/bundled_product/chat/todo.rs b/products/maka/cli/tests/integration/terminal/bundled_product/chat/todo.rs
index e79d0a3429..989824e0e7 100644
--- a/products/maka/cli/tests/integration/terminal/bundled_product/chat/todo.rs
+++ b/products/maka/cli/tests/integration/terminal/bundled_product/chat/todo.rs
@@ -31,6 +31,7 @@ fn default_product_checklist_tracks_model_progress_and_survives_restart() {
std::fs::create_dir(&workspace).unwrap();
let ui = directory.path().join("ui");
let host = ui.join("host");
+ let host_guard = HostRootGuard::new(&host);
let namespaces = RootNamespaces::for_current_account().unwrap();
drop(RootOwner::create(&ui, &namespaces).unwrap());
let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -92,6 +93,7 @@ fn default_product_checklist_tracks_model_progress_and_survives_restart() {
tui.wait_for("Message…");
tui.send(b"\x11");
tui.finish();
+ host_guard.stop();
let mut restored = Pty::spawn(&["application", "launch", "--root", ui.to_str().unwrap()]);
restored.wait_until_with_timeout(|s| s.contains("Checklist 1/25"), Duration::from_secs(60));
click(&mut restored, "Checklist 1/25");
@@ -128,5 +130,6 @@ fn default_product_checklist_tracks_model_progress_and_survives_restart() {
restored.wait_for("This conversation has no checklist.");
restored.send(b"\x11");
restored.finish();
+ host_guard.stop();
server.join().unwrap();
}
diff --git a/products/maka/desktop/README.md b/products/maka/desktop/README.md
index 0df728efc6..b4faf4eb03 100644
--- a/products/maka/desktop/README.md
+++ b/products/maka/desktop/README.md
@@ -23,6 +23,6 @@
Builds the `maka-desktop` executable and owns native windows. It loads a committed Desktop composition, injects the UI JavaScript execution adapter and uses `maka-render-gpui` to present frames. Pages and workflows come from plugin packages.
-Build with `just build`, then launch an installed Desktop deployment with `target/debug/maka desktop --root PATH --deployment UUID --connections FILE`. The CLI starts its sibling frontend. Window shutdown waits for native resource release, accepted operations and Host connection cleanup. Platform-specific IME and accessibility behavior needs verification in a real window.
+Build with `just build`, then launch an installed Desktop deployment with `target/debug/maka desktop --root PATH --deployment UUID --connections FILE`. The CLI starts its sibling frontend. Startup connects to the configured local Host or starts it automatically through the sibling `maka` executable. Closing the last window or quitting also stops the Host when no other client remains. Choose “Close UI and Keep Host” from the application menu to keep background tasks running. Platform-spec [...]
[Source](src/main.rs) · [Cargo manifest](Cargo.toml)
diff --git a/products/maka/desktop/README.zh-CN.md b/products/maka/desktop/README.zh-CN.md
index 2a879af604..7bfb918093 100644
--- a/products/maka/desktop/README.zh-CN.md
+++ b/products/maka/desktop/README.zh-CN.md
@@ -23,6 +23,6 @@
构建 `maka-desktop` 可执行文件并管理原生窗口。它加载已提交的 Desktop 组合，注入 UI JavaScript 执行适配器，并通过 `maka-render-gpui` 呈现帧。页面与工作流来自插件包。
-使用 `just build` 构建，再以 `target/debug/maka desktop --root PATH --deployment UUID --connections FILE` 启动已安装的 Desktop 部署。CLI 会启动同目录前端。窗口关闭需要等待原生资源释放、已接受操作和 Host 连接清理。平台特有的 IME 与无障碍行为需要在真实窗口中验证。
+使用 `just build` 构建，再以 `target/debug/maka desktop --root PATH --deployment UUID --connections FILE` 启动已安装的 Desktop 部署。CLI 会启动同目录前端。启动时连接配置的本机 Host；不存在时，通过同目录 `maka` 自动启动。关闭最后一个窗口或退出应用时，若没有其他客户端连接，Host 也会退出。可通过应用菜单的“Close UI and Keep Host”仅关闭界面，保留后台任务。平台特有的 IME 与无障碍行为需要在真实窗口中验证。
[源码](src/main.rs) · [Cargo 配置](Cargo.toml)
diff --git a/products/maka/desktop/examples/native_lifecycle.rs b/products/maka/desktop/examples/native_lifecycle.rs
index 5f2330bbe1..003cea684d 100644
--- a/products/maka/desktop/examples/native_lifecycle.rs
+++ b/products/maka/desktop/examples/native_lifecycle.rs
@@ -17,9 +17,14 @@
* under the License.
*/
-//! Live macOS probe of the production launcher, native quit selector and Host close.
-//! Run with a fresh evidence directory and optionally `fail`; create `release`
-//! after `closing` appears. Failed close intentionally remains fenced until killed.
+//! Live macOS probe of native quit and public Host cleanup.
+//! Run with a fresh evidence directory and optionally `fail`, `keep`, or `sql`.
+//! Default native quit closes the last Host session: wait for `closing`, then
+//! create `release`; confirmed Host cleanup writes `closed`, `host_stopped`.
+//! The external driver verifies UI exit and Root release before writing `ui_process_exited`.
+//! Failed cleanup retains the Host and Root until killed. `keep` invokes the native
+//! "Close UI and Keep Host" menu: `ui_process_exited` precedes Host cleanup; create
+//! `close_host`, wait for `closing`, then create `release`. SQL uses `release_sql`.
#[cfg(target_os = "macos")]
#[allow(dead_code)]
#[path = "../src/main.rs"]
@@ -116,12 +121,17 @@ mod probe {
.nth(2)
.ok_or("probe directory required")?,
);
- let fail = std::env::args_os().nth(3).is_some_and(|a| a == "fail");
+ let deployment: uuid::Uuid = std::env::args()
+ .nth(3)
+ .ok_or("Host deployment required")?
+ .parse()?;
+ let fail = std::env::args_os().nth(4).is_some_and(|a| a == "fail");
let mut native = NativeRegistry::default();
+ let resource_directory = directory.clone();
native.insert(
"probe.File".into(),
Arc::new(move |_, _| {
- let directory = directory.clone();
+ let directory = resource_directory.clone();
Box::pin(async move {
let file = std::fs::File::create(directory.join("native-resource"));
let error = file.as_ref().err().map(ToString::to_string);
@@ -151,16 +161,70 @@ mod probe {
.worker_threads(1)
.enable_all()
.build()?;
- timer.block_on(maka_transport::transport::process::serve(
- native,
- Some(maka_host_js::factory(Default::default())),
- ))?;
- Ok(())
+ timer.block_on(async {
+ use maka_host_runtime::{application::host::Host, ipc::LocalListener};
+ use std::os::unix::fs::PermissionsExt;
+ let namespaces = RootNamespaces::for_current_account()?;
+ let owner = Arc::new(RootOwner::open(&directory.join("host"), &namespaces)?);
+ let store = Arc::new(EventLog::for_root(owner).await?);
+ let host = Host::open(
+ store.clone(),
+ deployment,
+ native,
+ Some(maka_host_js::factory(Default::default())),
+ endpoints()?,
+ vec![],
+ )
+ .await?;
+ let socket_directory = tempfile::Builder::new()
+ .prefix("maka-probe-")
+ .permissions(std::fs::Permissions::from_mode(0o700))
+ .tempdir_in("/tmp")?;
+ let address = socket_directory.path().join("h.sock");
+ let listener = LocalListener::bind(&address)
+ .map_err(|error| error as Box<dyn std::error::Error>)?;
+ let registration = host.publish_registration(&address, None)?;
+ let stop = tokio_util::sync::CancellationToken::new();
+ let shutdown = stop.clone();
+ let signal = directory.clone();
+ tokio::spawn(async move {
+ while !signal.join("close_host").is_file() {
+ tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+ }
+ shutdown.cancel();
+ });
+ std::fs::write(directory.join("host_ready"), std::process::id().to_string())?;
+ if let Err(error) = host.serve_local(listener, stop).await {
+ eprintln!("public Host cleanup unconfirmed: {error}");
+ // Retain the actual Host, SQL worker, registration and Root after
+ // a failed close; process exit is not a cleanup acknowledgement.
+ loop {
+ tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+ if host.shutdown().await.is_ok() {
+ break;
+ }
+ }
+ }
+ registration.remove()?;
+ drop(host);
+ drop(store);
+ std::fs::write(
+ directory.join("host_stopped"),
+ b"public Host cleanup confirmed",
+ )?;
+ Result::<(), Box<dyn std::error::Error>>::Ok(())
+ })
+ }
+ fn endpoints() -> Result<BTreeMap<String, maka_transport::HostEndpoint>, serde_json::Error> {
+ serde_json::from_value(json!({
+ "business":{"consumer":fixture::identity(13),"import":"business","grant":{"methods":["read","write"],"recovery":false}},
+ "receipts":{"consumer":fixture::identity(13),"import":"receipts","grant":{"methods":["read"],"recovery":true}}
+ }))
}
pub fn run() -> Result<(), Box<dyn std::error::Error>> {
if std::env::args_os()
.nth(1)
- .is_some_and(|a| a == "application-host")
+ .is_some_and(|a| a == "public-host")
{
return host();
}
@@ -171,7 +235,8 @@ mod probe {
);
std::fs::create_dir(&directory)?;
let mode = std::env::args().nth(2);
- let fail = mode.as_deref() == Some("fail");
+ let fail = matches!(mode.as_deref(), Some("fail" | "keep-fail"));
+ let keep = matches!(mode.as_deref(), Some("keep" | "keep-fail"));
let sql = mode.as_deref() == Some("sql");
let timer = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
@@ -209,25 +274,46 @@ mod probe {
} else {
std::fs::write(
&path,
- serde_json::to_vec(
- &json!({"hosts":[{"name":"local","location":{"kind":"process","root":directory.join("host")},"deployment":host,"endpoints":{"business":{"consumer":fixture::identity(13),"import":"business","grant":{"methods":["read","write"],"recovery":false}},"receipts":{"consumer":fixture::identity(13),"import":"receipts","grant":{"methods":["read"],"recovery":true}}}}]}),
- )?,
+ serde_json::to_vec(&json!({"hosts":[{
+ "name":"local",
+ "location":{"kind":"public","root":directory.join("host")},
+ "deployment":host,
+ "endpoints":endpoints()?
+ }]}))?,
)?;
}
+ if !sql {
+ let mut command = std::process::Command::new(std::env::current_exe()?);
+ command
+ .arg("public-host")
+ .arg(&directory)
+ .arg(host.to_string())
+ .stdin(std::process::Stdio::null());
+ if fail {
+ command.arg("fail");
+ }
+ // This fixture starts the independently owned service. The Desktop
+ // launcher only discovers it and never receives its process handle.
+ let mut child = command.spawn()?;
+ let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+ while !directory.join("host_ready").is_file() {
+ if let Some(status) = child.try_wait()? {
+ return Err(format!("public Host startup failed: {status}").into());
+ }
+ if std::time::Instant::now() >= deadline {
+ child.kill()?;
+ child.wait()?;
+ return Err("public Host startup timed out".into());
+ }
+ std::thread::sleep(std::time::Duration::from_millis(10));
+ }
+ }
let launcher = desktop::Launcher::start(
desktop::Args {
root: directory.join("caller"),
deployment: caller,
connections: Some(path),
},
- maka_transport::HostExecutable::new(
- std::env::current_exe()?,
- vec![
- "application-host".into(),
- directory.clone().into_os_string(),
- if fail { "fail".into() } else { "ok".into() },
- ],
- )?,
maka_render_gpui::registry(),
)?;
if sql {
@@ -278,13 +364,44 @@ mod probe {
let windows: *mut objc2::runtime::AnyObject = msg_send![app, windows];
let count: usize = msg_send![windows, count];
if count > 0 && !signal.join("requested").exists() {
- std::fs::write(
- signal.join("requested"),
- b"native terminate requested with a real window",
- )
- .unwrap();
- let _: () =
- msg_send![app,terminate:std::ptr::null_mut::<objc2::runtime::AnyObject>()];
+ if keep {
+ let active: objc2::runtime::Bool = msg_send![app, isActive];
+ let key: *mut objc2::runtime::AnyObject = msg_send![app, keyWindow];
+ if !active.as_bool() || key.is_null() {
+ return;
+ }
+ // Invoke the actual native menu action, including its
+ // explicit KeepHost mode and ordinary cleanup path.
+ let menu: *mut objc2::runtime::AnyObject = msg_send![app, mainMenu];
+ let item: *mut objc2::runtime::AnyObject = msg_send![menu, itemAtIndex:0usize];
+ let menu: *mut objc2::runtime::AnyObject = msg_send![item, submenu];
+ let _: () = msg_send![menu, update];
+ let count: usize = msg_send![menu, numberOfItems];
+ let mut invoked = false;
+ for index in 0..count {
+ let item: *mut objc2::runtime::AnyObject = msg_send![menu, itemAtIndex:index];
+ let title: *mut objc2::runtime::AnyObject = msg_send![item, title];
+ let title: *const std::ffi::c_char = msg_send![title, UTF8String];
+ if std::ffi::CStr::from_ptr(title).to_bytes() == b"Close UI and Keep Host" {
+ let enabled: objc2::runtime::Bool = msg_send![item, isEnabled];
+ assert!(enabled.as_bool(), "native Keep Host menu action disabled");
+ let action: objc2::runtime::Sel = msg_send![item, action];
+ let target: *mut objc2::runtime::AnyObject = msg_send![item, target];
+ // Use Cocoa's menu action/responder routing and
+ // verify delivery, rather than assuming a menu
+ // item lookup proves its action was invoked.
+ let sent: objc2::runtime::Bool = msg_send![app, sendAction: action, to: target, from: item];
+ assert!(sent.as_bool(), "native Keep Host menu action was not dispatched");
+ std::fs::write(signal.join("requested"), b"native Keep Host menu action dispatched").unwrap();
+ invoked = true;
+ break;
+ }
+ }
+ assert!(invoked, "native Keep Host menu action missing");
+ } else {
+ std::fs::write(signal.join("requested"), b"native terminate requested with a real window").unwrap();
+ let _: () = msg_send![app,terminate:std::ptr::null_mut::<objc2::runtime::AnyObject>()];
+ }
}
if signal.join("request_again").is_file() {
std::fs::remove_file(signal.join("request_again")).unwrap();
@@ -294,6 +411,8 @@ mod probe {
});
}
});
+ // Cocoa termination may exit the process without returning from run.
+ // The external driver verifies actual process exit and released Root.
launcher.run()
}
}
diff --git a/products/maka/desktop/src/actions.rs b/products/maka/desktop/src/actions.rs
index 7d4756df82..9d5c247970 100644
--- a/products/maka/desktop/src/actions.rs
+++ b/products/maka/desktop/src/actions.rs
@@ -34,6 +34,8 @@ use std::{
sync::Arc,
task::{Wake, Waker},
};
+gpui_kit::actions!(maka, [Quit, CloseUiAndKeepHost]);
+
type Connections = Arc<tokio::sync::Mutex<maka_transport::Connections>>;
type Observer = Shared<LocalBoxFuture<'static, ()>>;
struct Job {
@@ -47,6 +49,7 @@ struct State {
caller: Option<Arc<maka_event_log::EventLog>>,
draining: Cell<bool>,
quitting: Cell<bool>,
+ exit_mode: Cell<maka_transport::ExitMode>,
launch: RefCell<Option<launch::State>>,
failure: RefCell<Option<String>>,
native: RefCell<BTreeMap<uuid::Uuid, Rc<RefCell<super::resources::NativeResources>>>>,
@@ -68,6 +71,9 @@ impl Wake for UiWake {
}
}
impl Owner {
+ pub(super) fn exit_mode(&self) -> maka_transport::ExitMode {
+ self.0.exit_mode.get()
+ }
pub(super) fn retain_native(&self, resources: Rc<RefCell<super::resources::NativeResources>>) {
let root = resources.borrow().surface().id();
let mut native = self.0.native.borrow_mut();
@@ -239,6 +245,7 @@ pub(super) fn own(
caller,
draining: Cell::new(false),
quitting: Cell::new(false),
+ exit_mode: Cell::new(maka_transport::ExitMode::Quit),
launch: RefCell::new(None),
failure: RefCell::new(None),
native: RefCell::new(BTreeMap::new()),
@@ -307,6 +314,37 @@ pub(super) fn own(
.detach();
// Installed once per App. The callback captures no application/Root lease.
if !cx.has_global::<QuitGate>() {
+ cx.on_action(|_: &Quit, cx| {
+ // Menu/key actions run inside the active Window's update. Release
+ // that borrow before requesting removal of every native window.
+ cx.defer(|cx| {
+ request_quit(cx);
+ });
+ });
+ cx.on_action(|_: &CloseUiAndKeepHost, cx| {
+ if let Some(owner) = cx.try_global::<Owner>() {
+ if owner.0.quitting.get() || owner.0.draining.get() {
+ return;
+ }
+ owner.0.exit_mode.set(maka_transport::ExitMode::KeepHost);
+ }
+ cx.defer(|cx| {
+ request_quit(cx);
+ });
+ });
+ cx.set_menus([gpui_kit::Menu::new("Maka").items([
+ gpui_kit::MenuItem::action("Close UI and Keep Host", CloseUiAndKeepHost),
+ gpui_kit::MenuItem::action("Quit Maka", Quit),
+ ])]);
+ cx.bind_keys([gpui_kit::KeyBinding::new(
+ if cfg!(target_os = "macos") {
+ "cmd-q"
+ } else {
+ "ctrl-q"
+ },
+ Quit,
+ None,
+ )]);
cx.on_request_quit(request_quit)
.map_err(|e| maka_ui_runtime::Error::Invalid(e.to_string()))?;
cx.set_global(QuitGate);
@@ -470,13 +508,14 @@ pub(super) fn quit_when_drained(application: &Rc<RefCell<DesktopApplication>>, c
let connections = owner.0.connections.clone();
let caller = owner.0.caller.clone();
let updates = owner.take_updates();
+ let exit_mode = owner.exit_mode();
owner.0.draining.set(true);
let task = application.borrow().scheduler().spawn(async move {
if let Some(updates) = updates {
updates.drain().await;
}
if let Some(connections) = connections {
- connections.lock().await.drain().await;
+ connections.lock().await.drain_with(exit_mode).await;
}
if let Some(caller) = caller {
caller.drain().await;
diff --git a/products/maka/desktop/src/editor_tests.rs b/products/maka/desktop/src/editor_tests.rs
index 467a509beb..508404f362 100644
--- a/products/maka/desktop/src/editor_tests.rs
+++ b/products/maka/desktop/src/editor_tests.rs
@@ -298,6 +298,34 @@ impl Scene {
std::thread::park_timeout(std::time::Duration::from_millis(1));
}
}
+ // Multi-scene fixtures must observe the previous application's foreground
+ // owner release before creating its successor in the same GPUI context.
+ fn close_and_drain(&self, cx: &mut TestAppContext) {
+ self.close(cx);
+ let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+ loop {
+ cx.run_until_parked();
+ let drained = cx.update(|cx| {
+ actions::quit_when_drained(&self.app, cx);
+ !cx.has_global::<actions::Owner>()
+ });
+ let cleaned = match self.cleanup.borrow().as_ref() {
+ Some(result) => {
+ result.as_ref().unwrap();
+ true
+ }
+ None => false,
+ };
+ if cleaned && drained {
+ break;
+ }
+ assert!(
+ std::time::Instant::now() < deadline,
+ "original application owner did not finish cleanup"
+ );
+ std::thread::park_timeout(std::time::Duration::from_millis(1));
+ }
+ }
fn close(&self, cx: &mut TestAppContext) {
cx.update_window(self.window, |_, window, _| window.remove_window())
.unwrap();
@@ -1189,7 +1217,7 @@ fn real_composer_snapshot_precedes_reset_and_capture_conflict_does_not_fake_scen
}
assert_eq!(scene.app.borrow().accepted_action_count(), 0);
drop(owner);
- scene.close(cx);
+ scene.close_and_drain(cx);
}
}
struct HeldRead {
diff --git a/products/maka/desktop/src/editor_tests/cohost.rs b/products/maka/desktop/src/editor_tests/cohost.rs
index f64bcf8643..e2c24b02ea 100644
--- a/products/maka/desktop/src/editor_tests/cohost.rs
+++ b/products/maka/desktop/src/editor_tests/cohost.rs
@@ -34,7 +34,7 @@ use maka_plugin_core::{
plan::RemoteCapability,
runtime::Published,
};
-use maka_transport::{Connections, HostExecutable};
+use maka_transport::Connections;
use std::sync::atomic::{AtomicBool, Ordering};
struct CloseWrites {
@@ -239,7 +239,7 @@ fn actual_quit_drains_cohost_native_sql_write_before_shared_store_close(cx: &mut
PresentationTarget::Desktop,
configuration,
namespaces.clone(),
- HostExecutable::new("absent-cohost-executable".into(), vec![]).unwrap(),
+ std::path::PathBuf::from("unused-cohost-startup-helper"),
native,
Some(maka_host_js::factory(Default::default())),
))
diff --git a/products/maka/desktop/src/editor_tests/reading.rs b/products/maka/desktop/src/editor_tests/reading.rs
index 47aed69c69..bfc970713b 100644
--- a/products/maka/desktop/src/editor_tests/reading.rs
+++ b/products/maka/desktop/src/editor_tests/reading.rs
@@ -306,8 +306,7 @@ fn actual_reading_tail_checks_wheel_and_completed_selection_before_whole_scene_p
}
assert!(!scene.app.borrow().is_retiring(&scene.root));
assert_eq!(scene.app.borrow().accepted_action_count(), 0);
- scene.close(cx);
- scene.app.borrow_mut().shutdown().unwrap();
+ scene.close_and_drain(cx);
}
});
}
diff --git a/products/maka/desktop/src/main.rs b/products/maka/desktop/src/main.rs
index c479eb391d..9aeb33674a 100644
--- a/products/maka/desktop/src/main.rs
+++ b/products/maka/desktop/src/main.rs
@@ -717,30 +717,7 @@ impl Render for Surface {
fn main() -> Result<(), Box<dyn std::error::Error>> {
log::set_logger(&NativeLog)?;
log::set_max_level(log::LevelFilter::Warn);
- if std::env::args_os().nth(1).as_deref() == Some(std::ffi::OsStr::new("application-host")) {
- if std::env::args_os().nth(2).is_some() {
- return Err("unexpected application-host arguments".into());
- }
- let native = compiled_host()?;
- let timer = tokio::runtime::Builder::new_multi_thread()
- .worker_threads(1)
- .enable_all()
- .build()?;
- timer.block_on(maka_transport::transport::process::serve(
- native,
- Some(maka_host_js::factory(Default::default())),
- ))?;
- return Ok(());
- }
- Launcher::start(
- Args::parse(),
- maka_transport::HostExecutable::new(
- std::env::current_exe()?,
- vec!["application-host".into()],
- )?,
- widgets::registry(),
- )?
- .run()
+ Launcher::start(Args::parse(), widgets::registry())?.run()
}
fn compiled_host() -> Result<maka_host_runtime::application::NativeRegistry, String> {
@@ -761,7 +738,6 @@ pub(crate) struct Launcher {
impl Launcher {
pub(crate) fn start(
args: Args,
- executable: maka_transport::HostExecutable,
components: maka_ui_runtime::NativeRegistry<dyn widgets::Widget>,
) -> Result<Self, Box<dyn std::error::Error>> {
let timer = tokio::runtime::Builder::new_multi_thread()
@@ -776,6 +752,11 @@ impl Launcher {
.transpose()?
.unwrap_or_default();
let cohost_native = compiled_host()?;
+ let helper = std::env::current_exe()?.with_file_name(if cfg!(windows) {
+ "maka.exe"
+ } else {
+ "maka"
+ });
let owner = Arc::new(maka_event_log::root::RootOwner::open(
&args.root,
&namespaces,
@@ -787,7 +768,7 @@ impl Launcher {
PresentationTarget::Desktop,
configuration,
namespaces,
- executable,
+ helper,
cohost_native,
Some(maka_host_js::factory(Default::default())),
)) {
@@ -854,7 +835,12 @@ impl Launcher {
mut application,
mount,
} = self;
- let frame = match application.frame(&mount) {
+ // Startup may inspect the prepared window declaration; native resource
+ // publication still commits the frame after the actual window exists.
+ let frame = match application
+ .frame(&mount)
+ .or_else(|_| application.prepared_frame(&mount))
+ {
Ok(frame) => frame,
Err(error) => {
drain_ui(&mut application, &timer, true);
@@ -926,11 +912,16 @@ impl Launcher {
.borrow()
.as_ref()
.and_then(actions::Owner::take_updates);
+ let exit_mode = execution_owner
+ .borrow()
+ .as_ref()
+ .map(actions::Owner::exit_mode)
+ .unwrap_or(maka_transport::ExitMode::KeepHost);
timer.block_on(async {
if let Some(updates) = updates {
updates.drain().await;
}
- connections.lock().await.drain().await;
+ connections.lock().await.drain_with(exit_mode).await;
store.drain().await;
});
if let Some(error) = execution_owner
diff --git a/skills/maka-plugin-authoring/assets/starter/verify-host.py b/skills/maka-plugin-authoring/assets/starter/verify-host.py
index 518bf4bd96..50ffea8ff2 100644
--- a/skills/maka-plugin-authoring/assets/starter/verify-host.py
+++ b/skills/maka-plugin-authoring/assets/starter/verify-host.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""Isolated Unix launcher smoke test for the generated Host Echo service."""
+"""Public Unix Host smoke test for the generated Echo service."""
import argparse
import json
import pathlib
@@ -50,7 +50,7 @@ def read_frame(stream):
data.extend(chunk)
return data
size = struct.unpack(">I", exact(4))[0]
- if size > 8 * 1024 * 1024:
+ if size > 2 * 1024 * 1024:
raise ValueError("Host frame exceeds test budget")
return json.loads(exact(size))
@@ -97,32 +97,56 @@ def main():
source = run("application", "commit", "--root", root, "--deployment", str(uuid.uuid4()),
"--expected-revision", "0", "--input", project / "composition.json",
"--package", project / "plugin", "--package", consumer)
- nonce = str(uuid.uuid4())
- address = str(base / "host.sock")
- bootstrap = {
- "root": str(root), "deployment": source["deployment"],
- "namespaces": {"ownership": str(base / "owners"), "control": str(base / "control")},
- "source": source, "consumers": [[key]], "native": [], "address": address, "nonce": nonce,
- "endpoints": {"echo": {"consumer": [key], "import": "echo", "grant": {"methods": ["echo"], "recovery": False}}},
- }
- # The launcher owns bootstrap/control and grants only this declared query.
- # Plugin business code uses its bound capability, not this wire protocol.
- with subprocess.Popen([args.maka, "application", "serve"], stdin=subprocess.PIPE,
+ endpoints = base / "endpoints.json"
+ save(endpoints, {
+ "echo": {"consumer": [key], "import": "echo", "grant": {"methods": ["echo"], "recovery": False}},
+ })
+ # The trusted launcher publishes only this declared query. Plugin
+ # business code uses its bound capability, not this public wire adapter.
+ with subprocess.Popen([args.maka, "application", "serve", "--root", str(root),
+ "--deployment", source["deployment"], "--endpoints", str(endpoints)],
+ stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, bufsize=0) as host:
try:
- write_frame(host.stdin, bootstrap)
- ready = read_frame(host.stdout)
+ if not select.select([host.stdout], [], [], 15)[0]:
+ raise TimeoutError("Host startup timed out")
+ ready = json.loads(host.stdout.readline(16 * 1024))
assert ready["kind"] == "ready", ready
- epoch = ready["catalogue"]["endpoints"]["echo"]["registration_epoch"]
- for value, valid in [("hello Maka", True), ("😀" * 4096, True), (42, False), ("x" * 4097, False)]:
- with socket.socket(socket.AF_UNIX) as connection:
- connection.settimeout(15)
- connection.connect(address)
- with connection.makefile("rwb", buffering=0) as channel:
- write_frame(channel, {"nonce": nonce})
- read_frame(channel)
+ assert ready["rootId"] == source["stateDomain"], ready
+ address = ready["socketPath"]
+
+ def connect():
+ connection = socket.socket(socket.AF_UNIX)
+ connection.settimeout(15)
+ connection.connect(address)
+ return connection
+
+ def opening(channel, request, epoch=None):
+ write_frame(channel, {"version": 1, "epoch": epoch, "request": request})
+ return read_frame(channel)
+
+ with connect() as connection, connection.makefile("rwb", buffering=0) as control:
+ connected = opening(control, {"kind": "connect"})
+ assert connected["kind"] == "connected", connected
+ connected = connected["value"]
+ assert connected["version"] == 1 and connected["source"] == source, connected
+ assert connected["consumers"]["echo"] == [key], connected
+ epoch = connected["epoch"]
+ registration = connected["catalogue"]["endpoints"]["echo"]["registration_epoch"]
+ assert registration, connected
+ with connect() as probe, probe.makefile("rwb", buffering=0) as channel:
+ status = opening(channel, {"kind": "status"}, epoch)
+ assert status["kind"] == "status", status
+ assert status["value"]["lifecycle_mode"] == "service", status
+ assert status["value"]["source"] == source, status
+ assert not status["value"]["closing"], status
+ for value, valid in [("hello Maka", True), ("😀" * 4096, True), (42, False), ("x" * 4097, False)]:
+ with connect() as call, call.makefile("rwb", buffering=0) as channel:
+ accepted = opening(channel, {"kind": "call", "session": connected["session"]}, epoch)
+ assert accepted == {"kind": "accepted"}, accepted
write_frame(channel, {
- "endpoint": "echo", "registration_epoch": epoch, "method": "echo", "input": value,
+ "endpoint": "echo", "registration_epoch": registration,
+ "method": "echo", "input": value,
"source": {"deployment": source, "consumer": [key], "mount": str(uuid.uuid4()), "request": str(uuid.uuid4())},
})
write_frame(channel, {})
@@ -134,9 +158,24 @@ def main():
assert reply["kind"] == "finished" and reply["result"] == {"Ok": value}, reply
else:
assert reply["kind"] == "aborted", reply
- write_frame(host.stdin, {"kind": "close", "nonce": nonce})
- assert read_frame(host.stdout) == {"kind": "closed", "nonce": nonce}
+ # Closing the session confirms original call cleanup without
+ # stopping this Service-mode Host.
+ write_frame(control, {"kind": "close"})
+ assert read_frame(control) == {"kind": "closed"}
+ with connect() as connection, connection.makefile("rwb", buffering=0) as channel:
+ status = opening(channel, {"kind": "status"}, epoch)
+ assert status["value"]["sessions"] == 0, status
+ assert status["value"]["active_calls"] == 0, status
+ assert not status["value"]["closing"], status
+ with connect() as connection, connection.makefile("rwb", buffering=0) as channel:
+ retired = opening(channel, {"kind": "retire", "interrupt": False}, epoch)
+ assert retired == {"kind": "retired", "value": {"Ok": True}}, retired
assert host.wait(timeout=15) == 0
+ assert not pathlib.Path(address).exists(), "Host socket was not removed"
+ # Reopening validates both persisted content and release of the
+ # original Root lease after successful Host cleanup.
+ persisted = run("application", "read", "--root", root, "--deployment", source["deployment"])
+ assert persisted["reference"] == source, persisted
finally:
if host.poll() is None:
host.kill()
diff --git a/skills/maka-plugin-authoring/references/runtime-ui.md b/skills/maka-plugin-authoring/references/runtime-ui.md
index ba3e16fee4..3a382a6da0 100644
--- a/skills/maka-plugin-authoring/references/runtime-ui.md
+++ b/skills/maka-plugin-authoring/references/runtime-ui.md
@@ -68,4 +68,4 @@ For a direct HTTP(S) `.maka-plugin` download, submit `UIRequest` with `method: "
The download is bounded to 32 MiB and 60 seconds, follows at most five redirects without HTTPS downgrade, and uses the UI management process's proxy environment. A revision change during download rejects the stale install. A URL must point to an exported bundle, not a repository or web page.
-If the foreground cannot open, close its UI/local Host and run `maka application restore-defaults --root UI_ROOT`. This restores the originally installed Host/management/UI compositions and exact package versions while preserving their instance identities and data. It does not choose a new product or upgrade the installation. Interrupted restoration resumes from its saved preparation.
+If the foreground cannot open, close its UI and run `maka application restore-defaults --root UI_ROOT`. This restores the originally installed Host/management/UI compositions and exact package versions while preserving their instance identities and data. It does not choose a new product or upgrade the installation. Interrupted restoration resumes from its saved preparation.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org