codewhale-hq/Codewhale
1
0 Fork 0
mirror of https://github.com/codewhale-hq/Codewhale.git synced 2026-11-07 17:09:13 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

fix(runtime): recover owner startup and preserve reviewed plugin updates

Browse Source 
Concurrent fresh starts and verified-dead owner reentries converge on the
incumbent Runtime owner and retain its selected configuration, store and
frontend through cancelled bootstrap and retirement work. Only typed
pre-attach receipt retirement is re-observed; unsafe permissions, ownership,
hardlinks, replacement publications and uncertain post-write effects remain
refusals. macOS/Linux no-replace receipt publication keeps one filesystem link.

Local-path plugin updates verify the whole installed bundle and retain a
private unique recovery backup during replacement. Internal staging/backup
paths stay out of discovery. Runtime auto-allow paths use the shared approval
disposition, preserving forced human approval and Never posture.

Verification: repaired production source32af4d9 passed fmt, CLI/TUI Clippy,
CLI build, 77/77 focused TUI/plugin/startup/approval checks, and18/18 private
file/xAI credential checks. Final85da82f changes only a cfg(test) import and
embedded release prose: fmt and app-server all-target/all-feature Clippy
passed;22/22 daemon/client/server checks passed. Feature registry:6/6 passed.
All30 newly added regression functions have actual PASS receipts.

Actual matching immutable dcb50251908d CLI:80/80 owner startup/reentry checks
plus9/9 native aa70de8b97a5 app recovery checks; all25 task PIDs absent;
zero provider/inference requests. Independent whole-fingerprint reconstruction
proves final production Rust equivalent; final embedded notes differ.
Earlier6ac94bff8904 plugin update/repair/rollback qualification:24/24 checks,
two bounded real DeepSeek turns with one exact canary MCP invocation each.
That earlier provider proof is not relabeled as matching dcb binary evidence.

Original startup and peer-server test compile failures are retained.
Windows, hosted/release/customer acceptance, full plugin serialization and
native positive inference remain required. Local source commit only;
release owner controls Engine push and promotion.

