Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
47 commits
Select commit Hold shift + click to select a range
da418a1
test(mcp): add owner tests module for module client
senamakel Oct 12, 2026
88b8353
test(mcp): cover cancelled close retrying on the original owner
senamakel Oct 12, 2026
0cf63f9
test(mcp): make close poll release deterministic in owner tests
senamakel Oct 12, 2026
a23d47c
feat(modules): pin module client to its unique peer
senamakel Oct 12, 2026
824e354
refactor(mcp): move TinyMcpPort into its own module
senamakel Oct 12, 2026
53ee360
fix(modules): preserve call timeout on pinned and derived object clients
senamakel Oct 12, 2026
b38b32c
fix(mcp): reject calls to retired module clients
senamakel Oct 12, 2026
bd615a1
feat(mcp): expose agent cleanup helpers and module runtime spawn
senamakel Oct 12, 2026
7d9cab3
feat(agent): defer MCP cleanup when agent teardown fails
senamakel Oct 12, 2026
00632d5
refactor(agent): track pending cleanup in the registry
senamakel Oct 12, 2026
5ade390
fix(embed): retry pending agent cleanup on removal
senamakel Oct 12, 2026
ec184c1
test(mcp): cover failed agent cleanup and retry ownership
senamakel Oct 12, 2026
cdd6f17
chore: files changed crates/openhuman-core/src/mcp/module_client_owne…
senamakel Oct 12, 2026
c0435a6
test(agent): add test module for mcp cleanup
senamakel Oct 12, 2026
43cbf2f
docs(embed): clarify agent teardown and MCP cleanup semantics
senamakel Oct 12, 2026
c4794fd
test(mcp): cover blocked close cleanup deadline and owner retry
senamakel Oct 12, 2026
0311a42
docs(mcp): clarify agent cleanup runtime requirements
senamakel Oct 12, 2026
4c4fe6a
test(mcp): build host overrides via default and field assignment
senamakel Oct 12, 2026
04f37c1
chore: files changed crates/openhuman-core/src/mcp/agent_cleanup.rs,c…
senamakel Oct 12, 2026
4268df3
refactor(agent): extract guard release into complete_with_guard
senamakel Oct 12, 2026
0c85865
feat(harness): add mcp cleanup error variant
senamakel Oct 12, 2026
4981af5
fix(embed): map mcp cleanup harness errors
senamakel Oct 12, 2026
5d52949
test(agent): run cleanup guard test on a dedicated runtime
senamakel Oct 12, 2026
c210b83
docs(agent): clarify mcp cleanup guard lifetime
senamakel Oct 12, 2026
f9a2b81
test(agent): assert retry guard is retained after failed completion
senamakel Oct 12, 2026
0363183
test(agent): assert guard survives cleanup failure
senamakel Oct 12, 2026
a5aadbf
fix(openhuman-embed): keep admission guard alive during completion
senamakel Oct 12, 2026
542434b
merge: integrate canonical MCP cleanup and admitted tool changes
senamakel Oct 12, 2026
c0244a7
refactor(agent): extract cleanup retry loop into a helper
senamakel Oct 12, 2026
089df7e
test(agent): cover ephemeral workspace cleanup on cancel
senamakel Oct 12, 2026
1bc675b
fix(embed): preserve ephemeral workspace when cleanup is unconfirmed
senamakel Oct 12, 2026
233ca43
test(agent): cover purge fence surviving runtime replacement
senamakel Oct 12, 2026
563ded1
feat(mcp): reserve agent directories until host cleanup completes
senamakel Oct 12, 2026
bcc25f3
test(mcp): cover orphaned cleanup retries and ephemeral preservation
senamakel Oct 12, 2026
372eee9
chore: files changed crates/openhuman-core/src/mcp/module_client_test…
senamakel Oct 12, 2026
f07267d
refactor(mcp): move host cleanup out of the mcp feature gate
senamakel Oct 12, 2026
ee29ab1
chore(mcp): gate directory_pending behind non-mcp feature
senamakel Oct 12, 2026
dae8111
Merge remote-tracking branch 'pr7521/main' into mcp-embed-teardown-7513
senamakel Oct 18, 2026
6a9d8d5
chore(vendor): bump tinymcp submodule to 2df3935
senamakel Oct 18, 2026
6e10fa0
test(mcp): update cleanup tests for lifecycle table API
senamakel Oct 18, 2026
cf751c4
test(mcp): canonicalize temp workspace paths in lifecycle tests
senamakel Oct 18, 2026
9fa0fda
test(agent_lifecycle): log progress while waiting for agents to park
senamakel Oct 18, 2026
97b406d
test(agent_lifecycle): use a host-provided tool to park at the approv…
senamakel Oct 18, 2026
d9b127d
test(agent-lifecycle): scope parked agent tools to the named marker tool
senamakel Oct 18, 2026
231a5b1
test(openhuman-embed): gate link_tinyflows helper behind modules feature
senamakel Oct 18, 2026
3db4619
test(embed): gate tinyflows linking behind modules feature
senamakel Oct 18, 2026
26bc6bb
chore(vendor): bump tinybus and tinymcp submodules
senamakel Oct 18, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion crates/openhuman-app/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

