Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


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

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

13 changes: 9 additions & 4 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,12 @@ octosense-app-hub = { git = "https://github.com/OctoSense-org/OctoSense-App-Hub"
# and App Hub and Rinx no longer need the same App Hub commit.
octosense-app-contract = "1"
# --- octos: ONE source for every octos crate (octos-org/octos main at the
# merge of octos#2656, low fixes for host-owned app peers; after #2655, the
# merge of octos#2675, hydrate rows that name a tool row's call and tool;
# after #2672, the accepted client_commands echoed on session/open, #2671
# and #2669, docs, #2665, undeclared secrets kept from strict skills, #2664,
# octos's own dev profile, and #2261, a credential store that keeps
# credentials on a failed save; before that #2656, low fixes for host-owned
# app peers; #2655, the
# unknown-outcome interlock kept across a kernel restart, #2653, a bound
# workspace swapped for a symlink no longer re-roots the session, #2648, a
# host session's exact tool list; #2649,
Expand All @@ -120,12 +125,12 @@ octosense-app-contract = "1"
# keeps its peer/input, and #2645, Bing News's Chinese results).
# Move octos-core, octos-cli and octos-research together; the Android kernel
# artifact (liboctos.so) is built from this same revision.
octos-core = { git = "https://github.com/octos-org/octos.git", rev = "ae230ce04d57f3c29cf6c2518e5956a86c07d788" }
octos-cli = { git = "https://github.com/octos-org/octos.git", rev = "ae230ce04d57f3c29cf6c2518e5956a86c07d788", default-features = false }
octos-core = { git = "https://github.com/octos-org/octos.git", rev = "056173e85b150e387805fc307fe231064ac1ed35" }
octos-cli = { git = "https://github.com/octos-org/octos.git", rev = "056173e85b150e387805fc307fe231064ac1ed35", default-features = false }
# The research engine (the metasearch, the SSRF-safe polite reader and the
# toolbox scope) for the toolbox's `octos-engine` feature. Same source and
# revision as above.
octos-research = { git = "https://github.com/octos-org/octos.git", rev = "ae230ce04d57f3c29cf6c2518e5956a86c07d788", default-features = false }
octos-research = { git = "https://github.com/octos-org/octos.git", rev = "056173e85b150e387805fc307fe231064ac1ed35", default-features = false }
# --- AppCard's inline renderers.
streaming-markdown-kit = { git = "https://github.com/Project-Robius-China/streaming-markdown-kit.git", branch = "main" }
makepad-diagram-kit = { git = "https://github.com/Project-Robius-China/makepad-diagram-kit.git", branch = "main" }
Expand Down
42 changes: 33 additions & 9 deletions crates/kernel/src/kernel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,9 @@ struct Io {
/// host-managed server stops when it reaches EOF, which also happens
/// when this process dies, so a crashed shell leaves no kernel behind.
lifeline: Option<tokio::process::ChildStdin>,
/// The task copying the child's stderr into the [`Tail`]; it ends at
/// the end of the pipe, after the kernel's last line.
stderr: Option<tokio::task::JoinHandle<()>>,
}

/// Keeps the last lines the kernel wrote to stderr, to say why it exited.
Expand Down Expand Up @@ -116,7 +119,7 @@ fn start(launch: &Launch, network: &Network, log: &LogSink, tail: &Tail) -> Resu
return Err("the kernel's stdin/stdout are not piped".into());
};
// The kernel logs to stderr; it never carries protocol frames.
if let Some(stderr) = child.stderr.take() {
let stderr = child.stderr.take().map(|stderr| {
let log = log.clone();
let tail = tail.clone();
tokio::spawn(async move {
Expand All @@ -126,14 +129,14 @@ fn start(launch: &Launch, network: &Network, log: &LogSink, tail: &Tail) -> Resu
(log)(&format!("octos: {line}"));
tail.push(line);
}
});
}
})
});
let lines = BufReader::new(Box::pin(stdout) as Pin<Box<dyn AsyncRead + Send>>).lines();
Ok(if shared {
// Frames go over the WebSocket once it is up (`prepare_network`).
Io { writer: Box::pin(tokio::io::sink()), lines, running: Running::Child(child), lifeline: Some(stdin) }
Io { writer: Box::pin(tokio::io::sink()), lines, running: Running::Child(child), lifeline: Some(stdin), stderr }
} else {
Io { writer: Box::pin(stdin), lines, running: Running::Child(child), lifeline: None }
Io { writer: Box::pin(stdin), lines, running: Running::Child(child), lifeline: None, stderr }
})
}
#[cfg(target_env = "ohos")]
Expand All @@ -153,6 +156,7 @@ fn start(launch: &Launch, network: &Network, log: &LogSink, tail: &Tail) -> Resu
lines: BufReader::new(Box::pin(reader) as Pin<Box<dyn AsyncRead + Send>>).lines(),
running: Running::Embedded(task),
lifeline: None,
stderr: None,
})
}
#[cfg(not(target_env = "ohos"))]
Expand Down Expand Up @@ -266,7 +270,7 @@ async fn exited(running: &mut Running) -> String {
async fn stop(io: Io) {
// Closing stdin stops either kind of child: the stdio pipe drains owned
// turns on EOF; a host-managed server drains its connections and exits.
let Io { writer, lines, mut running, lifeline } = io;
let Io { writer, lines, mut running, lifeline, .. } = io;
let grace = if lifeline.is_some() { SHARED_GRACE } else { STDIO_GRACE };
drop(writer);
drop(lines);
Expand Down Expand Up @@ -401,16 +405,18 @@ pub(crate) async fn supervise(
}
}
}
Ok(None) => break CloseReason::Exited(with_tail("the kernel closed its output", &tail)),
Ok(None) => break CloseReason::Exited("the kernel closed its output".into()),
Err(e) => break CloseReason::Exited(format!("reading from the kernel failed: {e}")),
},
why = exited(&mut io.running) => break CloseReason::Exited(with_tail(&why, &tail)),
why = exited(&mut io.running) => break CloseReason::Exited(why),
}
}};
(log)(&format!("octos-core: stopping kernel {generation}: {reason}"));
let stderr = io.stderr.take();
stop(io).await;
match reason {
CloseReason::Failed(why) => CloseReason::Failed(with_tail(&why, &tail)),
CloseReason::Failed(why) => CloseReason::Failed(last_words(&why, stderr, &tail).await),
CloseReason::Exited(why) => CloseReason::Exited(last_words(&why, stderr, &tail).await),
other => other,
}
}
Expand Down Expand Up @@ -464,6 +470,24 @@ fn tool_list_reply(text: &str) -> Option<String> {
})
}