Qualified receipt SHA256 939d832b6f0ef8b0dc5451ef115678e97da0f1fcc2a372019d057a6b185cff46; immutable recovery receipts:
80checks650654aa6a50b1c5f278eda0ee21a979931ca20d382e83bb607b43845953643a
9checks2554c10eb0c21bb2027a18e2a9d465eaf862bf2c6e0d2f4ffc678a5c736813ee.
This commit is contained in:
CodeWhale Bot committed 2026-10-08 02:55:09 -07:00
1 parent bd0c325670
commit a8cdccc1d6
23 files changed
+2739 -307
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/app-server/src/daemon_client.rs
+327 -82
View File
@@ -34,6 +34,237 @@ mod platform {
frontend: crate::daemon_socket::AttachFrontend,
}
/// Host startup is the only caller allowed to acquire the Runtime lease.
/// Unavailable is pre-write transport evidence, never ownership or death.
pub enum HostOwnerProbe {
Absent,
Attached(Box<OwnerClient>),
Unavailable(Arc<UnavailablePublication>),
}
#[derive(Debug, thiserror::Error)]
#[error("selected owner publication changed before attach; re-observe")]
pub struct PublicationChangedBeforeAttach;
fn owner_receipt_error(error: anyhow::Error) -> anyhow::Error {
#[cfg(unix)]
if error.is::<codewhale_config::private_directory::UnlinkedPrivateFile>() {
return PublicationChangedBeforeAttach.into();
}
error
}
pub(crate) struct OwnerPublication {
pub(crate) parent: Arc<PrivateDirectory>,
pub(crate) receipt_name: String,
#[cfg(unix)]
pub(crate) socket_path: PathBuf,
pub(crate) bytes: Vec<u8>,
pub(crate) receipt: RuntimeOwnerReceipt,
pub(crate) file: std::fs::File,
#[cfg(unix)]
pub(crate) socket_identity:
Option<codewhale_config::private_directory::PrivateSocketIdentity>,
}
impl OwnerPublication {
pub(crate) fn receipt_is_current(&self) -> Result<bool> {
if !self.parent.is_at_selected_path()? {
return Ok(false);
}
let current = match self
.parent
.read_private_receipt(&self.receipt_name, 16384)
.map_err(owner_receipt_error)
{
Err(error) if error.is::<PublicationChangedBeforeAttach>() => return Ok(false),
result => result?,
};
let Some((bytes, file)) = current else {
return Ok(false);
};
Ok(bytes == self.bytes && PrivateDirectory::same_file_identity(&self.file, &file)?)
}
pub(crate) fn is_current(&self) -> Result<bool> {
if !self.receipt_is_current()? {
return Ok(false);
}
#[cfg(unix)]
{
let name = self
.socket_path
.file_name()
.and_then(|name| name.to_str())
.context("invalid owner endpoint basename")?;
if self.parent.socket_identity(name)? != self.socket_identity {
return Ok(false);
}
}
Ok(true)
}
}
/// Opaque retained publication. No public constructor or path-based cleanup.
pub struct UnavailablePublication {
pub(crate) publication: Arc<OwnerPublication>,
cause: std::io::Error,
}
impl UnavailablePublication {
pub fn receipt(&self) -> &RuntimeOwnerReceipt {
&self.publication.receipt
}
fn strict_error(&self) -> anyhow::Error {
let cause = match self.cause.raw_os_error() {
Some(code) => std::io::Error::from_raw_os_error(code),
None => std::io::Error::new(self.cause.kind(), self.cause.to_string()),
};
anyhow::Error::new(cause)
.context("selected owner publication unavailable; canonical host recovery required")
}
#[cfg(unix)]
fn require_dead_process(&self) -> Result<()> {
use codewhale_config::private_directory::{UnixProcessStatus, unix_process_status};
let receipt = self.receipt();
anyhow::ensure!(
receipt.principal == PrivateDirectory::current_user_id().to_string()
&& !receipt.process_start.is_empty(),
"selected owner process identity is invalid"
);
match unix_process_status(receipt.pid)? {
UnixProcessStatus::Absent => Ok(()),
UnixProcessStatus::Present { start } if start == receipt.process_start => {
bail!("selected owner process is still present; refusing recovery")
}
UnixProcessStatus::Present { .. } => {
bail!("selected owner PID was reused; refusing recovery")
}
}
}
/// Caller retains the real Runtime lease; this checks evidence only.
#[cfg(unix)]
pub(crate) fn revalidate_receipt(&self) -> Result<bool> {
if !self.publication.receipt_is_current()? {
return Ok(false);
}
self.require_dead_process()?;
Ok(true)
}
#[cfg(unix)]
pub(crate) fn revalidate_all(&self) -> Result<bool> {
if !self.publication.is_current()? {
return Ok(false);
}
self.require_dead_process()?;
Ok(true)
}
#[cfg(unix)]
pub async fn revalidate(self: &Arc<Self>) -> Result<bool> {
let held = self.clone();
owner_work(move || held.revalidate_all()).await
}
}
async fn observe_publication(
config_path: Option<PathBuf>,
selected: Option<PathBuf>,
) -> Result<Option<Arc<OwnerPublication>>> {
let socket_path = owner_work(move || match selected {
Some(path) => Ok(path),
None => default_socket_path().map_err(Into::into),
})
.await?;
#[cfg(unix)]
let directory = socket_path
.parent()
.context("owner endpoint has no private parent")?
.to_path_buf();
#[cfg(windows)]
let directory = owner_work(crate::daemon_windows::owner_directory).await?;
#[cfg(unix)]
let name = format!(
"{}.owner.json",
socket_path
.file_name()
.and_then(|name| name.to_str())
.context("invalid owner endpoint basename")?
);
#[cfg(windows)]
let name = "daemon.owner.json".to_string();
owner_work(move || {
let parent = match PrivateDirectory::inspect(&directory) {
Ok(parent) => Arc::new(parent),
Err(error)
if error
.downcast_ref::<std::io::Error>()
.is_some_and(|error| error.kind() == std::io::ErrorKind::NotFound) =>
{
return Ok(None);
}
Err(error) => return Err(error),
};
let Some((bytes, file)) = parent
.read_private_receipt(&name, 16384)
.map_err(owner_receipt_error)?
else {
return Ok(None);
};
let receipt: RuntimeOwnerReceipt = serde_json::from_slice(&bytes)?;
anyhow::ensure!(
receipt.version == 1
&& receipt.pid > 0
&& receipt.socket_path == socket_path
&& !receipt.lease_generation.is_empty(),
"selected owner receipt is invalid"
);
anyhow::ensure!(
receipt.config_path == config_path,
"selected owner config does not match; refusing another store"
);
#[cfg(unix)]
let socket_identity = parent.socket_identity(
socket_path
.file_name()
.and_then(|name| name.to_str())
.context("invalid owner endpoint basename")?,
)?;
Ok(Some(Arc::new(OwnerPublication {
parent,
receipt_name: name,
bytes,
receipt,
file,
#[cfg(unix)]
socket_path,
#[cfg(unix)]
socket_identity,
})))
})
.await
}
/// Read-only absence recheck after the actual Runtime lease was acquired.
pub async fn publication_is_absent(
config_path: Option<PathBuf>,
selected: Option<PathBuf>,
) -> Result<bool> {
Ok(observe_publication(config_path, selected).await?.is_none())
}
pub async fn probe_owner_for_host_startup(
config_path: Option<PathBuf>,
selected: Option<PathBuf>,
) -> Result<HostOwnerProbe> {
probe_frontend(
config_path,
selected,
crate::daemon_socket::AttachFrontend::Control,
None,
None,
None,
None,
)
.await
}
pub async fn connect(
config_path: Option<PathBuf>,
selected: Option<PathBuf>,
@@ -141,68 +372,65 @@ mod platform {
scope: Option<crate::RuntimeFrontendScope>,
acp_model: Option<String>,
) -> Result<Option<OwnerClient>> {
let socket_path = owner_work(move || match selected {
Some(path) => Ok(path),
None => default_socket_path().map_err(Into::into),
})
.await?;
#[cfg(unix)]
let directory = socket_path
.parent()
.context("owner endpoint has no private parent")?
.to_path_buf();
#[cfg(windows)]
let directory = owner_work(crate::daemon_windows::owner_directory).await?;
#[cfg(unix)]
let name = format!(
"{}.owner.json",
socket_path
.file_name()
.and_then(|name| name.to_str())
.context("invalid owner endpoint basename")?
);
#[cfg(windows)]
let name = "daemon.owner.json".to_string();
let receipt_name = name.clone();
let found = owner_work(move || {
let parent = match PrivateDirectory::inspect(&directory) {
Ok(parent) => Arc::new(parent),
Err(error)
if error
.downcast_ref::<std::io::Error>()
.is_some_and(|error| error.kind() == std::io::ErrorKind::NotFound) =>
{
return Ok(None);
}
Err(error) => return Err(error),
};
let Some((bytes, file)) = parent.read_private_receipt(&name, 16384)? else {
return Ok(None);
};
let receipt: RuntimeOwnerReceipt = serde_json::from_slice(&bytes)?;
Ok(Some((parent, receipt, file)))
})
.await?;
let Some((parent, receipt, file)) = found else {
return Ok(None);
match probe_frontend(
config_path,
selected,
frontend,
listener,
expected_owner,
scope,
acp_model,
)
.await?
{
HostOwnerProbe::Absent => Ok(None),
HostOwnerProbe::Attached(client) => Ok(Some(*client)),
HostOwnerProbe::Unavailable(publication) => Err(publication.strict_error()),
}
}
async fn probe_frontend(
config_path: Option<PathBuf>,
selected: Option<PathBuf>,
frontend: crate::daemon_socket::AttachFrontend,
listener: Option<crate::RuntimeListenerSelection>,
expected_owner: Option<RuntimeOwnerReceipt>,
scope: Option<crate::RuntimeFrontendScope>,
acp_model: Option<String>,
) -> Result<HostOwnerProbe> {
let Some(publication) = observe_publication(config_path, selected).await? else {
return Ok(HostOwnerProbe::Absent);
};
anyhow::ensure!(
receipt.version == 1
&& receipt.pid > 0
&& receipt.socket_path == socket_path
&& !receipt.lease_generation.is_empty(),
"selected owner receipt is invalid"
);
anyhow::ensure!(
receipt.config_path == config_path,
"selected owner config does not match; refusing another store"
);
#[cfg(unix)]
let socket_path = publication.socket_path.clone();
let receipt = publication.receipt.clone();
#[cfg(unix)]
let (stream, peer) = {
let stream =
tokio::time::timeout(Duration::from_secs(5), UnixStream::connect(&socket_path))
.await
.context("owner connect deadline expired")??;
let stream = match tokio::time::timeout(
Duration::from_secs(5),
UnixStream::connect(&socket_path),
)
.await
.context("owner connect deadline expired")?
{
Ok(stream) => stream,
Err(cause) => {
let held = publication.clone();
anyhow::ensure!(
owner_work(move || held.is_current()).await?,
PublicationChangedBeforeAttach
);
if cause.kind() == std::io::ErrorKind::ConnectionRefused
|| (cause.kind() == std::io::ErrorKind::NotFound
&& publication.socket_identity.is_none())
{
return Ok(HostOwnerProbe::Unavailable(Arc::new(
UnavailablePublication { publication, cause },
)));
}
return Err(cause.into());
}
};
let credential = stream
.peer_cred()
.context("owner peer credentials unavailable")?;
@@ -210,12 +438,17 @@ mod platform {
.pid()
.and_then(|pid| u32::try_from(pid).ok())
.context("owner peer PID unavailable")?;
anyhow::ensure!(
credential.uid() == PrivateDirectory::current_user_id()
&& receipt.principal == credential.uid().to_string()
&& pid == receipt.pid,
"connected process is not the selected Runtime owner"
);
if credential.uid() != PrivateDirectory::current_user_id()
|| receipt.principal != credential.uid().to_string()
|| pid != receipt.pid
{
let held = publication.clone();
anyhow::ensure!(
owner_work(move || held.is_current()).await?,
PublicationChangedBeforeAttach
);
bail!("connected process is not the selected Runtime owner");
}
(
stream,
Arc::new(AuthorizedPeer {
@@ -234,22 +467,13 @@ mod platform {
};
(stream, Arc::new(peer))
};
let held_parent = parent.clone();
let held_receipt = receipt.clone();
let held_publication = publication.clone();
let check = peer.clone();
owner_work(move || {
check.check()?;
anyhow::ensure!(
held_parent.is_at_selected_path()?,
"selected owner parent changed"
);
let (bytes, current) = held_parent
.read_private_receipt(&receipt_name, 16384)?
.context("selected owner receipt withdrawn")?;
anyhow::ensure!(
serde_json::from_slice::<RuntimeOwnerReceipt>(&bytes)? == held_receipt
&& PrivateDirectory::same_file_identity(&file, &current)?,
"selected owner receipt changed before attach"
held_publication.is_current()?,
PublicationChangedBeforeAttach
);
Ok(())
})
@@ -297,14 +521,14 @@ mod platform {
.filter(|value| !value.is_null())
.map(|value| serde_json::from_value(value.clone()))
.transpose()?;
Ok(Some(OwnerClient {
Ok(HostOwnerProbe::Attached(Box::new(OwnerClient {
read: replies,
write,
peer,
receipt,
routing,
frontend,
}))
})))
}
impl OwnerClient {
@@ -394,13 +618,34 @@ mod platform {
.forward(tokio::io::stdin(), tokio::io::stdout())
.await
}
#[cfg(all(test, unix))]
mod tests {
use super::*;
#[test]
fn owner_receipt_reobserves_only_typed_unlinked_file_refusals() {
let retired =
anyhow::Error::new(codewhale_config::private_directory::UnlinkedPrivateFile)
.context("bounded owner receipt read");
assert!(owner_receipt_error(retired).is::<PublicationChangedBeforeAttach>());
for refused in [
anyhow::anyhow!("private file was unlinked before validation"),
anyhow::anyhow!("xAI OAuth file must not have multiple filesystem links"),
std::io::Error::from(std::io::ErrorKind::PermissionDenied).into(),
anyhow::anyhow!("owner attach write deadline expired; outcome uncertain"),
] {
assert!(!owner_receipt_error(refused).is::<PublicationChangedBeforeAttach>());
}
}
}
}
#[cfg(any(unix, windows))]
pub use platform::{
OwnerClient, connect, connect_acp_if_published, connect_if_published,
connect_listener_if_published, connect_scoped_control_if_published,
connect_selected_acp_if_published, forward_stdio,
HostOwnerProbe, OwnerClient, PublicationChangedBeforeAttach, UnavailablePublication, connect,
connect_acp_if_published, connect_if_published, connect_listener_if_published,
connect_scoped_control_if_published, connect_selected_acp_if_published, forward_stdio,
probe_owner_for_host_startup, publication_is_absent,
};
#[cfg(not(any(unix, windows)))]
crates/app-server/src/daemon_socket.rs
+396 -39
View File
@@ -797,12 +797,20 @@ mod platform {
/// How long the stale-socket probe waits for a connect to resolve.
const PROBE_TIMEOUT: Duration = Duration::from_secs(1);
#[cfg(test)]
type BeforeOwnerRebind = Box<dyn FnOnce(&Path)>;
#[cfg(test)]
thread_local! {
static BEFORE_OWNER_REBIND: std::cell::RefCell<Option<BeforeOwnerRebind>> = const { std::cell::RefCell::new(None) };
}
/// Removes the socket file when the server stops, however it stops.
struct SocketFileGuard {
parent: Arc<PrivateDirectory>,
name: String,
identity: PrivateSocketIdentity,
receipt: Option<(String, File)>,
owner_state: Option<AppState>,
retired: bool,
}
@@ -816,7 +824,9 @@ mod platform {
let name = self.name.clone();
let identity = self.identity;
let receipt = self.receipt.take();
let owner_state = self.owner_state.take();
Some(move || {
let _held_owner = owner_state;
if let Some((name, file)) = receipt {
parent.retire_private_receipt(&name, &file)?;
}
@@ -835,6 +845,20 @@ mod platform {
}
}
// Blocking endpoint mutations retain the actual captured manager through
// its existing frontend, even after cancellation of the async caller.
async fn socket_owner_work<T, F>(owner_state: Option<AppState>, work: F) -> Result<T>
where
T: Send + 'static,
F: FnOnce() -> Result<T> + Send + 'static,
{
super::owner_work(move || {
let _held_owner = owner_state;
work()
})
.await
}
impl Drop for SocketFileGuard {
fn drop(&mut self) {
let Some(retire) = self.retirement() else {
@@ -939,41 +963,87 @@ mod platform {
pub async fn bind_daemon_socket(
options: DaemonSocketOptions,
) -> Result<DaemonSocket, DaemonSocketError> {
bind_with_owner(options, None).await
bind_with_owner(options, None, None).await
}
pub(crate) async fn bind_captured_owner(
state: AppState,
owner: super::RuntimeOwnerReceipt,
recovery: Option<Arc<crate::daemon_client::UnavailablePublication>>,
) -> Result<DaemonSocket, DaemonSocketError> {
let options = DaemonSocketOptions {
socket_path: Some(owner.socket_path.clone()),
config_path: None,
};
bind_with_owner(options, Some((state, owner))).await
bind_with_owner(options, Some((state, owner)), recovery).await
}
async fn bind_with_owner(
options: DaemonSocketOptions,
captured: Option<(AppState, super::RuntimeOwnerReceipt)>,
recovery: Option<Arc<crate::daemon_client::UnavailablePublication>>,
) -> Result<DaemonSocket, DaemonSocketError> {
let owner_state = captured.as_ref().map(|(state, _)| state.clone());
let path = resolve_socket_path(&SocketPathInputs::from_environment(options.socket_path)?)?;
let parent_path = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.ok_or(DaemonSocketError::RuntimeDirUnavailable)?
.to_path_buf();
let parent = Arc::new(
super::owner_work(move || PrivateDirectory::admit(&parent_path))
let parent = if let Some(previous) = recovery.as_ref() {
let owner = &captured
.as_ref()
.ok_or_else(|| {
DaemonSocketError::State(anyhow::anyhow!(
"stale publication requires the actual captured Runtime owner"
))
})?
.1;
let old = previous.receipt();
if old.version != owner.version
|| old.data_dir != owner.data_dir
|| old.execution_scope != owner.execution_scope
|| old.socket_path != path
|| old.config_path != owner.config_path
|| old.principal != owner.principal
{
return Err(DaemonSocketError::State(anyhow::anyhow!(
"captured stale owner belongs to another selected store/config/principal"
)));
}
if !previous
.revalidate()
.await
.map_err(DaemonSocketError::State)?
{
return Err(DaemonSocketError::State(anyhow::anyhow!(
"captured stale owner changed before bind"
)));
}
previous.publication.parent.clone()
} else {
Arc::new(
socket_owner_work(owner_state.clone(), move || {
PrivateDirectory::admit(&parent_path)
})
.await
.map_err(DaemonSocketError::State)?,
);
)
};
let name = path
.file_name()
.and_then(|name| name.to_str())
.ok_or(DaemonSocketError::RuntimeDirUnavailable)?
.to_string();
clear_stale_socket(&path, &parent, &name).await?;
clear_stale_socket(
&path,
&parent,
&name,
recovery.clone(),
captured.is_some() && recovery.is_none(),
owner_state.clone(),
)
.await?;
let check_parent = parent.clone();
if !super::owner_work(move || check_parent.is_at_selected_path())
.await
@@ -984,6 +1054,35 @@ mod platform {
)));
}
// Withdraw the exact dead publication before a new PID can answer on
// its endpoint. Contenders then re-observe absence and the held lease.
if let Some(previous) = recovery.as_ref() {
let previous = previous.clone();
let parent = parent.clone();
socket_owner_work(owner_state.clone(), move || {
anyhow::ensure!(
previous.revalidate_receipt()?,
"captured stale owner receipt changed before retirement"
);
anyhow::ensure!(
parent.retire_private_receipt(
&previous.publication.receipt_name,
&previous.publication.file
)?,
"captured stale owner receipt changed during retirement"
);
Ok(())
})
.await
.map_err(DaemonSocketError::State)?;
}
#[cfg(test)]
BEFORE_OWNER_REBIND.with(|slot| {
if let Some(observe) = slot.borrow_mut().take() {
observe(&path);
}
});
let listener = UnixListener::bind(&path).map_err(|source| DaemonSocketError::Io {
context: "failed to bind the daemon socket",
path: path.clone(),
@@ -1002,11 +1101,12 @@ mod platform {
name,
identity,
receipt: None,
owner_state: owner_state.clone(),
retired: false,
};
let protect_parent = parent.clone();
let protect_name = socket_file.name.clone();
super::owner_work(move || {
socket_owner_work(owner_state.clone(), move || {
anyhow::ensure!(
protect_parent.is_at_selected_path()?,
"private endpoint parent changed during bind"
@@ -1037,28 +1137,17 @@ mod platform {
}
let parent = parent.clone();
let name = receipt_name.clone();
let expected = owner.clone();
let guard = socket_file;
socket_file = super::owner_work(move || {
socket_file = socket_owner_work(owner_state, move || {
let mut guard = guard;
anyhow::ensure!(
parent.is_at_selected_path()?,
"private owner parent changed before publication"
);
if let Some((bytes, old)) = parent.read_private_receipt(&name, 16384)? {
let previous: super::RuntimeOwnerReceipt = serde_json::from_slice(&bytes)?;
anyhow::ensure!(
previous.version == expected.version
&& previous.data_dir == expected.data_dir
&& previous.execution_scope == expected.execution_scope
&& previous.socket_path == expected.socket_path,
"stale owner receipt belongs to a different selected store"
);
anyhow::ensure!(
parent.retire_private_receipt(&name, &old)?,
"stale owner receipt changed"
);
}
anyhow::ensure!(
parent.read_private_receipt(&name, 16384)?.is_none(),
"owner publication appeared during bootstrap; refusing replacement"
);
parent.write_owned_file(&name, &bytes, false)?;
let (_, file) = parent
.read_private_receipt(&name, 16384)?
@@ -1089,24 +1178,44 @@ mod platform {
path: &Path,
parent: &Arc<PrivateDirectory>,
name: &str,
recovery: Option<Arc<crate::daemon_client::UnavailablePublication>>,
expect_no_receipt: bool,
owner_state: Option<AppState>,
) -> Result<(), DaemonSocketError> {
let inspect_parent = parent.clone();
let inspect_name = name.to_string();
let Some(identity) =
super::owner_work(move || inspect_parent.socket_identity(&inspect_name))
.await
.map_err(|error| {
if error
.downcast_ref::<std::io::Error>()
.is_some_and(|error| error.kind() == std::io::ErrorKind::InvalidInput)
{
DaemonSocketError::NotASocket {
path: path.to_path_buf(),
}
} else {
DaemonSocketError::State(error)
}
})?
let expected = recovery.clone();
let Some(identity) = super::owner_work(move || {
if let Some(previous) = expected.as_ref() {
anyhow::ensure!(
previous.revalidate_all()?,
"captured stale owner/socket changed before retirement"
);
return Ok(previous.publication.socket_identity);
}
if expect_no_receipt {
anyhow::ensure!(
inspect_parent
.read_private_receipt(&format!("{inspect_name}.owner.json"), 16384)?
.is_none(),
"owner publication appeared before fresh endpoint bind; refusing replacement"
);
}
inspect_parent.socket_identity(&inspect_name)
})
.await
.map_err(|error| {
if error
.downcast_ref::<std::io::Error>()
.is_some_and(|error| error.kind() == std::io::ErrorKind::InvalidInput)
{
DaemonSocketError::NotASocket {
path: path.to_path_buf(),
}
} else {
DaemonSocketError::State(error)
}
})?
else {
return Ok(());
};
@@ -1117,7 +1226,15 @@ mod platform {
Ok(Err(error)) if error.kind() == std::io::ErrorKind::ConnectionRefused => {
let retire_parent = parent.clone();
let retire_name = name.to_string();
if super::owner_work(move || retire_parent.retire_socket(&retire_name, identity))
if socket_owner_work(owner_state, move || {
if let Some(previous) = recovery.as_ref() {
anyhow::ensure!(previous.revalidate_all()?, "captured stale owner/socket changed before retirement");
} else if expect_no_receipt {
anyhow::ensure!(retire_parent.read_private_receipt(&format!("{retire_name}.owner.json"), 16384)?.is_none(),
"owner publication appeared before fresh endpoint retirement; refusing replacement");
}
retire_parent.retire_socket(&retire_name, identity)
})
.await
.map_err(DaemonSocketError::State)?
{
@@ -1168,6 +1285,246 @@ mod platform {
#[cfg(test)]
mod tests {
use super::*;
use anyhow::Context as _;
struct FrontendLifetime {
_held: Arc<()>,
}
impl crate::RuntimeOwnerFrontend for FrontendLifetime {
fn validate_selection<'a>(
&'a self,
_selection: &'a crate::RuntimeOwnerFrontendSelection,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>>
{
Box::pin(async { anyhow::bail!("lifetime-only frontend cannot admit a command") })
}
fn serve(
&self,
_selection: crate::RuntimeOwnerFrontendSelection,
_compatibility: AppState,
_input: Box<dyn tokio::io::AsyncBufRead + Send + Unpin>,
_output: Box<dyn tokio::io::AsyncWrite + Send + Unpin>,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send + '_>>
{
Box::pin(async { anyhow::bail!("lifetime-only frontend cannot serve a command") })
}
}
#[tokio::test]
async fn owner_bootstrap_cancelled_mutations_and_retirement_retain_captured_frontend()
-> Result<()> {
// Proves transport custody. The TUI bootstrap tests separately
// prove the captured manager retains its real store lease.
let temporary = tempfile::Builder::new()
.prefix("cw-ob-cancel-")
.tempdir_in("/tmp")?;
let root = temporary.path().canonicalize()?;
let config = root.join("config.toml");
std::fs::write(&config, "")?;
for retirement in [false, true] {
let mut state =
build_state_off_runtime(Some(config.clone()), None, AppTransport::Socket)
.await?;
let held = Arc::new(());
let observed_owner = Arc::downgrade(&held);
state.owner_frontend = Some(Arc::new(FrontendLifetime { _held: held }));
let parent_path = root.join(if retirement { "retirement" } else { "mutation" });
let parent = Arc::new(PrivateDirectory::admit(&parent_path)?);
let path = parent_path.join("daemon.sock");
let socket = std::os::unix::net::UnixListener::bind(&path)?;
let identity = parent
.socket_identity("daemon.sock")?
.context("socket identity")?;
parent.protect_socket("daemon.sock", identity)?;
drop(socket);
parent.write_owned_file("daemon.sock.owner.json", b"{}", false)?;
let (_, receipt_file) = parent
.read_private_receipt("daemon.sock.owner.json", 16384)?
.context("receipt")?;
let (release, released) = std::sync::mpsc::channel();
let (started, observed_start) = tokio::sync::oneshot::channel();
let (finished, observed_finish) = tokio::sync::oneshot::channel();
let worker = if retirement {
let mut guard = SocketFileGuard {
parent,
name: "daemon.sock".into(),
identity,
receipt: Some(("daemon.sock.owner.json".into(), receipt_file)),
owner_state: Some(state.clone()),
retired: false,
};
let retire = guard.retirement().context("first retirement")?;
drop(guard);
tokio::spawn(super::super::owner_work(move || {
let _ = started.send(());
let _ = released.recv();
let result = retire();
let _ = finished.send(());
result
}))
} else {
tokio::spawn(socket_owner_work(Some(state.clone()), move || {
let _ = started.send(());
let _ = released.recv();
parent.protect_socket("daemon.sock", identity)?;
let _ = finished.send(());
Ok(())
}))
};
drop(state);
tokio::time::timeout(Duration::from_secs(5), observed_start).await??;
worker.abort();
let _ = worker.await;
assert!(
observed_owner.upgrade().is_some(),
"running filesystem mutation retains captured owner after waiter cancellation"
);
release.send(())?;
tokio::time::timeout(Duration::from_secs(5), observed_finish).await??;
tokio::time::timeout(Duration::from_secs(5), async {
while observed_owner.upgrade().is_some() {
tokio::task::yield_now().await;
}
})
.await?;
if retirement {
assert!(!path.exists());
assert!(!path.with_file_name("daemon.sock.owner.json").exists());
} else {
assert!(path.exists(), "protection alone preserves publication");
assert!(path.with_file_name("daemon.sock.owner.json").exists());
}
}
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn owner_bootstrap_withdraws_dead_receipt_before_replacement_listener_can_answer()
-> Result<()> {
struct OwnedChild(std::process::Child);
impl Drop for OwnedChild {
fn drop(&mut self) {
if self.0.try_wait().ok().flatten().is_none() {
let _ = self.0.kill();
}
let _ = self.0.wait();
}
}
let temporary = tempfile::Builder::new()
.prefix("cw-ob-order-")
.tempdir_in("/tmp")?;
let root = temporary.path().canonicalize()?;
let path = root.join("run/daemon.sock");
let config = root.join("config.toml");
std::fs::write(&config, "")?;
let parent = PrivateDirectory::admit(path.parent().unwrap())?;
let stale = std::os::unix::net::UnixListener::bind(&path)?;
let identity = parent.socket_identity("daemon.sock")?.unwrap();
parent.protect_socket("daemon.sock", identity)?;
drop(stale);
let mut child = OwnedChild(
std::process::Command::new("/bin/sleep")
.arg("60")
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()?,
);
let mut owner = codewhale_protocol::RuntimeOwnerReceipt {
version: 1,
data_dir: root.join("runtime"),
execution_scope: "ordering-store".into(),
lease_generation: "ordering-old".into(),
pid: child.0.id(),
process_start: super::super::capture_process_start(child.0.id()).await?,
principal: PrivateDirectory::current_user_id().to_string(),
socket_path: path.clone(),
config_path: Some(config.clone()),
};
child.0.kill()?;
child.0.wait()?;
parent.write_owned_file(
"daemon.sock.owner.json",
&serde_json::to_vec(&owner)?,
false,
)?;
let crate::daemon_client::HostOwnerProbe::Unavailable(captured) =
crate::daemon_client::probe_owner_for_host_startup(
Some(config.clone()),
Some(path.clone()),
)
.await?
else {
anyhow::bail!("dead owner must be captured")
};
let observed = Arc::new(std::sync::atomic::AtomicBool::new(false));
let observed_hook = observed.clone();
let selected = path.clone();
let selected_config = config.clone();
let stale_capture = captured.clone();
BEFORE_OWNER_REBIND.with(|slot| {
*slot.borrow_mut() = Some(Box::new(move |endpoint| {
assert_eq!(endpoint, selected);
assert!(!selected.with_file_name("daemon.sock.owner.json").exists());
let endpoint = selected.clone();
std::thread::spawn(move || {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
.block_on(async {
assert!(
!stale_capture.revalidate().await.unwrap(),
"withdrawn capture must never authorize a replacement"
);
assert!(matches!(
crate::daemon_client::probe_owner_for_host_startup(
Some(selected_config),
Some(endpoint),
)
.await
.unwrap(),
crate::daemon_client::HostOwnerProbe::Absent
));
});
})
.join()
.unwrap();
// Force bind failure after the exact observation point;
// the product must preserve this task-owned replacement.
drop(std::os::unix::net::UnixListener::bind(&selected).unwrap());
observed_hook.store(true, std::sync::atomic::Ordering::SeqCst);
}));
});
owner.pid = std::process::id();
owner.process_start = super::super::capture_process_start(owner.pid).await?;
owner.lease_generation = "ordering-new".into();
let result = crate::bind_runtime_frontends(
Some(config),
None,
owner,
crate::RuntimeOwnerRouting {
endpoint: "127.0.0.1:1".parse()?,
workspace: None,
workers: None,
mobile: false,
web: false,
acp: false,
acp_only: false,
},
None,
Some(captured),
)
.await;
assert!(result.is_err(), "injected replacement prevents new bind");
assert!(observed.load(std::sync::atomic::Ordering::SeqCst));
assert!(
parent.socket_identity("daemon.sock")?.is_some(),
"replacement endpoint preserved"
);
assert!(!path.with_file_name("daemon.sock.owner.json").exists());
Ok(())
}
fn client(name: &str) -> ClientIdentity {
ClientIdentity {
crates/app-server/src/lib.rs
+12 -1
View File
@@ -610,6 +610,7 @@ pub async fn bind_runtime_owner(
acp_only: false,
},
None,
None,
)
.await?;
Ok(daemon)
@@ -623,6 +624,7 @@ pub async fn bind_runtime_frontends(
owner: codewhale_protocol::RuntimeOwnerReceipt,
routing: RuntimeOwnerRouting,
owner_frontend: Option<Arc<dyn RuntimeOwnerFrontend>>,
recovery: Option<Arc<daemon_client::UnavailablePublication>>,
) -> Result<(daemon_socket::DaemonSocket, AppState)> {
anyhow::ensure!(
owner.version == 1 && owner.pid == std::process::id() && !owner.lease_generation.is_empty(),
@@ -660,7 +662,16 @@ pub async fn bind_runtime_frontends(
state.captured_routing = Some(routing);
state.owner_frontend = owner_frontend;
*state.runtime_bridge.lock().await = Some(Arc::new(Mutex::new(bridge)));
let daemon = daemon_socket::bind_captured_owner(state.clone(), owner).await?;
#[cfg(unix)]
let daemon = daemon_socket::bind_captured_owner(state.clone(), owner, recovery).await?;
#[cfg(windows)]
let daemon = {
anyhow::ensure!(
recovery.is_none(),
"Windows stale-owner recovery is not qualified; refusing fallback"
);
daemon_socket::bind_captured_owner(state.clone(), owner).await?
};
Ok((daemon, state))
}
crates/app-server/tests/daemon_socket.rs
+272
View File
@@ -85,6 +85,277 @@ impl Drop for Harness {
}
}
struct OwnedBootstrapChild(std::process::Child);
impl Drop for OwnedBootstrapChild {
fn drop(&mut self) {
if self.0.try_wait().ok().flatten().is_none() {
let _ = self.0.kill();
}
let _ = self.0.wait();
}
}
/// Transport-only publication fixture; actual store-lease proof lives in TUI.
struct BootstrapPublicationFixture {
harness: Harness,
parent: codewhale_config::private_directory::PrivateDirectory,
owner: codewhale_protocol::RuntimeOwnerReceipt,
bytes: Vec<u8>,
}
impl BootstrapPublicationFixture {
async fn new(label: &str, dead: bool, reused: bool) -> Self {
use codewhale_config::private_directory::PrivateDirectory;
let harness = Harness::new(label);
let parent = PrivateDirectory::admit(harness.socket_path.parent().unwrap()).unwrap();
let socket = std::os::unix::net::UnixListener::bind(&harness.socket_path).unwrap();
let identity = parent.socket_identity("daemon.sock").unwrap().unwrap();
parent.protect_socket("daemon.sock", identity).unwrap();
drop(socket);
let (pid, mut process_start) = if dead {
let mut child = OwnedBootstrapChild(
std::process::Command::new("/bin/sleep")
.arg("60")
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap(),
);
let pid = child.0.id();
let start = codewhale_app_server::daemon_socket::capture_process_start(pid)
.await
.unwrap();
child.0.kill().unwrap();
child.0.wait().unwrap();
(pid, start)
} else {
(
std::process::id(),
codewhale_app_server::daemon_socket::capture_process_start(std::process::id())
.await
.unwrap(),
)
};
if reused {
process_start.push_str("-different-generation");
}
let owner = codewhale_protocol::RuntimeOwnerReceipt {
version: 1,
data_dir: harness.root.join("runtime"),
execution_scope: "bootstrap-transport-store".into(),
lease_generation: "bootstrap-dead-generation".into(),
pid,
process_start,
principal: PrivateDirectory::current_user_id().to_string(),
socket_path: harness.socket_path.clone(),
config_path: harness.options().config_path,
};
let bytes = serde_json::to_vec(&owner).unwrap();
parent
.write_owned_file("daemon.sock.owner.json", &bytes, false)
.unwrap();
Self {
harness,
parent,
owner,
bytes,
}
}
async fn capture(
&self,
) -> std::sync::Arc<codewhale_app_server::daemon_client::UnavailablePublication> {
match codewhale_app_server::daemon_client::probe_owner_for_host_startup(
self.owner.config_path.clone(),
Some(self.owner.socket_path.clone()),
)
.await
.unwrap()
{
codewhale_app_server::daemon_client::HostOwnerProbe::Unavailable(publication) => {
publication
}
_ => panic!("refused private endpoint must be captured before bootstrap"),
}
}
async fn bind_replacement(
&self,
recovery: Option<
std::sync::Arc<codewhale_app_server::daemon_client::UnavailablePublication>,
>,
) -> anyhow::Result<codewhale_app_server::daemon_socket::DaemonSocket> {
let mut owner = self.owner.clone();
owner.pid = std::process::id();
owner.process_start =
codewhale_app_server::daemon_socket::capture_process_start(owner.pid).await?;
owner.lease_generation = "bootstrap-new-generation".into();
let (daemon, _) = codewhale_app_server::bind_runtime_frontends(
owner.config_path.clone(),
None,
owner,
codewhale_app_server::RuntimeOwnerRouting {
endpoint: "127.0.0.1:1".parse().unwrap(),
workspace: None,
workers: None,
mobile: false,
web: false,
acp: false,
acp_only: false,
},
None,
recovery,
)
.await?;
Ok(daemon)
}
fn assert_publication(&self, bytes: &[u8]) {
assert_eq!(
self.parent
.read_private_receipt("daemon.sock.owner.json", 16384)
.unwrap()
.unwrap()
.0,
bytes
);
}
}
#[tokio::test]
async fn owner_bootstrap_unavailable_is_host_only_and_read_only() {
let fixture = BootstrapPublicationFixture::new("boot-read", true, false).await;
let identity = fixture.parent.socket_identity("daemon.sock").unwrap();
let captured = fixture.capture().await;
assert_eq!(captured.receipt(), &fixture.owner);
assert!(captured.revalidate().await.unwrap());
assert!(
codewhale_app_server::daemon_client::connect_if_published(
fixture.owner.config_path.clone(),
Some(fixture.owner.socket_path.clone()),
)
.await
.is_err()
);
assert!(
codewhale_app_server::daemon_client::connect(
fixture.owner.config_path.clone(),
Some(fixture.owner.socket_path.clone()),
)
.await
.is_err()
);
fixture.assert_publication(&fixture.bytes);
assert_eq!(
fixture.parent.socket_identity("daemon.sock").unwrap(),
identity
);
assert!(
!fixture.owner.data_dir.exists(),
"standalone control must not open or mint a store"
);
}
#[tokio::test]
async fn owner_bootstrap_live_process_and_reused_pid_refuse_retirement() {
for reused in [false, true] {
let fixture = BootstrapPublicationFixture::new("boot-live", false, reused).await;
let captured = fixture.capture().await;
let error = captured.revalidate().await.err().unwrap().to_string();
assert!(
error.contains(if reused { "reused" } else { "still present" }),
"{error}"
);
assert!(fixture.bind_replacement(Some(captured)).await.is_err());
fixture.assert_publication(&fixture.bytes);
assert!(
fixture
.parent
.socket_identity("daemon.sock")
.unwrap()
.is_some()
);
}
}
#[tokio::test]
async fn owner_bootstrap_equal_bytes_do_not_authorize_replaced_receipt() {
let fixture = BootstrapPublicationFixture::new("boot-file", true, false).await;
let captured = fixture.capture().await;
let identity = fixture.parent.socket_identity("daemon.sock").unwrap();
std::fs::rename(
fixture
.harness
.socket_path
.with_file_name("daemon.sock.owner.json"),
fixture.harness.socket_path.with_file_name("old.owner.json"),
)
.unwrap();
fixture
.parent
.write_owned_file("daemon.sock.owner.json", &fixture.bytes, false)
.unwrap();
assert!(!captured.revalidate().await.unwrap());
assert!(fixture.bind_replacement(Some(captured)).await.is_err());
fixture.assert_publication(&fixture.bytes);
assert_eq!(
fixture.parent.socket_identity("daemon.sock").unwrap(),
identity
);
}
#[tokio::test]
async fn owner_bootstrap_same_inode_content_change_is_preserved() {
let fixture = BootstrapPublicationFixture::new("boot-edit", true, false).await;
let captured = fixture.capture().await;
let mut edited = fixture.bytes.clone();
edited.push(b'\n');
std::fs::write(
fixture
.harness
.socket_path
.with_file_name("daemon.sock.owner.json"),
&edited,
)
.unwrap();
assert!(!captured.revalidate().await.unwrap());
assert!(fixture.bind_replacement(Some(captured)).await.is_err());
fixture.assert_publication(&edited);
}
#[tokio::test]
async fn owner_bootstrap_replaced_socket_is_preserved() {
let fixture = BootstrapPublicationFixture::new("boot-sock", true, false).await;
let captured = fixture.capture().await;
std::fs::rename(
&fixture.harness.socket_path,
fixture.harness.socket_path.with_file_name("old.sock"),
)
.unwrap();
let replacement = std::os::unix::net::UnixListener::bind(&fixture.harness.socket_path).unwrap();
let identity = fixture.parent.socket_identity("daemon.sock").unwrap();
drop(replacement);
assert!(!captured.revalidate().await.unwrap());
assert!(fixture.bind_replacement(Some(captured)).await.is_err());
assert_eq!(
fixture.parent.socket_identity("daemon.sock").unwrap(),
identity
);
fixture.assert_publication(&fixture.bytes);
}
#[tokio::test]
async fn owner_bootstrap_fresh_bind_cannot_retire_uncaptured_publication() {
let fixture = BootstrapPublicationFixture::new("boot-fresh", true, false).await;
let identity = fixture.parent.socket_identity("daemon.sock").unwrap();
assert!(fixture.bind_replacement(None).await.is_err());
fixture.assert_publication(&fixture.bytes);
assert_eq!(
fixture.parent.socket_identity("daemon.sock").unwrap(),
identity
);
}
struct Client {
reader: BufReader<OwnedReadHalf>,
writer: OwnedWriteHalf,
@@ -790,6 +1061,7 @@ async fn canonical_cli_scoped_resume_fork_keep_exact_owner_workers_and_refuse_wr
acp_only: false,
},
Some(frontend),
None,
)
.await
.unwrap();
crates/config/src/private_directory.rs
+220 -43
View File
@@ -26,6 +26,22 @@ pub struct PrivateDirectory {
pub(crate) _component_handles: Vec<File>,
}
/// A retained file was retired before it could be validated. This is still a
/// refusal; an owner observer may re-observe, but must never accept this file.
#[cfg(unix)]
#[derive(Debug)]
pub struct UnlinkedPrivateFile;
#[cfg(unix)]
impl std::fmt::Display for UnlinkedPrivateFile {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("private file was unlinked before validation")
}
}
#[cfg(unix)]
impl std::error::Error for UnlinkedPrivateFile {}
impl PrivateDirectory {
pub fn open(directory: &Path) -> Result<Self> {
open_owned_directory(directory, true, true)
@@ -287,38 +303,45 @@ impl PrivateDirectory {
.context("atomically replacing xAI OAuth credentials");
}
} else {
// `linkat` installs the unique generation without clobbering an
// existing path. The temporary link is removed immediately.
// SAFETY: all descriptors/names remain valid for both calls.
if unsafe {
libc::linkat(
self.directory_handle.as_raw_fd(),
temporary.as_ptr(),
self.directory_handle.as_raw_fd(),
target.as_ptr(),
0,
)
} != 0
#[cfg(any(target_os = "macos", target_os = "linux"))]
// Exclusive rename publishes one link atomically. A hard-link
// publication briefly exposes two links to concurrent readers.
self.move_no_replace(&temp_name, name)?;
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
{
return Err(std::io::Error::last_os_error())
.context("installing a new xAI OAuth generation without replacement");
}
// SAFETY: same descriptor and staging name as the `linkat` above.
if unsafe {
libc::unlinkat(self.directory_handle.as_raw_fd(), temporary.as_ptr(), 0)
} != 0
{
let error = std::io::Error::last_os_error();
// The target and staging name still reference the same
// inode. Remove the just-installed target so the generic
// error cleanup can safely retire the single remaining
// staging link instead of leaving an inert secret with
// link count two.
// SAFETY: same descriptor; `target` was just installed above.
unsafe {
libc::unlinkat(self.directory_handle.as_raw_fd(), target.as_ptr(), 0)
};
return Err(error).context("removing xAI OAuth generation staging link");
// `linkat` installs the unique generation without clobbering an
// existing path. The temporary link is removed immediately.
// SAFETY: all descriptors/names remain valid for both calls.
if unsafe {
libc::linkat(
self.directory_handle.as_raw_fd(),
temporary.as_ptr(),
self.directory_handle.as_raw_fd(),
target.as_ptr(),
0,
)
} != 0
{
return Err(std::io::Error::last_os_error())
.context("installing a new xAI OAuth generation without replacement");
}
// SAFETY: same descriptor and staging name as the `linkat` above.
if unsafe {
libc::unlinkat(self.directory_handle.as_raw_fd(), temporary.as_ptr(), 0)
} != 0
{
let error = std::io::Error::last_os_error();
// The target and staging name still reference the same
// inode. Remove the just-installed target so the generic
// error cleanup can safely retire the single remaining
// staging link instead of leaving an inert secret with
// link count two.
// SAFETY: same descriptor; `target` was just installed above.
unsafe {
libc::unlinkat(self.directory_handle.as_raw_fd(), target.as_ptr(), 0)
};
return Err(error).context("removing xAI OAuth generation staging link");
}
}
}
self.directory_handle
@@ -401,6 +424,7 @@ pub(crate) fn validate_owned_file_handle(file: &File, path: &Path) -> Result<fs:
metadata.uid() == unsafe { libc::geteuid() },
"xAI OAuth file must be owned by the current user"
);
anyhow::ensure!(metadata.nlink() != 0, UnlinkedPrivateFile);
anyhow::ensure!(
metadata.nlink() == 1,
"xAI OAuth file must not have multiple filesystem links"
@@ -1286,7 +1310,7 @@ impl PrivateDirectory {
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
fn move_socket_no_replace(&self, from: &str, to: &str) -> Result<()> {
fn move_no_replace(&self, from: &str, to: &str) -> Result<()> {
use std::os::fd::AsRawFd as _;
validate_private_basename(from)?;
validate_private_basename(to)?;
@@ -1318,13 +1342,13 @@ impl PrivateDirectory {
};
if result != 0 {
return Err(std::io::Error::last_os_error())
.context("retiring endpoint without replacement");
.context("moving private entry without replacement");
}
Ok(())
}
#[cfg(not(any(target_os = "macos", target_os = "linux")))]
fn move_socket_no_replace(&self, _from: &str, _to: &str) -> Result<()> {
fn move_no_replace(&self, _from: &str, _to: &str) -> Result<()> {
bail!("exclusive endpoint retirement is unsupported on this platform")
}
@@ -1340,9 +1364,9 @@ impl PrivateDirectory {
SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos(),
COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
);
self.move_socket_no_replace(name, &retired)?;
self.move_no_replace(name, &retired)?;
if self.socket_identity(&retired)? != Some(expected) {
let restored = self.move_socket_no_replace(&retired, name);
let restored = self.move_no_replace(&retired, name);
return Err(anyhow::anyhow!(
"endpoint identity changed; replacement preserved (restore: {restored:?})"
));
@@ -1400,6 +1424,75 @@ mod tests {
assert!(parent.read_private_receipt("alias.json", 8).is_err());
}
#[test]
fn owner_receipt_opened_before_retirement_is_rejected_as_unlinked_not_hardlinked() {
let root = root();
let parent = PrivateDirectory::admit(&root.path().join("run")).unwrap();
parent
.write_owned_file("owner.json", b"receipt", false)
.unwrap();
let file = parent
.open_owned_file_for_read("owner.json")
.unwrap()
.unwrap();
assert!(parent.remove_raw("owner.json").unwrap());
let error = parent
.read_private_receipt_file("owner.json", 8, file)
.unwrap_err();
assert!(error.is::<UnlinkedPrivateFile>());
parent
.write_owned_file("owner.json", b"receipt", false)
.unwrap();
fs::hard_link(
parent.directory.join("owner.json"),
parent.directory.join("alias.json"),
)
.unwrap();
let error = parent.read_private_receipt("owner.json", 8).unwrap_err();
assert!(
!error.is::<UnlinkedPrivateFile>(),
"a linked receipt must remain unsafe, not retryable"
);
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
#[test]
fn private_exclusive_publication_keeps_one_link_and_preserves_existing_destination() {
use std::os::unix::fs::MetadataExt as _;
let root = root();
let parent = PrivateDirectory::admit(&root.path().join("run")).unwrap();
let mut staging = parent.open_internal_file("staging").unwrap();
staging.write_all(b"receipt").unwrap();
staging.sync_all().unwrap();
parent.move_no_replace("staging", "owner.json").unwrap();
assert_eq!(staging.metadata().unwrap().nlink(), 1);
assert!(
parent
.open_owned_file_for_read("staging")
.unwrap()
.is_none()
);
assert!(
parent
.write_owned_file("owner.json", b"replacement", false)
.is_err()
);
assert_eq!(
parent
.read_private_receipt("owner.json", 8)
.unwrap()
.unwrap()
.0,
b"receipt"
);
assert_eq!(
fs::read_dir(&parent.directory).unwrap().count(),
1,
"failed publication must clean only its own staging file"
);
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
#[test]
fn owner_receipt_retirement_preserves_a_replacement_and_uses_captured_file() {
@@ -1456,6 +1549,45 @@ mod tests {
assert!(unix_process_start(u32::MAX).is_err());
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
#[test]
fn unix_process_status_requires_positive_kernel_absence() {
assert_eq!(
unix_process_status(std::process::id()).unwrap(),
UnixProcessStatus::Present {
start: unix_process_start(std::process::id()).unwrap()
}
);
let mut child = std::process::Command::new("/bin/sh")
.args(["-c", "exit 0"])
.spawn()
.unwrap();
let pid = child.id();
child.wait().unwrap();
assert_eq!(unix_process_status(pid).unwrap(), UnixProcessStatus::Absent);
assert!(unix_process_status(0).is_err());
assert!(unix_process_status(u32::MAX).is_err());
}
#[cfg(unix)]
#[test]
fn unix_process_status_never_converts_uncertainty_to_death() {
assert!(unix_pid_is_absent(
-1,
&std::io::Error::from_raw_os_error(libc::ESRCH)
));
for code in [libc::EPERM, libc::EACCES, libc::EINVAL] {
assert!(!unix_pid_is_absent(
-1,
&std::io::Error::from_raw_os_error(code)
));
}
assert!(!unix_pid_is_absent(
0,
&std::io::Error::from_raw_os_error(libc::ESRCH)
));
}
#[cfg(target_os = "linux")]
#[test]
fn linux_process_generation_survives_nondumpable_hardening() {
@@ -1584,7 +1716,7 @@ mod tests {
let destination_identity = parent.socket_identity("retired.sock").unwrap().unwrap();
let error = parent
.move_socket_no_replace("owner.sock", "retired.sock")
.move_no_replace("owner.sock", "retired.sock")
.unwrap_err();
assert_eq!(
error.downcast_ref::<std::io::Error>().unwrap().kind(),
@@ -1630,6 +1762,42 @@ mod tests {
}
}
/// A current-user process identity or positive kernel absence. Query failures
/// remain errors; an unreadable process record never establishes death.
#[cfg(unix)]
#[derive(Debug, PartialEq, Eq)]
pub enum UnixProcessStatus {
Present { start: String },
Absent,
}
#[cfg(unix)]
fn unix_pid_is_absent(result: i32, error: &std::io::Error) -> bool {
result == -1 && error.raw_os_error() == Some(libc::ESRCH)
}
/// Read on the existing bounded owner worker. Signal zero never terminates a
/// process; only ESRCH after validating a positive PID proves absence.
#[cfg(unix)]
pub fn unix_process_status(pid: u32) -> Result<UnixProcessStatus> {
let kernel_pid = i32::try_from(pid).context("invalid local process PID")?;
anyhow::ensure!(kernel_pid > 0, "invalid local process PID");
match unix_process_start(pid) {
Ok(start) => Ok(UnixProcessStatus::Present { start }),
Err(identity_error) => {
// SAFETY: a validated positive PID and signal zero only query
// existence. No process-group selection or signal is possible.
let result = unsafe { libc::kill(kernel_pid, 0) };
let error = std::io::Error::last_os_error();
if unix_pid_is_absent(result, &error) {
Ok(UnixProcessStatus::Absent)
} else {
Err(identity_error).context("selected process identity/liveness is uncertain")
}
}
}
}
/// Kernel process creation identity for the authenticated local Unix peer.
/// Run on the existing bounded owner worker; PID alone is never sufficient.
#[cfg(unix)]
@@ -1770,11 +1938,20 @@ impl PrivateDirectory {
/// The same anchored owned-file opener used by credentials, with a bounded
/// read and retained object for exact receipt retirement.
pub fn read_private_receipt(&self, name: &str, max: usize) -> Result<Option<(Vec<u8>, File)>> {
use std::io::Read as _;
use std::os::unix::fs::MetadataExt as _;
let Some(mut file) = self.open_owned_file_for_read(name)? else {
let Some(file) = self.open_owned_file_for_read(name)? else {
return Ok(None);
};
self.read_private_receipt_file(name, max, file).map(Some)
}
fn read_private_receipt_file(
&self,
name: &str,
max: usize,
mut file: File,
) -> Result<(Vec<u8>, File)> {
use std::io::Read as _;
use std::os::unix::fs::MetadataExt as _;
let before = validate_owned_file_handle(&file, &self.directory.join(name))?;
anyhow::ensure!(
before.mode() & 0o077 == 0 && before.len() <= u64::try_from(max)?,
@@ -1797,7 +1974,7 @@ impl PrivateDirectory {
&& before.mtime_nsec() == after.mtime_nsec(),
"private owner receipt changed while reading"
);
Ok(Some((bytes, file)))
Ok((bytes, file))
}
pub fn retire_private_receipt(&self, name: &str, captured: &File) -> Result<bool> {
@@ -1815,13 +1992,13 @@ impl PrivateDirectory {
std::process::id(),
SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos()
);
self.move_socket_no_replace(name, &retired)?;
self.move_no_replace(name, &retired)?;
let Some(moved) = self.open_owned_file_for_read(&retired)? else {
bail!("private receipt retirement is uncertain")
};
let moved_metadata = validate_owned_file_handle(&moved, &self.directory.join(&retired))?;
if (expected.dev(), expected.ino()) != (moved_metadata.dev(), moved_metadata.ino()) {
let restored = self.move_socket_no_replace(&retired, name);
let restored = self.move_no_replace(&retired, name);
bail!("private receipt changed; retaining replacement (restore: {restored:?})");
}
self.remove_raw(&retired)?;
crates/tui/src/acp_server.rs
+5 -1
View File
@@ -1183,11 +1183,14 @@ mod tests {
};
let plugins = Arc::new(crate::plugins::PluginRegistry::empty(&workspace));
let manager = Arc::new(if acp {
RuntimeThreadManager::open_acp(
RuntimeThreadManager::open_host(
config.clone(),
workspace.clone(),
manager_config,
plugins,
crate::core::engine::EngineHostProfile::Acp,
None,
None,
)?
} else {
RuntimeThreadManager::open_with_plugin_registry(
@@ -1367,6 +1370,7 @@ mod tests {
acp_only: base_acp,
},
Some(frontend),
None,
)
.await?;
let shutdown = daemon.shutdown_handle();
crates/tui/src/commands/groups/plugins/mod.rs
+2 -2
View File
@@ -874,8 +874,8 @@ fn update_bundle(
PluginMutationOutcome::Updated => {
let name = receipt.name.clone();
let mut output = format!(
"Updated plugin '{name}'. Its content changed, so the previous trust receipt no \
longer matches — review and trust it again before enabling.\n"
"Updated plugin '{name}'. Review its current capabilities and trust it \
if needed before enabling.\n"
);
if let Some(review) = review_bundle(presentation, plugin, &name).message {
output.push('\n');
crates/tui/src/commands/groups/plugins/tests.rs
+27 -3
View File
@@ -603,10 +603,10 @@ fn install_update_uninstall_verbs_drive_the_guided_trust_flow() {
.exists()
);
// Local-path installs cannot be updated from the network.
// An unchanged local source succeeds without changing the review token.
let update = plugins_with_kimi_home_override(&mut app, Some("update installed-demo"), None);
assert!(update.is_error);
assert!(update.message.unwrap().contains("local path"));
assert!(!update.is_error, "{:?}", update.message);
assert!(update.message.unwrap().contains("already up to date"));
let arg = confirmation.trim_start_matches("/plugin ").to_string();
assert!(!plugins_with_kimi_home_override(&mut app, Some(&arg), None).is_error);
@@ -616,6 +616,30 @@ fn install_update_uninstall_verbs_drive_the_guided_trust_flow() {
);
assert!(app.plugin_registry.is_active("installed-demo"));
fs::write(
source.join("plugin.toml"),
"schema_version = 1\n[plugin]\nname = \"installed-demo\"\nversion = \"2.0.0\"\n",
)
.unwrap();
let updated =
plugins_with_kimi_home_override(&mut app, Some("update installed-demo"), None);
assert!(!updated.is_error, "{:?}", updated.message);
let message = updated.message.unwrap();
let current_confirmation = message
.lines()
.find(|line| line.starts_with("/plugin trust installed-demo "))
.expect("changed local content must route into its current review");
assert_ne!(current_confirmation, confirmation);
assert!(!app.plugin_registry.is_active("installed-demo"));
assert!(plugins_with_kimi_home_override(&mut app, Some(&arg), None).is_error);
let current_arg = current_confirmation.trim_start_matches("/plugin ");
assert!(!plugins_with_kimi_home_override(&mut app, Some(current_arg), None).is_error);
assert!(
!plugins_with_kimi_home_override(&mut app, Some("enable installed-demo"), None)
.is_error
);
assert!(app.plugin_registry.is_active("installed-demo"));
// Uninstall requires disabled, then removes bits and prunes state.
let refused =
plugins_with_kimi_home_override(&mut app, Some("uninstall installed-demo"), None);
crates/tui/src/plugins/builtin.rs
+3 -3
View File
@@ -430,7 +430,7 @@ fn verify_snapshot(root: &Path, files: &BTreeMap<PathBuf, &[u8]>) -> io::Result<
/// Atomic no-replace directory publication. A check followed by ordinary Unix
/// rename is insufficient: rename is allowed to replace an existing empty dir.
#[cfg(any(target_os = "macos", target_os = "linux"))]
fn publish_snapshot(source: &Path, destination: &Path) -> io::Result<()> {
pub(super) fn publish_snapshot(source: &Path, destination: &Path) -> io::Result<()> {
use std::ffi::CString;
use std::os::unix::ffi::OsStrExt as _;
@@ -461,7 +461,7 @@ fn publish_snapshot(source: &Path, destination: &Path) -> io::Result<()> {
}
#[cfg(windows)]
fn publish_snapshot(source: &Path, destination: &Path) -> io::Result<()> {
pub(super) fn publish_snapshot(source: &Path, destination: &Path) -> io::Result<()> {
use std::os::windows::ffi::OsStrExt as _;
use windows::Win32::Storage::FileSystem::{MOVEFILE_WRITE_THROUGH, MoveFileExW};
use windows::core::PCWSTR;
@@ -485,7 +485,7 @@ fn publish_snapshot(source: &Path, destination: &Path) -> io::Result<()> {
}
#[cfg(not(any(target_os = "macos", target_os = "linux", windows)))]
fn publish_snapshot(_source: &Path, _destination: &Path) -> io::Result<()> {
pub(super) fn publish_snapshot(_source: &Path, _destination: &Path) -> io::Result<()> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
"atomic built-in snapshot publication is unsupported on this platform",
crates/tui/src/plugins/discovery.rs
+24
View File
@@ -173,6 +173,23 @@ pub(crate) fn discover_with_context(
)
}
// These exact operation-name shapes are reserved by stage::fresh_staging_dir
// and place::finalize_install. Do not hide human paths such as demo.bak or
// .staging-not-a-uuid, which can legitimately contain reviewed bundles.
fn is_internal_publication_directory(name: &std::ffi::OsStr) -> bool {
let Some(name) = name.to_str() else {
return false;
};
name.strip_prefix(".staging-").is_some_and(|suffix| {
suffix.len() == 32
&& suffix
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}) || name.strip_prefix(".plugin-backup-").is_some_and(|suffix| {
suffix.len() == 16 && suffix.bytes().all(|byte| byte.is_ascii_alphanumeric())
})
}
fn scan_root(
root: &Path,
scope: PluginScope,
@@ -244,6 +261,9 @@ fn scan_root(
entries.sort_by_key(fs::DirEntry::file_name);
for entry in entries {
if is_internal_publication_directory(&entry.file_name()) {
continue;
}
let plugin_root = entry.path();
let Ok(metadata) = fs::symlink_metadata(&plugin_root) else {
continue;
@@ -837,3 +857,7 @@ mod tests {
assert!(!home.join("plugins/state.json").exists());
}
}
#[cfg(test)]
#[path = "discovery_internal_paths_tests.rs"]
mod internal_publication_tests;
crates/tui/src/plugins/discovery_internal_paths_tests.rs
+96
View File
@@ -0,0 +1,96 @@
use super::*;
fn bundle(root: &Path, directory: &str, name: &str) -> PathBuf {
let path = root.join(directory);
fs::create_dir_all(&path).unwrap();
fs::write(
path.join("plugin.toml"),
format!("schema_version = 1\n[plugin]\nname = {name:?}\nversion = \"1.0.0\"\n"),
)
.unwrap();
path
}
fn config(root: &Path) -> DiscoveryConfig {
DiscoveryConfig {
workspace: root.join("project"),
user_plugins_dir: root.join("plugins"),
workspace_plugins_dir: root.join("project/.codewhale/plugins"),
builtin_plugin_dirs: vec![],
state_path: root.join("state.json"),
}
}
#[test]
fn private_staging_copies_cannot_shadow_the_installed_incumbent() {
let tmp = tempfile::tempdir().unwrap();
let cfg = config(tmp.path());
let installed = bundle(&cfg.user_plugins_dir, "demo", "demo");
let operation = format!(".staging-{}", uuid::Uuid::new_v4().simple());
let staged = bundle(&cfg.user_plugins_dir, &operation, "demo");
fs::write(staged.join("canary.txt"), b"private staged payload").unwrap();
let registry = discover_with_config(&cfg);
assert_eq!(registry.len(), 1);
assert_eq!(
registry.get("demo").unwrap().canonical_root,
installed.canonicalize().unwrap()
);
assert!(
!registry
.diagnostics()
.iter()
.any(|d| d.code == "name-conflict")
);
assert_eq!(
fs::read(staged.join("canary.txt")).unwrap(),
b"private staged payload"
);
assert!(!cfg.state_path.exists(), "discovery remains read-only");
}
#[test]
fn reserved_backup_containers_are_hidden_but_human_bundle_paths_remain_visible() {
let tmp = tempfile::tempdir().unwrap();
let cfg = config(tmp.path());
fs::create_dir_all(&cfg.user_plugins_dir).unwrap();
let backup = tempfile::Builder::new()
.prefix(".plugin-backup-")
.rand_bytes(16)
.tempdir_in(&cfg.user_plugins_dir)
.unwrap();
// Even a manifest accidentally written directly in private recovery
// storage must not turn it into a discoverable public bundle.
bundle(backup.path(), "", "private-backup");
bundle(
&cfg.user_plugins_dir,
".staging-not-a-uuid",
"human-staging",
);
bundle(
&cfg.user_plugins_dir,
".plugin-backup-short",
"human-backup",
);
let dotted = cfg.user_plugins_dir.join("demo.bak");
fs::create_dir(&dotted).unwrap();
fs::write(
dotted.join("plugin.json"),
serde_json::json!({
"$schema": crate::plugins::agent_plugin::PLUGIN_SCHEMA_URL,
"name": "demo.bak", "version": "1.0.0"
})
.to_string(),
)
.unwrap();
let registry = discover_with_config(&cfg);
assert!(registry.get("private-backup").is_none());
assert!(registry.get("human-staging").is_some());
assert!(registry.get("human-backup").is_some());
assert_eq!(
registry.get("demo.bak").unwrap().canonical_root,
dotted.canonicalize().unwrap()
);
assert_eq!(registry.len(), 3);
assert!(backup.path().join("plugin.toml").is_file());
assert!(!cfg.state_path.exists());
}
crates/tui/src/plugins/install/mod.rs
+66 -17
View File
@@ -23,8 +23,8 @@
//! * The manifest `[plugin].name` must be a single path-safe segment; it
//! becomes the destination directory name.
//! * Overwriting a bundle that lacks the `.installed-from` marker is refused
//! — hand-placed bundles are never clobbered. `update` swaps atomically
//! only when the upstream bytes changed; a changed bundle automatically
//! — hand-placed bundles are never clobbered. `update` publishes a replacement
//! when source or installed content differs; a changed bundle automatically
//! invalidates the hash-bound trust receipt at the next discovery.
//! * Installed bits land **disabled and untrusted**; trust/enablement is the
//! existing registry flow, not this module's concern.
@@ -240,9 +240,9 @@ pub struct InstalledPlugin {
/// Result of an [`update`] call.
#[derive(Debug)]
pub enum PluginUpdateResult {
/// Upstream tarball is byte-identical to the recorded checksum; no action.
/// Source is unchanged; local bundles also match the actual installed tree.
NoChange,
/// Upstream changed and the on-disk bundle was atomically replaced.
/// Content changed and a replacement bundle was published.
Updated(InstalledPlugin),
/// Network policy requires approval for the download host.
NeedsApproval(String),
@@ -285,8 +285,9 @@ pub enum PluginInstallError {
/// `.installed-from` last.
///
/// `update = false` rejects an existing destination. `update = true` (only
/// called from [`update`]) requires the marker and replaces atomically with a
/// backup-restore on failure.
/// called from [`update`]) requires the marker and reserves a unique backup.
/// A failed placement restores only into an absent destination; errors after
/// publication retain the current path and prior backup for recovery.
///
/// `name_conflict` is consulted with the validated manifest name before the
/// rename; returning `Some(message)` aborts the install. It lets the caller
@@ -520,13 +521,12 @@ fn install_remote_bytes(
)
}
/// Re-fetch a previously installed plugin and atomically replace it if the
/// upstream tarball changed. The replaced bundle carries new content, so the
/// existing hash-bound trust receipt stops matching at the next discovery —
/// re-review is forced by the registry, not by this function.
/// Refresh a previously installed plugin from its recorded source and publish
/// a replacement if its content differs. A changed discovery-time hash requires
/// current review through the registry's existing hash-bound trust receipts.
///
/// Bundles installed from a local path cannot be re-downloaded; reinstall
/// them with `/plugin install <path>` instead.
/// Local bundles use the same bounded staging and backup/restore publication
/// as installation. A failed source validation leaves the installed copy intact.
pub async fn update(
name: &str,
user_plugins_dir: &Path,
@@ -547,6 +547,59 @@ pub async fn update(
let marker: InstalledFromMarker = serde_json::from_str(&marker_body)
.with_context(|| format!("malformed {INSTALLED_FROM_MARKER} for {name}"))?;
let source = PluginInstallSource::parse(&marker.spec)?;
if let PluginInstallSource::LocalPath(path) = &source {
let path = path.clone();
let user_plugins_dir = user_plugins_dir.to_path_buf();
let name = name.to_string();
#[cfg(test)]
let env_scope = crate::test_support::env_scope_ticket();
return tokio::task::spawn_blocking(move || {
#[cfg(test)]
let _env_scope = crate::test_support::join_env_scope(env_scope);
let staged = stage_local_copy(&path, &user_plugins_dir, max_size)?;
let staged_path = staged.staged_path.clone();
let result = (|| {
if staged.name != name {
return Err(PluginNameConflict(format!(
"updated plugin changed name from {name} to {}; original plugin preserved",
staged.name
))
.into());
}
// Compare the actual installed tree using the existing complete
// bundle hash, including executable intent and provenance. A historical
// source digest alone cannot detect an altered installed payload.
fs::write(staged_path.join(INSTALLED_FROM_MARKER), marker_body)?;
let (_, expected_hash) = stage::validate_staged(&staged_path)?;
let (_, installed_hash) = stage::validate_staged(&target).context(
"installed plugin failed validation; unchanged source was not accepted and the installed copy was preserved",
)?;
fs::remove_file(staged_path.join(INSTALLED_FROM_MARKER))?;
if expected_hash == installed_hash {
fs::remove_dir_all(&staged_path)
.context("failed to remove unchanged plugin staging copy")?;
return Ok(PluginUpdateResult::NoChange);
}
match finalize_install(staged, &marker.spec, None, "", &user_plugins_dir, true)? {
PluginInstallOutcome::Installed(installed) => {
Ok(PluginUpdateResult::Updated(installed))
}
PluginInstallOutcome::NeedsApproval(host) => {
Ok(PluginUpdateResult::NeedsApproval(host))
}
PluginInstallOutcome::NetworkDenied(host) => {
Ok(PluginUpdateResult::NetworkDenied(host))
}
}
})();
if result.is_err() {
let _ = fs::remove_dir_all(&staged_path);
}
result
})
.await
.context("local plugin update task failed")?;
}
if let PluginInstallSource::Dsh(package) = &source {
// Re-convert the recorded package. An identical converted bundle is
// no change; a different one replaces the installed copy, and its
@@ -589,11 +642,7 @@ pub async fn update(
};
}
let PluginInstallSource::Remote(remote) = source else {
bail!(
"plugin '{name}' was installed from a local path ({}) and cannot be updated from the network; \
reinstall it with /plugin install <path>",
marker.spec
);
unreachable!("local and converted sources returned above");
};
let (bytes, url) = match fetch_tarball(&remote, network, max_size).await? {
crates/tui/src/plugins/install/place.rs
+116 -40
View File
@@ -1,11 +1,16 @@
//! Place a staged bundle at its final path, and the containment guards.
//!
//! [`finalize_install`] is the only writer of the destination directory: it
//! renames the staged tree into place atomically, restores the backup on any
//! failure, and writes `.installed-from` **last** so a partial install never
//! leaves a stale marker. [`ensure_target_within_plugins_dir`] mirrors
//! discovery's fail-closed rule that a bundle must resolve to a direct child
//! of the plugins root.
//! [`finalize_install`] reserves a private unique backup before moving an
//! incumbent and writes `.installed-from` last. Failed publication restores
//! only into an unoccupied destination. After publication, an error retains
//! both the incumbent backup and the current path: without serialization we
//! cannot prove that path still names this operation's publication, so it is
//! never recursively deleted or overwritten as rollback.
//! [`ensure_target_within_plugins_dir`] mirrors discovery's fail-closed rule
//! that a bundle must resolve to a direct child of the plugins root.
//!
//! This is not a serialized or crash-atomic bundle swap. A retained recovery
//! path is not proof that the old bundle was restored or remains active.
use std::fs;
use std::path::{Path, PathBuf};
@@ -26,7 +31,7 @@ pub(super) fn finalize_install(
update: bool,
) -> Result<PluginInstallOutcome> {
let final_path = user_plugins_dir.join(&staged.name);
let mut backup_path: Option<PathBuf> = None;
let mut backup_container: Option<PathBuf> = None;
if final_path.exists() {
if !update {
let has_marker = final_path.join(INSTALLED_FROM_MARKER).exists();
@@ -40,34 +45,52 @@ pub(super) fn finalize_install(
let _ = fs::remove_dir_all(&staged.staged_path);
return Err(PluginInstallError::NotInstalledHere(staged.name).into());
}
let backup = user_plugins_dir.join(format!("{}.bak", staged.name));
if backup.exists() {
fs::remove_dir_all(&backup).ok();
let mut builder = tempfile::Builder::new();
builder.prefix(".plugin-backup-").rand_bytes(16);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
builder.permissions(fs::Permissions::from_mode(0o700));
}
fs::rename(&final_path, &backup).with_context(|| {
format!(
"failed to backup existing plugin at {}",
final_path.display()
)
})?;
if let Err(error) = fs::rename(&staged.staged_path, &final_path) {
fs::rename(&backup, &final_path).ok();
return Err(error).context("failed to install staged plugin");
let reserved = builder
.tempdir_in(user_plugins_dir)
.context("failed to reserve a private plugin backup container")?;
crate::plugins::registry::ensure_private_plugin_state_directory(reserved.path())
.map_err(anyhow::Error::msg)?;
// Persist before moving any user data: TempDir cleanup must never
// delete the incumbent during an error return or unwind.
let container = reserved.keep();
let backup = container.join("bundle");
if let Err(error) = crate::plugins::builtin::publish_snapshot(&final_path, &backup) {
if let Err(cleanup) = fs::remove_dir(&container) {
tracing::warn!(path = %container.display(), %cleanup, "empty plugin backup container could not be removed");
}
return Err(error).with_context(|| {
format!(
"failed to backup existing plugin at {}",
final_path.display()
)
});
}
backup_path = Some(backup);
} else if let Err(error) = fs::rename(&staged.staged_path, &final_path) {
let _ = fs::remove_dir_all(&staged.staged_path);
return Err(error).context("failed to install staged plugin");
backup_container = Some(container);
}
if let Err(error) = crate::plugins::builtin::publish_snapshot(&staged.staged_path, &final_path)
{
if let Some(container) = backup_container.as_deref() {
return Err(restore_backup(container, &final_path, error.into()));
}
return Err(error)
.context("failed to install staged plugin; existing destination preserved");
}
// Discovery fail-closed rule: the installed bundle must canonicalize to a
// direct child of the user plugins root.
if let Err(error) = ensure_target_within_plugins_dir(&final_path, user_plugins_dir) {
let _ = fs::remove_dir_all(&final_path);
if let Some(backup) = backup_path.take() {
let _ = fs::rename(&backup, &final_path);
}
return Err(error);
return Err(retained_publication_error(
error,
&final_path,
backup_container.as_deref(),
));
}
// Write the marker last so a partial install never leaves a stale
@@ -80,12 +103,13 @@ pub(super) fn finalize_install(
&staged.content_hash,
&staged.name,
) {
let _ = fs::remove_dir_all(&final_path);
if let Some(backup) = backup_path.take() {
let _ = fs::rename(&backup, &final_path);
}
return Err(error);
return Err(retained_publication_error(
error,
&final_path,
backup_container.as_deref(),
));
}
let installed_content_hash =
match crate::plugins::agent_plugin::resolve_manifest_path(&final_path)
.ok_or_else(|| anyhow::anyhow!("installed plugin has no supported manifest"))
@@ -96,15 +120,19 @@ pub(super) fn finalize_install(
}) {
Ok(hash) => hash,
Err(error) => {
let _ = fs::remove_dir_all(&final_path);
if let Some(backup) = backup_path.take() {
let _ = fs::rename(&backup, &final_path);
}
return Err(error).context("installed plugin failed post-copy validation");
return Err(retained_publication_error(
error.context("installed plugin failed post-copy validation"),
&final_path,
backup_container.as_deref(),
));
}
};
if let Some(backup) = backup_path {
fs::remove_dir_all(&backup).ok();
if let Some(container) = backup_container
&& let Err(error) = fs::remove_dir_all(&container)
{
// The new bundle is already installed and validated. Cleanup failure
// does not undo it; retain the exact old-copy path in the diagnostic.
tracing::warn!(path = %container.display(), %error, "plugin installed but prior backup cleanup failed; retained backup requires recovery cleanup");
}
Ok(PluginInstallOutcome::Installed(InstalledPlugin {
@@ -116,6 +144,54 @@ pub(super) fn finalize_install(
}))
}
/// A failed no-replace publication did not install our staged tree. Restore
/// with the same exclusive primitive, so an intervening creator is preserved.
pub(super) fn restore_backup(
container: &Path,
final_path: &Path,
error: anyhow::Error,
) -> anyhow::Error {
let backup = container.join("bundle");
match crate::plugins::builtin::publish_snapshot(&backup, final_path) {
Ok(()) => {
if let Err(cleanup) = fs::remove_dir(container) {
tracing::warn!(path = %container.display(), %cleanup, "restored plugin but empty recovery container could not be removed");
}
error.context(format!(
"failed to install staged plugin; previous plugin restored at {}",
final_path.display()
))
}
Err(restore) => error.context(format!(
"failed to install staged plugin and automatic restoration failed: {restore}; \
previous plugin retained at {}; current destination {} was not replaced; recovery required",
backup.display(),
final_path.display()
)),
}
}
fn retained_publication_error(
error: anyhow::Error,
final_path: &Path,
backup_container: Option<&Path>,
) -> anyhow::Error {
let previous = backup_container.map_or_else(
|| "no previous plugin backup exists".to_string(),
|container| {
format!(
"previous plugin retained at {}",
container.join("bundle").display()
)
},
);
error.context(format!(
"plugin publication failed validation; {previous}; current destination {} retained; \
automatic restoration refused because its publication identity is not pinned; recovery required",
final_path.display()
))
}
// ─────────────────────────────────────────────────────────────────────────────
// Path guards
// ─────────────────────────────────────────────────────────────────────────────
crates/tui/src/plugins/install/stage.rs
+3 -3
View File
@@ -28,9 +28,9 @@ pub(super) struct StagedPlugin {
pub(super) fn fresh_staging_dir(user_plugins_dir: &Path) -> Result<PathBuf> {
ensure_plugins_dir(user_plugins_dir)?;
// A crashed stage can leave residue that discovery will surface as an
// untrusted, disabled bundle; the next install attempt cleans it up by
// using a fresh uuid path and never reuses the stale one.
// Discovery excludes this reserved staging namespace. A crashed stage
// can remain for recovery; later attempts reserve a fresh UUID path and
// never treat the stale copy as an installed bundle or reuse its path.
let staged_path = user_plugins_dir.join(format!(".staging-{}", uuid::Uuid::new_v4().simple()));
fs::create_dir(&staged_path)
.with_context(|| format!("failed to create staging dir {}", staged_path.display()))?;
crates/tui/src/plugins/install/tests.rs
+106 -18
View File
@@ -3,6 +3,9 @@ use super::*;
// by the verbs, so it is not in `super`'s namespace.
use super::tarball::scan_tarball;
#[path = "tests/place_recovery.rs"]
mod place_recovery;
fn write_bundle(root: &Path, dir: &str, name: &str) -> PathBuf {
let bundle = root.join(dir);
fs::create_dir_all(&bundle).unwrap();
@@ -757,26 +760,9 @@ async fn install_enforces_the_name_conflict_hook() {
// ── update / uninstall ────────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn update_refuses_local_installs_and_missing_markers() {
async fn update_refuses_missing_markers() {
let tmp = tempfile::tempdir().unwrap();
let plugins = tmp.path().join("plugins");
let source = write_bundle(tmp.path(), "src/demo", "demo");
install(
PluginInstallSource::parse(source.to_str().unwrap()).unwrap(),
&plugins,
DEFAULT_MAX_SIZE_BYTES,
&allow_all(),
false,
&no_conflict(),
)
.await
.unwrap();
let err = update("demo", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
.unwrap_err();
assert!(format!("{err:#}").contains("local path"), "got: {err:#}");
write_bundle(&plugins, "hand", "hand");
let err = update("hand", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
@@ -790,6 +776,108 @@ async fn update_refuses_local_installs_and_missing_markers() {
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn local_update_reuses_staging_and_preserves_the_installed_bundle_on_failure() {
let tmp = tempfile::tempdir().unwrap();
let plugins = tmp.path().join("plugins");
let source = write_bundle(tmp.path(), "src/demo", "demo");
fs::write(source.join("data.txt"), b"v1").unwrap();
install(
PluginInstallSource::LocalPath(source.clone()),
&plugins,
DEFAULT_MAX_SIZE_BYTES,
&allow_all(),
false,
&no_conflict(),
)
.await
.unwrap();
let target = plugins.join("demo");
let original_marker = fs::read(target.join(INSTALLED_FROM_MARKER)).unwrap();
assert!(matches!(
update("demo", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
.unwrap(),
PluginUpdateResult::NoChange
));
assert_eq!(
fs::read(target.join(INSTALLED_FROM_MARKER)).unwrap(),
original_marker
);
for altered in [Some(b"altered".as_slice()), None] {
if let Some(bytes) = altered {
fs::write(target.join("data.txt"), bytes).unwrap();
} else {
fs::remove_file(target.join("data.txt")).unwrap();
}
assert!(matches!(
update("demo", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
.unwrap(),
PluginUpdateResult::Updated(_)
));
assert_eq!(fs::read(target.join("data.txt")).unwrap(), b"v1");
assert_eq!(
fs::read(target.join(INSTALLED_FROM_MARKER)).unwrap(),
original_marker
);
}
fs::write(source.join("data.txt"), b"v2").unwrap();
let PluginUpdateResult::Updated(updated) =
update("demo", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
.unwrap()
else {
panic!("changed local source must update through the existing installer");
};
assert_eq!(fs::read(updated.path.join("data.txt")).unwrap(), b"v2");
let updated_marker = fs::read(target.join(INSTALLED_FROM_MARKER)).unwrap();
assert_ne!(updated_marker, original_marker);
assert!(matches!(
update("demo", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
.unwrap(),
PluginUpdateResult::NoChange
));
let manifest = fs::read(source.join("plugin.toml")).unwrap();
fs::write(
source.join("plugin.toml"),
b"schema_version = 1\n[plugin]\nname = \"other\"\nversion = \"2.0.0\"\n",
)
.unwrap();
let error = update("demo", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
.unwrap_err();
assert!(error.downcast_ref::<PluginNameConflict>().is_some());
assert!(!plugins.join("other").exists());
fs::write(source.join("plugin.toml"), b"malformed = [").unwrap();
assert!(
update("demo", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
.is_err()
);
fs::write(source.join("plugin.toml"), manifest).unwrap();
fs::remove_dir_all(&source).unwrap();
assert!(
update("demo", &plugins, DEFAULT_MAX_SIZE_BYTES, &allow_all())
.await
.is_err()
);
assert_eq!(fs::read(target.join("data.txt")).unwrap(), b"v2");
assert_eq!(
fs::read(target.join(INSTALLED_FROM_MARKER)).unwrap(),
updated_marker
);
assert_eq!(
fs::read_dir(&plugins).unwrap().count(),
1,
"no staging or backup residue"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn uninstall_requires_the_marker_and_removes_the_bundle() {
let tmp = tempfile::tempdir().unwrap();
crates/tui/src/plugins/install/tests/place_recovery.rs
+160
View File
@@ -0,0 +1,160 @@
use super::*;
use crate::plugins::install::place::{finalize_install, restore_backup};
use crate::plugins::install::stage::stage_local_copy;
fn incumbent(plugins: &Path) -> PathBuf {
let target = write_bundle(plugins, "demo", "demo");
fs::write(target.join("data.txt"), b"old payload").unwrap();
fs::write(target.join(INSTALLED_FROM_MARKER), b"old provenance").unwrap();
target
}
fn prepared_update(root: &Path, plugins: &Path) -> super::super::stage::StagedPlugin {
let source = write_bundle(root, "source", "demo");
fs::write(source.join("data.txt"), b"new payload").unwrap();
stage_local_copy(&source, plugins, DEFAULT_MAX_SIZE_BYTES).unwrap()
}
#[test]
fn update_preserves_a_legitimate_dot_bak_plugin() {
let tmp = tempfile::tempdir().unwrap();
let plugins = tmp.path().join("plugins");
let target = incumbent(&plugins);
let other = plugins.join("demo.bak");
fs::create_dir(&other).unwrap();
let manifest = serde_json::json!({
"$schema": crate::plugins::agent_plugin::PLUGIN_SCHEMA_URL,
"name": "demo.bak", "version": "1.0.0"
});
fs::write(other.join("plugin.json"), manifest.to_string()).unwrap();
fs::write(other.join("canary.txt"), b"unrelated installed plugin").unwrap();
let before =
crate::plugins::manifest::PluginManifest::validate_from_path(&other.join("plugin.json"))
.unwrap()
.content_hash;
let staged = prepared_update(tmp.path(), &plugins);
let outcome =
finalize_install(staged, "path:/controlled/source", None, "", &plugins, true).unwrap();
assert!(matches!(outcome, PluginInstallOutcome::Installed(_)));
assert_eq!(fs::read(target.join("data.txt")).unwrap(), b"new payload");
assert_eq!(
fs::read(other.join("canary.txt")).unwrap(),
b"unrelated installed plugin"
);
let after =
crate::plugins::manifest::PluginManifest::validate_from_path(&other.join("plugin.json"))
.unwrap()
.content_hash;
assert_eq!(
before, after,
"unrelated valid bundle must be byte-identical"
);
assert_eq!(
fs::read_dir(&plugins).unwrap().count(),
2,
"successful publication leaves no backup container"
);
}
#[test]
fn failed_publication_restores_the_original_bytes_and_marker() {
let tmp = tempfile::tempdir().unwrap();
let plugins = tmp.path().join("plugins");
let target = incumbent(&plugins);
let staged = prepared_update(tmp.path(), &plugins);
fs::remove_dir_all(&staged.staged_path).unwrap();
let error =
finalize_install(staged, "path:/controlled/source", None, "", &plugins, true).unwrap_err();
assert!(format!("{error:#}").contains("previous plugin restored"));
assert_eq!(fs::read(target.join("data.txt")).unwrap(), b"old payload");
assert_eq!(
fs::read(target.join(INSTALLED_FROM_MARKER)).unwrap(),
b"old provenance"
);
assert_eq!(fs::read_dir(&plugins).unwrap().count(), 1);
}
#[test]
fn failed_restore_retains_the_old_copy_and_never_replaces_an_empty_contender() {
let tmp = tempfile::tempdir().unwrap();
let container = tempfile::Builder::new()
.prefix(".plugin-backup-")
.rand_bytes(16)
.tempdir_in(tmp.path())
.unwrap()
.keep();
let old = write_bundle(&container, "bundle", "demo");
fs::write(old.join("data.txt"), b"recoverable old payload").unwrap();
fs::write(old.join(INSTALLED_FROM_MARKER), b"recoverable old marker").unwrap();
let contender = tmp.path().join("demo");
fs::create_dir(&contender).unwrap();
let error = restore_backup(
&container,
&contender,
anyhow::anyhow!("publication refused"),
);
let message = format!("{error:#}");
assert!(message.contains("automatic restoration failed"));
assert!(message.contains(&old.display().to_string()));
assert!(message.contains("current destination") && message.contains("was not replaced"));
assert_eq!(
fs::read(old.join("data.txt")).unwrap(),
b"recoverable old payload"
);
assert_eq!(
fs::read(old.join(INSTALLED_FROM_MARKER)).unwrap(),
b"recoverable old marker"
);
assert!(contender.is_dir());
assert_eq!(
fs::read_dir(&contender).unwrap().count(),
0,
"atomic no-replace must preserve even an empty contender directory"
);
}
#[test]
fn marker_failure_retains_prior_and_current_publications_for_recovery() {
let tmp = tempfile::tempdir().unwrap();
let plugins = tmp.path().join("plugins");
let target = incumbent(&plugins);
let staged = prepared_update(tmp.path(), &plugins);
// A deterministic real filesystem refusal after rename: a directory
// cannot be overwritten by the provenance marker's fs::write.
fs::create_dir(staged.staged_path.join(INSTALLED_FROM_MARKER)).unwrap();
let error =
finalize_install(staged, "path:/controlled/source", None, "", &plugins, true).unwrap_err();
let message = format!("{error:#}");
assert!(message.contains("automatic restoration refused"));
assert!(message.contains("publication identity is not pinned"));
assert!(!message.contains("previous plugin restored"));
assert_eq!(fs::read(target.join("data.txt")).unwrap(), b"new payload");
assert!(target.join(INSTALLED_FROM_MARKER).is_dir());
let containers = fs::read_dir(&plugins)
.unwrap()
.map(|entry| entry.unwrap().path())
.filter(|path| {
path.file_name()
.unwrap()
.to_string_lossy()
.starts_with(".plugin-backup-")
})
.collect::<Vec<_>>();
assert_eq!(containers.len(), 1);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
assert_eq!(
fs::metadata(&containers[0]).unwrap().permissions().mode() & 0o077,
0,
"recovery container must be owner-only at creation"
);
}
let retained = containers[0].join("bundle");
assert!(message.contains(&retained.display().to_string()));
assert_eq!(fs::read(retained.join("data.txt")).unwrap(), b"old payload");
assert_eq!(
fs::read(retained.join(INSTALLED_FROM_MARKER)).unwrap(),
b"old provenance"
);
}
crates/tui/src/plugins/mutation.rs
+1 -1
View File
@@ -30,7 +30,7 @@ pub enum PluginMutationRequest {
source: PluginInstallSource,
expected_content_hash: String,
},
/// Re-download a previously installed bundle by name or id. A changed
/// Refresh a previously installed bundle by name or id. A changed
/// bundle automatically invalidates its trust receipt at next discovery.
Update { selector: String },
/// Delete an installed bundle and prune its persisted state entry.
crates/tui/src/runtime_api.rs
+136 -13
View File
@@ -1588,7 +1588,15 @@ fn open_runtime_threads_for_server(
SharedRuntimeThreadManager,
crate::tools::large_output_router::WorkshopConfig,
)> {
open_runtime_threads_for_host(config, workspace, manager_config, plugin_registry, false)
open_runtime_threads_for_host(
config,
workspace,
manager_config,
plugin_registry,
false,
None,
None,
)
}
pub(crate) fn open_runtime_threads_for_host(
@@ -1597,6 +1605,8 @@ pub(crate) fn open_runtime_threads_for_host(
manager_config: RuntimeThreadManagerConfig,
plugin_registry: Arc<crate::plugins::PluginRegistry>,
acp: bool,
preheld_lock: Option<crate::runtime_threads::RuntimeProcessOwnerLock>,
binding: Option<&crate::runtime_threads::RuntimeStoreBinding>,
) -> Result<(
SharedRuntimeThreadManager,
crate::tools::large_output_router::WorkshopConfig,
@@ -1606,16 +1616,20 @@ pub(crate) fn open_runtime_threads_for_host(
// thread manager can spawn any of those engines, matching interactive and
// headless exec startup.
let workshop_activation = install_runtime_server_workshop_budgets(config);
let manager = Arc::new(if acp {
RuntimeThreadManager::open_acp(config.clone(), workspace, manager_config, plugin_registry)
let profile = if acp {
crate::core::engine::EngineHostProfile::Acp
} else {
RuntimeThreadManager::open_with_plugin_registry(
config.clone(),
workspace,
manager_config,
plugin_registry,
)
}?);
crate::core::engine::EngineHostProfile::Normal
};
let manager = Arc::new(RuntimeThreadManager::open_host(
config.clone(),
workspace,
manager_config,
plugin_registry,
profile,
preheld_lock,
binding,
)?);
// Publish the same exact endpoint-scoped catalog as interactive startup
// before the server admits turns. A cached model list alone does not make
// its capabilities available to route resolution.
@@ -1623,6 +1637,79 @@ pub(crate) fn open_runtime_threads_for_host(
Ok((manager, workshop_activation))
}
#[cfg(unix)]
enum RuntimeHostAdmission {
Attached(Box<codewhale_app_server::daemon_client::OwnerClient>),
Bootstrap {
lock: crate::runtime_threads::RuntimeProcessOwnerLock,
binding: Option<crate::runtime_threads::RuntimeStoreBinding>,
recovery: Option<Arc<codewhale_app_server::daemon_client::UnavailablePublication>>,
},
}
/// Observe first, then hold the existing store lease through all recovery.
/// Only pre-write publication changes or lease contention can re-observe.
#[cfg(unix)]
async fn admit_runtime_host(
config_path: Option<PathBuf>,
selected_socket: Option<PathBuf>,
selected_store: PathBuf,
) -> Result<RuntimeHostAdmission> {
use codewhale_app_server::daemon_client::{HostOwnerProbe, PublicationChangedBeforeAttach};
use codewhale_app_server::daemon_socket::owner_work;
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let probe = match codewhale_app_server::daemon_client::probe_owner_for_host_startup(
config_path.clone(), selected_socket.clone(),
).await {
Err(error) if error.is::<PublicationChangedBeforeAttach>() => {
tokio::time::sleep(Duration::from_millis(25)).await;
continue;
}
result => result?,
};
let (binding, recovery) = match probe {
HostOwnerProbe::Attached(client) => return Ok(RuntimeHostAdmission::Attached(client)),
HostOwnerProbe::Absent => (None, None),
HostOwnerProbe::Unavailable(publication) => {
let binding = validate_selected_owner_receipt(publication.receipt().clone(), selected_store.clone()).await?;
(Some(binding), Some(publication))
}
};
let store = selected_store.clone();
let existing = binding.clone();
let create = recovery.is_none();
let lock = owner_work(move || {
if let Some(binding) = existing.as_ref() { binding.validate_existing_store()?; }
crate::runtime_threads::RuntimeProcessOwnerLock::try_acquire_for_host(&store, create)
}).await?;
let Some(lock) = lock else {
tokio::time::sleep(Duration::from_millis(25)).await;
continue;
};
let store = selected_store.clone();
let existing = binding.clone();
let lock = owner_work(move || {
lock.validate_at_root(&store)?;
if let Some(binding) = existing.as_ref() { binding.validate_existing_store()?; }
Ok(lock)
}).await?;
let current = match recovery.as_ref() {
Some(publication) => publication.revalidate().await?,
None => codewhale_app_server::daemon_client::publication_is_absent(
config_path.clone(), selected_socket.clone(),
).await?,
};
if !current {
drop(lock);
tokio::time::sleep(Duration::from_millis(25)).await;
continue;
}
return Ok(RuntimeHostAdmission::Bootstrap { lock, binding, recovery });
}
}).await.context("Runtime host admission deadline expired; attachment outcome may be uncertain and is not replayed")?
}
/// Prefix of the first line the Runtime prints once it holds its listener.
pub const RUNTIME_LISTENING_PREFIX: &str = "Runtime API listening on http://";
@@ -1646,12 +1733,34 @@ pub async fn run_http_server(
Some(task_default_model.clone()),
Some(options.workers),
);
#[cfg(any(unix, windows))]
#[cfg(unix)]
let admission = admit_runtime_host(
options.config_path.clone(),
selected_control_socket(&options),
RuntimeThreadManagerConfig::from_task_data_dir(task_cfg.data_dir.clone()).data_dir,
)
.await?;
#[cfg(unix)]
let (published, preheld_lock, existing_binding, recovery) = match admission {
RuntimeHostAdmission::Attached(client) => (Some(*client), None, None, None),
RuntimeHostAdmission::Bootstrap {
lock,
binding,
recovery,
} => (None, Some(lock), binding, recovery),
};
#[cfg(windows)]
let published = codewhale_app_server::daemon_client::connect_if_published(
options.config_path.clone(),
selected_control_socket(&options),
)
.await?;
#[cfg(not(unix))]
let preheld_lock = None;
#[cfg(not(unix))]
let existing_binding: Option<crate::runtime_threads::RuntimeStoreBinding> = None;
#[cfg(windows)]
let recovery = None;
#[cfg(any(unix, windows))]
if let Some(control) = published {
let selected_store =
@@ -1761,6 +1870,8 @@ pub async fn run_http_server(
manager_config,
plugin_discovery.registry_for_workspace(&workspace),
acp_selected,
preheld_lock,
existing_binding.as_ref(),
)?;
let sessions_dir = runtime_threads.sessions_dir().to_path_buf();
let task_manager =
@@ -1865,6 +1976,7 @@ pub async fn run_http_server(
},
options.workers,
resolved_auth.generated,
recovery,
)
.await?;
let listener_workspace = state.workspace.clone();
@@ -2003,7 +2115,16 @@ pub(crate) async fn validate_selected_owner(
client: &codewhale_app_server::daemon_client::OwnerClient,
selected_store: PathBuf,
) -> Result<()> {
let receipt = client.receipt().clone();
validate_selected_owner_receipt(client.receipt().clone(), selected_store)
.await
.map(|_| ())
}
#[cfg(any(unix, windows))]
async fn validate_selected_owner_receipt(
receipt: codewhale_protocol::RuntimeOwnerReceipt,
selected_store: PathBuf,
) -> Result<crate::runtime_threads::RuntimeStoreBinding> {
codewhale_app_server::daemon_socket::owner_work(move || {
let selected = crate::runtime_threads::RuntimeStoreBinding::for_store_dir(&selected_store)?;
selected.validate_existing_store()?;
@@ -2013,7 +2134,7 @@ pub(crate) async fn validate_selected_owner(
&& selected.execution_scope == receipt.execution_scope,
"authenticated Runtime owner belongs to another selected store; refusing attachment"
);
Ok(())
Ok(selected)
})
.await
}
@@ -2025,6 +2146,7 @@ async fn bind_captured_runtime_frontends(
model: String,
worker_setting: usize,
generated_auth: bool,
recovery: Option<Arc<codewhale_app_server::daemon_client::UnavailablePublication>>,
) -> Result<(
codewhale_app_server::daemon_socket::DaemonSocket,
codewhale_app_server::AppState,
@@ -2079,6 +2201,7 @@ async fn bind_captured_runtime_frontends(
CapturedRuntimeFrontend::capture(state.clone(), model, worker_setting, generated_auth)
.await?,
),
recovery,
)
.await
}
crates/tui/src/runtime_api/plugins.rs
+2 -2
View File
@@ -459,8 +459,8 @@ async fn run_plugin_mutation(
to review its capabilities, then trust and enable it.",
),
PluginMutationOutcome::Updated => Some(
"Content changed; the previous trust receipt no longer matches. \
Review and trust it again before enabling.",
"Installed content refreshed. Review its current capabilities \
and trust it if needed before enabling.",
),
_ => None,
};
crates/tui/src/runtime_api/tests.rs
+3 -1
View File
@@ -16,6 +16,8 @@ use uuid::Uuid;
mod command_catalog;
mod headless_catalog;
#[cfg(unix)]
mod owner_bootstrap;
mod profile_constitution;
#[cfg(any(unix, windows))]
mod runtime_store_convergence;
@@ -1355,7 +1357,7 @@ async fn build_test_server(
let model = runtime_request_model(&state.config.read(), None)
.map_err(|error| anyhow::anyhow!(error.message))?;
let (owner, compatibility) =
bind_captured_runtime_frontends(&state, Some(socket), model, 1, false).await?;
bind_captured_runtime_frontends(&state, Some(socket), model, 1, false, None).await?;
(Some(owner), Some(compatibility))
} else {
(None, None)
crates/tui/src/runtime_api/tests/owner_bootstrap.rs
+380
View File
@@ -0,0 +1,380 @@
//! Actual manager/store/lease bootstrap, without any provider turn. Transport
//! fixtures do not substitute for the separate rebuilt two-process proof.
use super::*;
use crate::runtime_threads::{RuntimeProcessOwnerLock, RuntimeStoreBinding, ThreadListFilter};
use codewhale_config::private_directory::PrivateDirectory;
use codewhale_protocol::RuntimeOwnerReceipt;
struct OwnedChild(std::process::Child);
impl Drop for OwnedChild {
fn drop(&mut self) {
if self.0.try_wait().ok().flatten().is_none() {
let _ = self.0.kill();
}
let _ = self.0.wait();
}
}
struct Rig {
_temporary: tempfile::TempDir,
root: PathBuf,
store: PathBuf,
socket: PathBuf,
config_path: PathBuf,
config: Config,
}
impl Rig {
fn new() -> Result<Self> {
let temporary = tempfile::Builder::new()
.prefix("cw-ob-")
.tempdir_in("/tmp")?;
let root = temporary.path().canonicalize()?;
let config_path = root.join("config.toml");
fs::write(&config_path, "")?;
fs::create_dir(root.join("workspace"))?;
let config = Config::default().with_legacy_root(
Some("bootstrap-no-provider-fixture".into()),
Some("http://127.0.0.1:1/v1".into()),
);
Ok(Self {
store: root.join("runtime"),
socket: root.join("run/daemon.sock"),
_temporary: temporary,
root,
config_path,
config,
})
}
fn manager(
&self,
lock: Option<RuntimeProcessOwnerLock>,
binding: Option<&RuntimeStoreBinding>,
) -> Result<SharedRuntimeThreadManager> {
let (manager, _) = open_runtime_threads_for_host(
&self.config,
self.root.join("workspace"),
RuntimeThreadManagerConfig {
data_dir: self.store.clone(),
task_data_dir: self.root.join("tasks"),
sessions_dir: Some(self.root.join("sessions")),
max_active_threads: 8,
},
Arc::new(crate::plugins::PluginRegistry::new()),
false,
lock,
binding,
)?;
Ok(manager)
}
async fn receipt(&self, manager: &SharedRuntimeThreadManager) -> Result<RuntimeOwnerReceipt> {
let (binding, generation) = manager.capture_control_owner()?;
Ok(RuntimeOwnerReceipt {
version: 1,
data_dir: binding.data_dir,
execution_scope: binding.execution_scope,
lease_generation: generation,
pid: std::process::id(),
process_start: codewhale_app_server::daemon_socket::capture_process_start(
std::process::id(),
)
.await?,
principal: PrivateDirectory::current_user_id().to_string(),
socket_path: self.socket.clone(),
config_path: Some(self.config_path.clone()),
})
}
fn publish(&self, owner: &RuntimeOwnerReceipt) -> Result<()> {
let parent = PrivateDirectory::admit(self.socket.parent().unwrap())?;
let listener = std::os::unix::net::UnixListener::bind(&self.socket)?;
let identity = parent
.socket_identity("daemon.sock")?
.context("owned socket")?;
parent.protect_socket("daemon.sock", identity)?;
drop(listener);
parent.write_owned_file("daemon.sock.owner.json", &serde_json::to_vec(owner)?, false)?;
Ok(())
}
async fn dead_publication(&self) -> Result<(RuntimeOwnerReceipt, String, Value)> {
let manager = self.manager(None, None)?;
let draft = manager
.create_thread(CreateThreadRequest {
allow_shell: Some(false),
system_prompt: Some("Preserve this unrelated saved draft".into()),
..CreateThreadRequest::default()
})
.await?;
let detail = serde_json::to_value(manager.get_thread_detail(&draft.id).await?)?;
let mut owner = self.receipt(&manager).await?;
manager.shutdown_and_wait().await?;
drop(manager);
let mut child = OwnedChild(
std::process::Command::new("/bin/sleep")
.arg("60")
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()?,
);
owner.pid = child.0.id();
owner.process_start =
codewhale_app_server::daemon_socket::capture_process_start(owner.pid).await?;
child.0.kill()?;
child.0.wait()?;
self.publish(&owner)?;
Ok((owner, draft.id, detail))
}
async fn admit(&self) -> Result<RuntimeHostAdmission> {
admit_runtime_host(
Some(self.config_path.clone()),
Some(self.socket.clone()),
self.store.clone(),
)
.await
}
}
#[tokio::test]
async fn preheld_bootstrap_lease_stays_with_real_manager_until_drop() -> Result<()> {
let _env = lock_test_env();
let rig = Rig::new()?;
let RuntimeHostAdmission::Bootstrap {
lock,
binding,
recovery,
} = rig.admit().await?
else {
bail!("fresh host must bootstrap")
};
assert!(binding.is_none() && recovery.is_none());
assert!(RuntimeProcessOwnerLock::try_acquire_for_host(&rig.store, false)?.is_none());
let manager = rig.manager(Some(lock), binding.as_ref())?;
assert!(RuntimeProcessOwnerLock::try_acquire_for_host(&rig.store, false)?.is_none());
let (owner, generation) = manager.capture_control_owner()?;
assert_eq!(owner.data_dir, rig.store);
assert!(!generation.is_empty());
manager.shutdown_and_wait().await?;
assert!(
RuntimeProcessOwnerLock::try_acquire_for_host(&rig.store, false)?.is_none(),
"drain alone must not release retained ownership"
);
drop(manager);
assert!(RuntimeProcessOwnerLock::try_acquire_for_host(&rig.store, false)?.is_some());
Ok(())
}
#[tokio::test]
async fn two_crash_reentries_converge_on_one_owner_and_preserve_saved_draft() -> Result<()> {
let _env = lock_test_env();
let rig = Rig::new()?;
let (old, draft, detail) = rig.dead_publication().await?;
let gate = Arc::new(tokio::sync::Barrier::new(3));
let start = || {
let gate = gate.clone();
let config = rig.config_path.clone();
let socket = rig.socket.clone();
let store = rig.store.clone();
tokio::spawn(async move {
gate.wait().await;
admit_runtime_host(Some(config), Some(socket), store).await
})
};
let mut first = start();
let mut second = start();
gate.wait().await;
let (admission, loser) = tokio::select! {
result = &mut first => (result??, second),
result = &mut second => (result??, first),
};
let RuntimeHostAdmission::Bootstrap {
lock,
binding,
recovery,
} = admission
else {
bail!("one crash reentry must win the existing lease")
};
let recovery = recovery.context("exact stale publication")?;
assert_eq!(recovery.receipt(), &old);
assert!(RuntimeProcessOwnerLock::try_acquire_for_host(&rig.store, false)?.is_none());
let manager = rig.manager(Some(lock), binding.as_ref())?;
let owner = rig.receipt(&manager).await?;
assert_ne!(owner.lease_generation, old.lease_generation);
assert_eq!(owner.execution_scope, old.execution_scope);
let (daemon, _) = codewhale_app_server::bind_runtime_frontends(
Some(rig.config_path.clone()),
None,
owner.clone(),
codewhale_app_server::RuntimeOwnerRouting {
endpoint: "127.0.0.1:1".parse()?,
workspace: None,
workers: None,
mobile: false,
web: false,
acp: false,
acp_only: false,
},
None,
Some(recovery),
)
.await?;
let stop = daemon.shutdown_handle();
let serving = tokio::spawn(daemon.serve());
let RuntimeHostAdmission::Attached(client) = loser.await?? else {
bail!("loser must attach to the winner, never open another manager")
};
assert_eq!(client.receipt(), &owner);
validate_selected_owner(&client, rig.store.clone()).await?;
assert_eq!(
serde_json::to_value(manager.get_thread_detail(&draft).await?)?,
detail
);
assert_eq!(
manager
.list_threads(ThreadListFilter::IncludeArchived, None)
.await?
.len(),
1
);
drop(client);
stop.trigger();
serving.await??;
manager.shutdown_and_wait().await?;
drop(manager);
assert!(RuntimeProcessOwnerLock::try_acquire_for_host(&rig.store, false)?.is_some());
Ok(())
}
#[tokio::test]
async fn dead_publication_with_missing_existing_lease_refuses_without_recreation() -> Result<()> {
let _env = lock_test_env();
let rig = Rig::new()?;
let (owner, _, _) = rig.dead_publication().await?;
let lease = rig.store.join("runtime-process.owner.lock");
fs::remove_file(&lease)?;
assert!(rig.admit().await.is_err());
assert!(
!lease.exists(),
"dead publication cannot manufacture a replacement lease"
);
assert_eq!(
serde_json::from_slice::<RuntimeOwnerReceipt>(&fs::read(
rig.socket.with_file_name("daemon.sock.owner.json")
)?)?,
owner
);
Ok(())
}
#[tokio::test]
async fn owner_bootstrap_attach_response_loss_never_falls_back_or_changes_store() -> Result<()> {
use tokio::io::AsyncBufReadExt as _;
let _env = lock_test_env();
let rig = Rig::new()?;
let manager = rig.manager(None, None)?;
let owner = rig.receipt(&manager).await?;
manager.shutdown_and_wait().await?;
drop(manager);
let store_owner = fs::read(rig.store.join("owner.json"))?;
let parent = PrivateDirectory::admit(rig.socket.parent().unwrap())?;
let listener = tokio::net::UnixListener::bind(&rig.socket)?;
let identity = parent.socket_identity("daemon.sock")?.context("socket")?;
parent.protect_socket("daemon.sock", identity)?;
let bytes = serde_json::to_vec(&owner)?;
parent.write_owned_file("daemon.sock.owner.json", &bytes, false)?;
let peer = tokio::spawn(async move {
let (stream, _) = listener.accept().await?;
let mut line = String::new();
tokio::io::BufReader::new(stream)
.read_line(&mut line)
.await?;
let request: Value = serde_json::from_str(&line)?;
anyhow::ensure!(
request["method"] == "daemon/attach",
"one real attach write"
);
// Simulate lost acknowledgement after the valid peer saw attach.
Ok::<_, anyhow::Error>(())
});
let error = rig.admit().await.err().context("lost attach must refuse")?;
assert!(error.to_string().contains("closed before attach"));
peer.await??;
assert_eq!(fs::read(rig.store.join("owner.json"))?, store_owner);
assert_eq!(
parent
.read_private_receipt("daemon.sock.owner.json", 16384)?
.context("original receipt")?
.0,
bytes
);
assert_eq!(parent.socket_identity("daemon.sock")?, Some(identity));
assert!(
RuntimeProcessOwnerLock::try_acquire_for_host(&rig.store, false)?.is_some(),
"no owner manager was opened after attach uncertainty"
);
Ok(())
}
#[tokio::test]
async fn stale_publication_cannot_select_another_store_or_execution_scope() -> Result<()> {
let _env = lock_test_env();
for wrong_store in [false, true] {
let rig = Rig::new()?;
let (mut owner, _, _) = rig.dead_publication().await?;
if wrong_store {
owner.data_dir = rig.root.join("other-store");
} else {
owner.execution_scope.push_str("-wrong");
}
fs::write(
rig.socket.with_file_name("daemon.sock.owner.json"),
serde_json::to_vec(&owner)?,
)?;
assert!(rig.admit().await.is_err());
assert!(!rig.root.join("other-store").exists());
assert_eq!(
serde_json::from_slice::<RuntimeOwnerReceipt>(&fs::read(
rig.socket.with_file_name("daemon.sock.owner.json")
)?)?,
owner
);
}
Ok(())
}
#[tokio::test]
async fn moved_preheld_lease_refuses_before_store_recovery() -> Result<()> {
let _env = lock_test_env();
let rig = Rig::new()?;
let RuntimeHostAdmission::Bootstrap { lock, .. } = rig.admit().await? else {
bail!("fresh bootstrap")
};
let lease = rig.store.join("runtime-process.owner.lock");
fs::rename(&lease, rig.store.join("moved.owner.lock"))?;
fs::write(&lease, b"")?;
assert!(rig.manager(Some(lock), None).is_err());
assert!(
!rig.store.join("owner.json").exists(),
"invalid held lease must not recover/mint the store"
);
Ok(())
}
#[tokio::test]
async fn busy_unpublished_owner_has_bounded_refusal_without_store_recovery() -> Result<()> {
let _env = lock_test_env();
let rig = Rig::new()?;
let held = RuntimeProcessOwnerLock::try_acquire_for_host(&rig.store, true)?
.context("first owner lease")?;
let started = tokio::time::Instant::now();
let error = rig
.admit()
.await
.err()
.context("busy host must remain bounded")?;
assert!(error.to_string().contains("deadline expired"));
assert!(started.elapsed() < Duration::from_secs(8));
assert!(!rig.store.join("owner.json").exists());
assert!(!rig.socket.with_file_name("daemon.sock.owner.json").exists());
drop(held);
Ok(())
}
crates/tui/src/runtime_threads.rs
+86 -38
View File
@@ -6635,6 +6635,23 @@ impl RuntimeProcessOwnerLock {
Ok(None)
}
/// Host admission reuses this exact store lease, never a PID-file lock.
pub(crate) fn try_acquire_for_host(root: &Path, create: bool) -> Result<Option<Self>> {
if !create {
checked_existing_runtime_store_dir(root)?;
}
Self::try_acquire_file(&root.join(RUNTIME_PROCESS_OWNER_LOCK_FILE), create)
}
pub(crate) fn validate_at_root(&self, root: &Path) -> Result<()> {
let root = checked_runtime_store_root(root.to_path_buf())?;
anyhow::ensure!(
owner_lock_is_at_path(&self._file, &root.join(RUNTIME_PROCESS_OWNER_LOCK_FILE))?,
"held Runtime owner lease changed before host recovery"
);
Ok(())
}
pub(crate) fn acquire(root: &Path) -> Result<Self> {
let root = checked_runtime_store_root(root.to_path_buf())?;
let path = root.join(RUNTIME_PROCESS_OWNER_LOCK_FILE);
@@ -7408,6 +7425,7 @@ impl RuntimeThreadManager {
None,
None,
crate::core::engine::EngineHostProfile::Normal,
None,
)
}
@@ -7424,6 +7442,7 @@ impl RuntimeThreadManager {
Some(plugin_registry),
None,
crate::core::engine::EngineHostProfile::Normal,
None,
)
}
@@ -7457,6 +7476,7 @@ impl RuntimeThreadManager {
Some(plugin_registry),
None,
crate::core::engine::EngineHostProfile::Normal,
None,
);
}
binding.validate_existing_store()?;
@@ -7469,6 +7489,7 @@ impl RuntimeThreadManager {
Some(plugin_registry),
binding,
crate::core::engine::EngineHostProfile::Normal,
None,
)
}
@@ -7496,6 +7517,7 @@ impl RuntimeThreadManager {
Some(plugin_registry),
Some(binding),
crate::core::engine::EngineHostProfile::Normal,
None,
)
}
@@ -7521,19 +7543,23 @@ impl RuntimeThreadManager {
Self::open_existing_session(config, workspace, manager_cfg, plugin_registry, &binding)
}
pub(crate) fn open_acp(
pub(crate) fn open_host(
config: Config,
workspace: PathBuf,
manager_cfg: RuntimeThreadManagerConfig,
plugins: Arc<crate::plugins::PluginRegistry>,
host_profile: crate::core::engine::EngineHostProfile,
preheld_lock: Option<RuntimeProcessOwnerLock>,
binding: Option<&RuntimeStoreBinding>,
) -> Result<Self> {
Self::open_inner(
config,
workspace,
manager_cfg,
Some(plugins),
None,
crate::core::engine::EngineHostProfile::Acp,
binding,
host_profile,
preheld_lock,
)
}
@@ -7544,6 +7570,7 @@ impl RuntimeThreadManager {
plugin_registry: Option<Arc<crate::plugins::PluginRegistry>>,
binding: Option<&RuntimeStoreBinding>,
host_profile: crate::core::engine::EngineHostProfile,
preheld_lock: Option<RuntimeProcessOwnerLock>,
) -> Result<Self> {
// A public RuntimeThreadManager owns independent native threads. They
// may run concurrently with the interactive TUI because their events
@@ -7560,9 +7587,18 @@ impl RuntimeThreadManager {
.sessions_dir()
.to_path_buf(),
);
let process_owner_lock = Arc::new(RuntimeProcessOwnerLock::acquire(&manager_cfg.data_dir)?);
let process_owner_lock = match preheld_lock {
Some(lock) => lock,
None => RuntimeProcessOwnerLock::acquire(&manager_cfg.data_dir)?,
};
process_owner_lock.validate_at_root(&manager_cfg.data_dir)?;
let process_owner_lock = Arc::new(process_owner_lock);
// Recheck under the exclusive host lock, before store recovery can write.
if let Some(binding) = binding {
anyhow::ensure!(
checked_runtime_store_root(manager_cfg.data_dir.clone())? == binding.data_dir,
"captured Runtime store differs from the held host lease"
);
binding.validate_existing_store()?;
}
let store = RuntimeThreadStore::open(manager_cfg.data_dir.clone())?;
@@ -17308,6 +17344,7 @@ impl RuntimeThreadManager {
tool_name,
description,
input,
approval_key,
approval_grouping_key,
intent_summary,
approval_force_prompt,
@@ -17350,7 +17387,30 @@ impl RuntimeThreadManager {
summary: Some(summary.clone()),
};
if auto_approve {
use crate::core::authority::{
ApprovalRequestDisposition, TurnAuthority,
resolve_approval_request_disposition,
};
let grant = self.session_grant_for(&thread_id, &approval_grouping_key);
// This resolver reads the exact live approval posture only;
// Core already owns mode and shell admission for this call.
let approval_authority = TurnAuthority::from_effective_fields(
codewhale_config::AppMode::Agent,
true,
trust_mode,
auto_approve,
approval_mode,
);
let disposition = resolve_approval_request_disposition(
&approval_authority,
grant.is_some(),
false, // Runtime has grants, but no session-denial cache.
approval_force_prompt,
// Rust mints this namespace; tool text cannot claim origin.
approval_key.starts_with("extcall:ext:"),
);
if disposition == ApprovalRequestDisposition::AutoApprove && auto_approve {
// No waiter is registered on this path, but the emitted
// identity still has to obey the one contract clients
// read: `approval_id` is ours, `tool_call_id` is the
@@ -17373,13 +17433,6 @@ impl RuntimeThreadManager {
}),
)
.await?;
let auto_decision =
Self::approval_decision(auto_approve, trust_mode, false);
let (dec_str, approved) = match auto_decision {
RuntimeApprovalDecision::ApproveTool => ("allow", true),
RuntimeApprovalDecision::DenyTool
| RuntimeApprovalDecision::RetryWithFullAccess => ("deny", false),
};
// Emit approval.decided so external clients (GUI)
// know the approval was resolved automatically and
// can clear any pending approval UI. Without this
@@ -17393,37 +17446,33 @@ impl RuntimeThreadManager {
json!({
"approval_id": approval_id,
"tool_call_id": id,
"decision": dec_str,
"decision": "allow",
"remember": false,
"auto": true,
}),
)
.await
.ok();
if approved {
let _ = engine
.approve_tool_call_by(
id,
crate::approval_log::ApprovalDecider::Posture,
)
.await;
} else {
let _ = engine
.deny_tool_call_by(
id,
crate::approval_log::ApprovalDecider::Posture,
)
.await;
}
let _ = engine
.approve_tool_call_by(id, crate::approval_log::ApprovalDecider::Posture)
.await;
continue;
}
// Auto-Review never opens an approval modal. The engine
// resolves gated tools under Auto itself, so reaching
// this branch means a host injected the event directly:
// fail closed (the audit trail stays authoritative)
// instead of pausing the turn.
if approval_mode == ApprovalMode::Auto {
// The same typed dispositions used by TUI fail closed:
// a Full Access policy hold or Never posture must not be
// answered by posture or a remembered conversation grant.
let denied_posture = match disposition {
ApprovalRequestDisposition::AutoDenyFullAccessPolicyHold => {
Some("full_access_policy_hold")
}
ApprovalRequestDisposition::AutoDenyAutoReview => Some("auto_review"),
ApprovalRequestDisposition::AutoDenyNeverPosture => Some("never"),
ApprovalRequestDisposition::AutoDenySessionDenied => Some("session_denied"),
ApprovalRequestDisposition::AutoApprove
| ApprovalRequestDisposition::Prompt => None,
};
if let Some(posture) = denied_posture {
self.emit_event(
&thread_id,
Some(&turn_id),
@@ -17435,7 +17484,7 @@ impl RuntimeThreadManager {
"decision": "deny",
"remember": false,
"auto": true,
"posture": "auto_review",
"posture": posture,
}),
)
.await
@@ -17449,9 +17498,8 @@ impl RuntimeThreadManager {
// A session grant for this tool and argument class
// answers the prompt without a modal and without touching
// posture (E1). A forced prompt is never pre-answered.
if !approval_force_prompt
&& let Some(grant) =
self.session_grant_for(&thread_id, &approval_grouping_key)
if disposition == ApprovalRequestDisposition::AutoApprove
&& let Some(grant) = grant
{
let approval_id = Self::mint_approval_id();
self.emit_event(
crates/tui/src/runtime_threads/tests.rs
+296
View File
@@ -15903,6 +15903,302 @@ async fn identical_raw_tool_call_ids_on_two_threads_stay_independently_gated() -
Ok(())
}
/// Drive Runtime's actual event monitor and inspect the Engine inbox, rather
/// than restating the pure resolver's truth table (#6471).
#[tokio::test]
async fn runtime_approval_disposition_auto_paths_obey_force_and_posture() -> Result<()> {
use crate::approval_log::ApprovalDecider;
for (posture, forced, remembered, approved, denied_posture) in [
("full-access", false, false, true, None),
("full-access", false, true, true, None),
(
"full-access",
true,
false,
false,
Some("full_access_policy_hold"),
),
(
"full-access",
true,
true,
false,
Some("full_access_policy_hold"),
),
("never", false, false, false, Some("never")),
("never", false, true, false, Some("never")),
("ask", false, true, true, None),
] {
let manager = test_manager(test_runtime_dir())?;
let thread = manager
.create_thread(CreateThreadRequest {
permission_posture: Some(
if posture == "never" { "ask" } else { posture }.to_string(),
),
..Default::default()
})
.await?;
let mut harness = install_mock_engine(&manager, &thread.id).await;
let turn = manager
.start_turn(
&thread.id,
StartTurnRequest {
prompt: "honor the exact approval disposition".to_string(),
..Default::default()
},
)
.await?;
assert!(matches!(
harness.rx_op.recv().await,
Some(Op::SendMessage(TurnSpec { .. }))
));
// Never is an internal live authority, not a public Runtime posture.
if posture == "never" {
harness.handle.publish_turn_authority(
AppMode::Agent,
true,
false,
false,
ApprovalMode::Never,
None,
);
}
let grant = if remembered {
Some(
manager
.add_session_grant(&thread.id, &turn.id, "bash", "policy-group", "s")
.await
.context("fixture conversation grant")?,
)
} else {
None
};
harness
.tx_event
.send(EngineEvent::ApprovalRequired {
approval_key: "policy-key".to_string(),
approval_grouping_key: "policy-group".to_string(),
id: "policy-call".to_string(),
tool_name: "bash".to_string(),
// Text and input cannot impersonate a Rust-minted origin key.
description: "Requested by extension:claimed".to_string(),
input: json!({"approval_key": "extcall:ext:claimed", "command": "echo fixture"}),
intent_summary: None,
approval_force_prompt: forced,
})
.await?;
let decision =
tokio::time::timeout(APPROVAL_READINESS_TIMEOUT, harness.recv_approval_decision())
.await
.context("automatic disposition must not open a modal")?;
let expected = if approved {
MockApprovalEvent::Approved {
id: "policy-call".to_string(),
}
} else {
MockApprovalEvent::Denied {
id: "policy-call".to_string(),
}
};
let by = if approved && posture == "ask" {
ApprovalDecider::SessionRule
} else {
ApprovalDecider::Posture
};
assert_eq!(
decision,
Some((expected, Some(by))),
"{posture}/{forced}/{remembered}"
);
assert_eq!(manager.pending_approvals_count(), 0);
let detail = manager.get_thread_detail(&thread.id).await?;
assert!(detail.pending_approvals.is_empty());
assert_eq!(detail.approval_grants.len(), usize::from(remembered));
let decided = manager
.events_since(&thread.id, None)?
.into_iter()
.find(|event| {
event.event == "approval.decided" && event.payload["tool_call_id"] == "policy-call"
})
.context("automatic decision receipt")?;
assert_eq!(
decided.payload["decision"],
if approved { "allow" } else { "deny" }
);
assert_eq!(decided.payload["auto"], true);
assert_eq!(decided.payload["remember"], false);
if let Some(posture) = denied_posture {
assert_eq!(decided.payload["posture"], posture);
}
if approved && posture == "ask" {
assert_eq!(decided.payload["grant_id"], grant.unwrap().grant_id);
} else {
assert!(decided.payload.get("grant_id").is_none());
}
let approval_id = decided.payload["approval_id"]
.as_str()
.context("opaque decision identity")?;
assert!(approval_id.starts_with("approval_"));
assert_ne!(approval_id, "policy-call");
for id in ["policy-call", approval_id] {
assert!(!manager.deliver_external_approval(
id,
ExternalApprovalDecision::Allow { remember: false },
));
}
harness
.tx_event
.send(EngineEvent::TurnComplete {
usage: Usage::default(),
parent_route_usage: Usage::default(),
routed_usage_dropped_records: 0,
status: TurnOutcomeStatus::Completed,
error: None,
tool_catalog: None,
base_url: None,
})
.await?;
assert_eq!(
wait_for_terminal_turn(&manager, &turn.id).await?.status,
RuntimeTurnStatus::Completed,
);
}
Ok(())
}
#[tokio::test]
async fn runtime_approval_disposition_forced_requests_require_human() -> Result<()> {
use crate::approval_log::ApprovalDecider;
for (posture, tool_name, extension_origin) in [
("full-access", "bash", true),
("full-access", "web_search", true),
("ask", "bash", false),
] {
let manager = test_manager(test_runtime_dir())?;
let thread = manager
.create_thread(CreateThreadRequest {
permission_posture: Some(posture.to_string()),
..Default::default()
})
.await?;
let mut harness = install_mock_engine(&manager, &thread.id).await;
let turn = manager
.start_turn(
&thread.id,
StartTurnRequest {
prompt: "wait for the exact human decision".to_string(),
..Default::default()
},
)
.await?;
assert!(matches!(
harness.rx_op.recv().await,
Some(Op::SendMessage(TurnSpec { .. }))
));
let input = json!({"command": "echo fixture", "query": "fixture"});
let (key, group) = if extension_origin {
crate::tools::approval_cache::extension_origin_approval_keys(
"ext:fixture@reviewed-hash",
None,
tool_name,
&input,
)
} else {
crate::tools::approval_cache::approval_keys_for_call(None, tool_name, &input)
};
let grant = manager
.add_session_grant(&thread.id, &turn.id, tool_name, &group.0, "s")
.await
.context("fixture conversation grant")?;
harness
.tx_event
.send(EngineEvent::ApprovalRequired {
approval_key: key.0,
approval_grouping_key: group.0,
id: "human-call".to_string(),
tool_name: tool_name.to_string(),
description: "exact human approval required".to_string(),
input,
intent_summary: None,
approval_force_prompt: true,
})
.await?;
let approval_id = await_approval_identity(&manager, &thread.id, "human-call").await?;
let detail = manager.get_thread_detail(&thread.id).await?;
assert_eq!(detail.pending_approvals.len(), 1, "{posture}/{tool_name}");
assert_eq!(detail.pending_approvals[0].id, approval_id);
assert_eq!(detail.pending_approvals[0].turn_id, turn.id);
assert_eq!(
detail.pending_approvals[0].tool_call_id.as_deref(),
Some("human-call")
);
assert!(
tokio::time::timeout(Duration::from_millis(100), harness.recv_approval_decision())
.await
.is_err(),
"neither Full Access nor a remembered grant may answer: {posture}/{tool_name}"
);
assert!(!manager.deliver_external_approval(
"human-call",
ExternalApprovalDecision::Allow { remember: false },
));
assert!(manager.deliver_external_approval(
&approval_id,
ExternalApprovalDecision::Allow { remember: false },
));
assert_eq!(
tokio::time::timeout(APPROVAL_READINESS_TIMEOUT, harness.recv_approval_decision())
.await?,
Some((
MockApprovalEvent::Approved {
id: "human-call".to_string()
},
Some(ApprovalDecider::User),
)),
);
assert_eq!(manager.pending_approvals_count(), 0);
assert!(!manager.deliver_external_approval(
&approval_id,
ExternalApprovalDecision::Allow { remember: false },
));
assert_eq!(
manager.approval_grants_for_thread(&thread.id)[0].grant_id,
grant.grant_id
);
let decided = manager
.events_since(&thread.id, None)?
.into_iter()
.find(|event| {
event.event == "approval.decided" && event.payload["approval_id"] == approval_id
})
.context("human decision receipt")?;
assert_eq!(decided.payload["decision"], "allow");
assert_ne!(
decided.payload.get("auto").and_then(Value::as_bool),
Some(true)
);
harness
.tx_event
.send(EngineEvent::TurnComplete {
usage: Usage::default(),
parent_route_usage: Usage::default(),
routed_usage_dropped_records: 0,
status: TurnOutcomeStatus::Completed,
error: None,
tool_catalog: None,
base_url: None,
})
.await?;
assert_eq!(
wait_for_terminal_turn(&manager, &turn.id).await?.status,
RuntimeTurnStatus::Completed,
);
}
Ok(())
}
#[tokio::test]
async fn auto_review_force_prompt_is_denied_without_opening_a_modal() -> Result<()> {
let manager = test_manager(test_runtime_dir())?;