49 changes: 49 additions & 0 deletions crates/openhuman-core/src/mcp/agent_cleanup.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
//! Captured cleanup of an agent's existing configured module object.
use super::*;

/// A retired agent's original MCP owner and its retryable close receipts.
/// Capturing this handle never creates a client or loads a module.
#[derive(Clone)]
pub struct AgentMcpCleanup {
directory: PathBuf,
client: Arc<McpConfiguredClient>,
}

/// Retire the existing configured client for this exact agent directory.
/// An absent entry requires no cleanup and leaves the module loader untouched.
pub fn take_agent_cleanup(workspace: &std::path::Path, agent_id: &str) -> Option<AgentMcpCleanup> {
let directory = super::lifecycle::client_key(&workspace.join("agents").join(agent_id));
let client = super::lifecycle::table().retire(&directory)?;
Some(AgentMcpCleanup { directory, client })
}

impl AgentMcpCleanup {
/// Close the original object and evict it only after confirmed cleanup.
/// The whole attempt, including lock acquisition and bus calls, has a
/// 30-second deadline. Timeout returns `TransportFailed`; cancellation or
/// error retains the same client and transition receipt for explicit retry.
pub async fn finish(&self) -> Result<(), ModuleCallError> {
tokio::time::timeout(TRANSITION_TIMEOUT, self.client.close())
.await
.map_err(|_| ModuleCallError::TransportFailed)??;
self.evict();
Ok(())
}

/// Finish an unused client without starting an asynchronous executor.
/// An accepted call holding the lifecycle lock requires asynchronous drain.
pub fn finish_if_idle(&self) -> bool {
let Ok(state) = self.client.state.try_lock() else {
return false;
};
if !state.is_closed() {
return false;
}
self.evict();
true
}

fn evict(&self) {
super::lifecycle::table().evict(&self.directory, &self.client);
}
}
14 changes: 11 additions & 3 deletions crates/openhuman-core/src/mcp/host.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,14 @@ use std::sync::OnceLock;

use crate::config::Config;

pub use super::host_cleanup::{
reserve_agent_directory, spawn_agent_cleanup, AgentDirectoryCleanup,
};
#[cfg(not(feature = "mcp"))]
pub use super::host_cleanup::{take_agent_cleanup, AgentMcpCleanup};
#[cfg(feature = "mcp")]
pub use super::module_client::agent_cleanup::{take_agent_cleanup, AgentMcpCleanup};

const PROXY_SERVICE_KEY: &str = "tool.mcp_client";
const DEFAULT_CORE_PORT: u16 = 7788;
static DEFAULT_CONFIG: OnceLock<Config> = OnceLock::new();
Expand Down Expand Up @@ -93,10 +101,10 @@ pub fn schedule_agent_close_with_cleanup(
let _ = after_close();
}