/// How long a stopped kernel's stderr may take to reach its end. The pipe
/// closes with the kernel; only a process it started that inherited the
/// pipe can hold it open longer.
const LAST_WORDS_GRACE: Duration = Duration::from_secs(2);

/// `why` with the kernel's last words (the [`Tail`]), once `stderr`, the
/// task reading them, has reached the end of the pipe. That task runs
/// beside the supervisor, so lines the kernel wrote just before its stdout
/// closed or it exited can still be on their way to the tail: read at once,
/// the reason said "the kernel closed its output: fake kernel up" without
/// the "asked to exit" the kernel wrote last.
async fn last_words(why: &str, stderr: Option<tokio::task::JoinHandle<()>>, tail: &Tail) -> String {
if let Some(reader) = stderr {
let _ = tokio::time::timeout(LAST_WORDS_GRACE, reader).await;
}
with_tail(why, tail)
}

fn with_tail(why: &str, tail: &Tail) -> String {
let tail = tail.text();
if tail.is_empty() {
Expand Down
4 changes: 3 additions & 1 deletion crates/kernel/tests/core.rs
Original file line number Diff line number Diff line change
Expand Up @@ -238,7 +238,9 @@ async fn a_kernel_that_dies_closes_its_connections_with_its_last_words() {
call(&mut a, "1", "session/list", json!({})).await;
a.send(json!({"jsonrpc": "2.0", "method": "test/exit", "params": {}}).to_string()).unwrap();
match next(&mut a).await {
Err(CloseReason::Exited(why)) => assert!(why.contains("asked to exit") || why.contains("exit"), "{why}"),
// The line it wrote to stderr last, whichever end the core saw
// first: its output closing or its exit status.
Err(CloseReason::Exited(why)) => assert!(why.contains("asked to exit"), "{why}"),
other => panic!("expected Exited, got {other:?}"),
}
for _ in 0..100 {
Expand Down
52 changes: 33 additions & 19 deletions crates/shell/src/approvals/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,31 @@ impl Approvals {
pub fn consent_granted(&self, app: &str) -> bool {
self.consent.granted(app, self.router.hooks().grants_all(app))
}
/// The relay's entry point (octos#2567) on these approvals: a call
/// needs an approval.
pub fn approval_requested(&mut self, app: &str, tool: ToolSpec, args: serde_json::Value, caller: Caller, context: RequestContext) -> Route {
self.router.approval_requested(app, tool, args, caller, context)
}
/// The relay installs itself; decisions made before are handed over
/// first.
pub fn set_relay(&mut self, mut relay: Box<dyn ApprovalRelay>) {
for (id, decision, reason) in self.queue.take() {
relay.approval_decided(&id, decision, &reason);
}
*self.external.lock().unwrap_or_else(|e| e.into_inner()) = Some(relay);
}
/// The AI bus's held `confirm: host` call as a request like any other.
pub fn bus_requested(&mut self, held: &crate::ai_bus::HeldCall) -> Route {
let mut tool = ToolSpec::host(&held.tool);
tool.auto_approvable = held.auto_approvable;
if held.command {
tool = tool.command();
}
let args = serde_json::from_str(&held.args).unwrap_or_else(|_| serde_json::Value::String(held.args.clone()));
// The pane is the person's own conversation with the system agent.
let context = RequestContext { call_id: held.key.clone(), trigger: Trigger::Person, ..RequestContext::default() };
self.approval_requested(&held.app, tool, args, Caller::SystemAgent, context)
}
}

static STATE: Mutex<Option<Approvals>> = Mutex::new(None);
Expand Down Expand Up @@ -160,34 +185,23 @@ pub fn now() -> u64 {

// ------------------------------------------------------------ the relay

/// The relay's entry point (octos#2567): a call needs an approval.
/// The relay's entry point (octos#2567): a call needs an approval
/// ([`Approvals::approval_requested`] on the shell's approvals).
pub fn approval_requested(app: &str, tool: ToolSpec, args: serde_json::Value, caller: Caller, context: RequestContext) -> Route {
with(|a| a.router.approval_requested(app, tool, args, caller, context)).unwrap_or_else(|| Route::Refused("approvals are not set up".into()))
with(|a| a.approval_requested(app, tool, args, caller, context)).unwrap_or_else(|| Route::Refused("approvals are not set up".into()))
}

/// The relay installs itself; decisions made before are handed over first.
pub fn set_relay(mut relay: Box<dyn ApprovalRelay>) {
with(|a| {
for (id, decision, reason) in a.queue.take() {
relay.approval_decided(&id, decision, &reason);
}
*a.external.lock().unwrap_or_else(|e| e.into_inner()) = Some(relay);
});
/// The relay installs itself; decisions made before are handed over first
/// ([`Approvals::set_relay`] on the shell's approvals).
pub fn set_relay(relay: Box<dyn ApprovalRelay>) {
with(|a| a.set_relay(relay));
}

/// The AI bus's `confirm: host` calls (`ai_bus::Route::Approval`): each is
/// a request like any other; the shell drains the answers here and
/// releases the held call (`AiBus::release`).
pub fn bus_requested(held: &crate::ai_bus::HeldCall) -> Route {
let mut tool = ToolSpec::host(&held.tool);
tool.auto_approvable = held.auto_approvable;
if held.command {
tool = tool.command();
}
let args = serde_json::from_str(&held.args).unwrap_or_else(|_| serde_json::Value::String(held.args.clone()));
// The pane is the person's own conversation with the system agent.
let context = RequestContext { call_id: held.key.clone(), trigger: Trigger::Person, ..RequestContext::default() };
approval_requested(&held.app, tool, args, Caller::SystemAgent, context)
with(|a| a.bus_requested(held)).unwrap_or_else(|| Route::Refused("approvals are not set up".into()))
}

pub fn take_bus_decisions() -> Vec<(RequestId, Decision, String)> {
Expand Down
Loading
Loading