/// No agent-scoped module cleanup is pending when MCP is disabled.
/// Host-owned directory cleanup can remain pending even without MCP.
#[cfg(not(feature = "mcp"))]
pub fn agent_close_pending(_workspace_dir: &std::path::Path, _agent_id: &str) -> bool {
false
pub fn agent_close_pending(workspace_dir: &std::path::Path, agent_id: &str) -> bool {
super::host_cleanup::directory_pending(workspace_dir, agent_id)
}

/// Converts the MCP section of product configuration to the module contract.
Expand Down
133 changes: 133 additions & 0 deletions crates/openhuman-core/src/mcp/host_cleanup.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
//! Process-local host cleanup fencing, also used when MCP is compiled out.
use std::{
path::PathBuf,
sync::{Arc, OnceLock},
};
use tinybus::module::ModuleCallError;
use tinymcp_bus::client_lifecycle::{CleanupReservation, ClientTable};

#[cfg(feature = "mcp")]
type CleanupClient = super::module_client::McpConfiguredClient;
#[cfg(not(feature = "mcp"))]
pub(super) struct CleanupClient;
#[cfg(not(feature = "mcp"))]
impl tinymcp_bus::client_lifecycle::Retire for CleanupClient {
fn mark_retiring(&self) {}
fn is_retiring(&self) -> bool {
false
}
}
static CLIENTS: OnceLock<ClientTable<CleanupClient>> = OnceLock::new();
pub(super) fn table() -> &'static ClientTable<CleanupClient> {
CLIENTS.get_or_init(ClientTable::new)
}

pub(super) fn directory_key(directory: &std::path::Path) -> PathBuf {
#[cfg(feature = "mcp")]
{
super::module_client::lifecycle::client_key(directory)
}
#[cfg(not(feature = "mcp"))]
{
let mut missing = Vec::new();
let mut ancestor = directory;
loop {
if let Ok(real) = ancestor.canonicalize() {
return missing
.iter()
.rev()
.fold(real, |path, part| path.join(part));
}
match (ancestor.parent(), ancestor.file_name()) {
(Some(parent), Some(name)) => {
missing.push(name.to_owned());
ancestor = parent;
}
_ => return directory.to_path_buf(),
}
}
}
}

pub(super) fn directory_pending(workspace: &std::path::Path, agent_id: &str) -> bool {
table().close_pending(&directory_key(&workspace.join("agents").join(agent_id)))
}

/// Run captured cleanup on the process module executor when modules are enabled.
/// Without modules, the caller must keep its current Tokio runtime alive.
/// This creates no configured client and loads no MCP artifact.
pub fn spawn_agent_cleanup<F>(
future: F,
) -> Result<tokio::task::JoinHandle<F::Output>, ModuleCallError>
where
F: std::future::Future + Send + 'static,
F::Output: Send + 'static,
{
#[cfg(feature = "modules")]
{
crate::modules::runtime::spawn_cleanup(future).map_err(|_| ModuleCallError::TransportFailed)
}
#[cfg(not(feature = "modules"))]
{
tokio::runtime::Handle::try_current()
.map(|handle| handle.spawn(future))
.map_err(|_| ModuleCallError::Unavailable)
}
}

/// A process-local directory reservation with no runtime guard or callback.
/// Dropping the token retains the fence; only confirmed host completion clears it.
#[derive(Clone)]
pub struct AgentDirectoryCleanup {
directory: PathBuf,
reservation: Arc<CleanupReservation>,
}

/// Reserve an agent's directory until host-owned cleanup succeeds.
/// This never creates a client or loads a module.
pub fn reserve_agent_directory(
workspace: &std::path::Path,
agent_id: &str,
) -> AgentDirectoryCleanup {
let directory = directory_key(&workspace.join("agents").join(agent_id));
let reservation = table().reserve_host_completion(&directory);
AgentDirectoryCleanup {
directory,
reservation,
}
}

impl AgentDirectoryCleanup {
/// Release this exact reservation after module close and host purge succeed.
/// An independently registered canonical callback keeps its own reservation.
pub fn complete(&self) {
table().complete_host_cleanup(&self.directory, &self.reservation);
}
}

/// An opaque cleanup handle for a build with no MCP client capability.
/// Such a build cannot create the handle or hold MCP resources.
#[cfg(not(feature = "mcp"))]
#[derive(Clone)]
pub struct AgentMcpCleanup {
_unconstructible: (),
}

#[cfg(not(feature = "mcp"))]
impl AgentMcpCleanup {
/// There are no MCP resources in this compiled-out capability.
pub async fn finish(&self) -> Result<(), ModuleCallError> {
Ok(())
}

/// There are no MCP resources to drain or retire.
pub fn finish_if_idle(&self) -> bool {
true
}
}

/// A build without MCP cannot contain a configured MCP client.
#[cfg(not(feature = "mcp"))]
pub fn take_agent_cleanup(_workspace: &std::path::Path, _id: &str) -> Option<AgentMcpCleanup> {
None
}
2 changes: 2 additions & 0 deletions crates/openhuman-core/src/mcp/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@ pub fn start(config: &crate::config::Config) {
#[cfg(feature = "mcp")]
pub(crate) mod module_client;

mod host_cleanup;

/// Brings this domain up — the build without it, where there is nothing to do.
#[cfg(not(feature = "mcp"))]
pub fn start(_config: &crate::config::Config) {}
Expand Down
119 changes: 10 additions & 109 deletions crates/openhuman-core/src/mcp/module_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -93,115 +93,12 @@ pub(crate) trait McpModulePort: Send + Sync {
}
}

struct TinyMcpPort {
root: ModuleClient,
}

#[async_trait]
impl McpModulePort for TinyMcpPort {
async fn open_configured(
&self,
data_dir: &str,
config: &McpClientConfig,
) -> Result<OpenConfiguredResponse, ModuleCallError> {
self.root
.call_confidential(
MODULE,
names::methods::OPEN_CONFIGURED,
(OpenConfiguredRequest {
data_dir: data_dir.to_owned(),
client: config.clone(),
},),
)
.await
}

async fn close(&self, data_dir: &str) -> Result<Option<String>, ModuleCallError> {
self.root
.call(MODULE, names::methods::CLOSE, (data_dir.to_owned(),))
.await
}

async fn transition_status(
&self,
transition_id: &str,
) -> Result<TransitionStatus, ModuleCallError> {
self.root
.call(
MODULE,
names::methods::TRANSITION_STATUS,
(transition_id.to_owned(),),
)
.await
}

async fn call_object(
&self,
object_path: &str,
member: &str,
args: Value,
confidential: bool,
) -> Result<Value, ModuleCallError> {
#[cfg(feature = "modules")]
{
let object = self.root.object_client(MODULE, object_path).await?;
if confidential {
object.call_confidential(member, args).await
} else {
object.call(member, args).await
}
}
#[cfg(not(feature = "modules"))]
{
let _ = (object_path, member, args, confidential);
Err(ModuleCallError::Unavailable)
}
}
#[path = "module_port.rs"]
mod module_port;
use module_port::TinyMcpPort;

async fn prepare_agent_tools(
&self,
request: PrepareAgentToolsRequest,
) -> Result<Vec<McpToolPresentation>, ModuleCallError> {
self.root
.call(MODULE, names::methods::PREPARE_AGENT_TOOLS, (request,))
.await
}

async fn validate_server_arguments(
&self,
request: ValidateServerArgumentsRequest,
) -> Result<ServerArgumentsValidation, ModuleCallError> {
self.root
.call_confidential(
MODULE,
names::methods::VALIDATE_SERVER_ARGUMENTS,
(request,),
)
.await
}

async fn normalize_tool_arguments(
&self,
arguments: Option<Value>,
) -> Result<Map<String, Value>, ModuleCallError> {
self.root
.call_confidential(
MODULE,
names::methods::NORMALIZE_TOOL_ARGUMENTS,
(arguments,),
)
.await
}

async fn transform_text(
&self,
request: tinymcp_bus::TransformTextRequest,
) -> Result<String, ModuleCallError> {
self.root
.call_confidential(MODULE, names::methods::TRANSFORM_TEXT, (request,))
.await
}
}
#[path = "agent_cleanup.rs"]
pub(crate) mod agent_cleanup;

/// One workspace's serialized TinyMCP lifecycle and private object proxy.
pub(crate) struct McpConfiguredClient {
Expand All @@ -211,7 +108,7 @@ pub(crate) struct McpConfiguredClient {
retiring: AtomicBool,
}

mod lifecycle;
pub(super) mod lifecycle;

impl McpConfiguredClient {
/// Calls one member on this workspace's private configured object.
Expand Down Expand Up @@ -540,3 +437,7 @@ impl Retire for McpConfiguredClient {
self.retiring.load(Ordering::Acquire)
}
}

#[cfg(all(test, feature = "modules"))]
#[path = "module_client_owner_tests.rs"]
mod owner_tests;
Loading
Loading