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

fix(tui): land the operation-span and pet-owner packet, restore a compiling main

Browse Source 
bdcc8cfc5 carried one half of this packet: they touched registry.rs, which
changed `execute_rich_full_with_context` to take an `on_admitted` callback and
to complete an `OperationSpanGuard`. Its caller in tool_execution.rs was left
uncommitted, so main no longer compiled — a 3-argument call against a
4-argument signature. This lands the rest of the packet and main compiles
again.

The work itself, byte-identical to the pet/activity owner's frozen manifest
(`artifacts/dot-whale-20261009/engine-final-frozen-paths.json`, all 20 paths
verified by SHA-256 against the working tree):

- Operation spans are admitted through a callback, so a call that is refused
  never opens a span, and a span that opened always completes with a typed
  outcome, including cancellation.
- `tools::activity::action_id` bounds the canonical action identity the
  Engine reports: empty, over-256-byte or control-bearing ids fold to
  "tool" rather than reaching the transcript.
- The shared pet owner, pet_watch presentation and the codemode/event parity
  rows move with it. Protocol `engine_owner` carries `OwnerActivityKind`;
  `events.rs` and `protocol_parity.rs` keep the wire mapping in step.
- runtime_threads and runtime_api tests join the account hunks already in
  e8717e427, completing those files.

Verification receipt — the account owner's native batch ran on exactly this
working tree, and its filter covers this packet:
  nextest -p codewhale-protocol -p codewhale-tui --lib
    -E '(package(codewhale-protocol) & test(projection))
      | (package(codewhale-tui) & (test(operation) | test(activity)
         | test(pet_watch) | test(dispatch_reports_typed_operation_activity)
         | test(dropped_operation_span)))'
  Summary: 290 tests run: 290 passed, 0 failed, 15736 skipped

That batch also built codewhale-tui, which is where the broken call lived, so
it is the compilation proof for this repair as well as the test proof for the
packet. Log: artifacts/github-account-settings-20261009/engine-native-account-resumed.log

Coordination: the account owner's nine runtime_threads hunks and runtime-api
registry summary landed separately in e8717e427; this commit is the concurrent
activity/pet half they explicitly left uncommitted.
This commit is contained in:
CodeWhale Bot committed 2026-10-09 04:18:00 -07:00
1 parent e8717e427f
commit 6c2cd67ca3
17 files changed
+810 -228
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/protocol/src/engine_owner.rs
+28 -7
View File
@@ -2,7 +2,7 @@
//!
//! The Engine owns operation and turn facts: it emits
//! `operation_activity_started` / `operation_activity_completed` with an
//! [`OwnerActivityKind`] and [`OwnerOperationOutcome`], never a tool name,
//! [`OwnerActivityKind`] and [`OwnerOperationOutcome`], a bounded canonical action identity, never an
//! argument, command or result. Apps may bind this session-scoped projection
//! to an authenticated account and serve it to another client; the account
//! identity is intentionally not supplied by the Engine.
@@ -17,7 +17,6 @@
//! validated here on the way back. No Rust producer exists yet; a host
//! that needs it outside the pet must decide whether the runtime produces
//! it in Rust rather than adding a second reducer.
//! - Code-mode (`execute_tools`) nested calls do not report activity yet.
use serde::{Deserialize, Serialize};
@@ -91,6 +90,8 @@ pub enum OwnerFreshness {
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct OwnerActiveSpan {
pub activity_kind: OwnerActivityKind,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub action_id: Option<String>,
pub started_at_ms: f64,
}
@@ -98,6 +99,8 @@ pub struct OwnerActiveSpan {
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct OwnerFailedToolAge {
pub activity_kind: OwnerActivityKind,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub action_id: Option<String>,
pub age_ms: f64,
}
@@ -114,6 +117,8 @@ pub struct EngineOwnerProjection {
pub freshness: OwnerFreshness,
pub authoritative_presence: OwnerPresence,
pub activity_kind: Option<OwnerActivityKind>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub action_id: Option<String>,
pub observed_at_ms: Option<f64>,
pub parallel_agent_count: u32,
pub active_spans: Vec<OwnerActiveSpan>,
@@ -126,7 +131,11 @@ pub struct EngineOwnerProjection {
impl EngineOwnerProjection {
#[must_use]
pub fn is_valid(&self) -> bool {
let valid_action =
|id: &String| !id.is_empty() && id.len() <= 256 && !id.chars().any(char::is_control);
self.schema_version == 1
&& self.action_id.as_ref().is_none_or(valid_action)
&& (self.action_id.is_none() || self.activity_kind.is_some())
&& self.session_id.as_ref().is_none_or(|id| {
!id.is_empty() && id.len() <= 256 && id.bytes().all(|byte| !byte.is_ascii_control())
})
@@ -139,16 +148,18 @@ impl EngineOwnerProjection {
.observed_at_ms
.is_none_or(|time| time.is_finite() && time >= 0.0)
&& self.active_spans.iter().all(|span| {
span.started_at_ms.is_finite()
span.action_id.as_ref().is_none_or(valid_action)
&& span.started_at_ms.is_finite()
&& span.started_at_ms >= 0.0
&& self
.observed_at_ms
.is_none_or(|observed_at| span.started_at_ms <= observed_at)
})
&& self
.failed_tool_age
.as_ref()
.is_none_or(|failure| failure.age_ms.is_finite() && failure.age_ms >= 0.0)
&& self.failed_tool_age.as_ref().is_none_or(|failure| {
failure.action_id.as_ref().is_none_or(valid_action)
&& failure.age_ms.is_finite()
&& failure.age_ms >= 0.0
})
&& (self.turn_outcome.is_none() || self.turn_id.is_some())
&& match self.freshness {
OwnerFreshness::Missing => {
@@ -202,10 +213,12 @@ mod tests {
freshness: OwnerFreshness::Fresh,
authoritative_presence: OwnerPresence::Working,
activity_kind: Some(OwnerActivityKind::Reading),
action_id: Some("read_file".into()),
observed_at_ms: Some(100.0),
parallel_agent_count: 0,
active_spans: vec![OwnerActiveSpan {
activity_kind: OwnerActivityKind::Reading,
action_id: Some("read_file".into()),
started_at_ms: 50.0,
}],
turn_id: Some("turn-1".into()),
@@ -218,6 +231,11 @@ mod tests {
#[test]
fn projection_accepts_trusted_scoped_activity_and_rejects_bad_clocks() {
assert!(projection().is_valid());
for invalid in ["", "bad\nname", &"鲸".repeat(86)] {
let mut invalid_action = projection();
invalid_action.action_id = Some(invalid.to_owned());
assert!(!invalid_action.is_valid());
}
let mut future_span = projection();
future_span.active_spans[0].started_at_ms = 101.0;
@@ -237,6 +255,7 @@ mod tests {
let mut done = projection();
done.authoritative_presence = OwnerPresence::Done;
done.activity_kind = None;
done.action_id = None;
done.active_spans.clear();
done.turn_outcome = Some(TurnOutcomeStatus::Completed);
done.done_effect_id = done.turn_id.clone();
@@ -258,6 +277,7 @@ mod tests {
freshness: OwnerFreshness::Missing,
authoritative_presence: OwnerPresence::Unknown,
activity_kind: None,
action_id: None,
observed_at_ms: None,
parallel_agent_count: 0,
active_spans: Vec::new(),
@@ -272,6 +292,7 @@ mod tests {
stale.freshness = OwnerFreshness::Stale;
stale.authoritative_presence = OwnerPresence::Unknown;
stale.activity_kind = None;
stale.action_id = None;
stale.parallel_agent_count = 0;
stale.active_spans.clear();
stale.failed_tool_age = None;
crates/protocol/src/event_msg.rs
+7 -5
View File
@@ -375,17 +375,15 @@ pub enum EventMsg {
result: ToolCallOutcome,
},
/// Trusted Engine-owned activity for an operation that passed dispatch
/// and authority checks. No tool name, arguments, command, or result is
/// and authority checks. No arguments, command, or result is
/// included in the pet-facing activity contract.
///
/// Reserved on the wire: `protocol_parity` maps the engine event, but no
/// runtime thread emits the pair yet, so Runtime API and GPUI clients do
/// not receive it. The shared pet (`pet_watch`) is the only consumer today.
OperationActivityStarted {
thread_id: ThreadId,
session_id: SessionId,
span_id: String,
activity_kind: crate::engine_owner::OwnerActivityKind,
#[serde(default, skip_serializing_if = "Option::is_none")]
action_id: Option<String>,
},
OperationActivityCompleted {
thread_id: ThreadId,
@@ -393,6 +391,8 @@ pub enum EventMsg {
span_id: String,
activity_kind: crate::engine_owner::OwnerActivityKind,
outcome: crate::engine_owner::OwnerOperationOutcome,
#[serde(default, skip_serializing_if = "Option::is_none")]
action_id: Option<String>,
},
// === Turn lifecycle ===
@@ -1133,12 +1133,14 @@ mod tests {
session_id: s.clone(),
span_id: "span-1".into(),
activity_kind: crate::engine_owner::OwnerActivityKind::Reading,
action_id: Some("read_file".into()),
},
EventMsg::OperationActivityCompleted {
thread_id: t.clone(),
session_id: s.clone(),
span_id: "span-1".into(),
activity_kind: crate::engine_owner::OwnerActivityKind::Reading,
action_id: Some("read_file".into()),
outcome: crate::engine_owner::OwnerOperationOutcome::Succeeded,
},
EventMsg::TurnStarted {
crates/tui/src/core/engine/tests/test_cases_01.rs
+2
View File
@@ -736,6 +736,7 @@ async fn event_capacity_cancelled_parallel_tool_keeps_completed_span_and_call_wh
Event::OperationActivityStarted {
span_id,
activity_kind,
..
} => Some((span_id, activity_kind)),
_ => None,
})
@@ -748,6 +749,7 @@ async fn event_capacity_cancelled_parallel_tool_keeps_completed_span_and_call_wh
span_id,
activity_kind,
outcome,
..
} => Some((span_id, activity_kind, outcome)),
_ => None,
})
crates/tui/src/core/engine/tests/test_cases_16.rs
+7 -6
View File
@@ -649,7 +649,7 @@ async fn execute_tools_dispatches_through_common_executor() {
}
#[tokio::test]
async fn dispatch_reports_typed_operation_activity_without_names_or_arguments() {
async fn dispatch_reports_typed_operation_activity_with_safe_identity_without_arguments() {
use crate::tools::file_tool::ReadTool;
use crate::tools::registry::ToolRegistryBuilder;
use crate::tools::spec::ToolContext;
@@ -702,13 +702,12 @@ async fn dispatch_reports_typed_operation_activity_without_names_or_arguments()
matches!(
events.as_slice(),
[
Event::OperationActivityStarted { span_id: started, activity_kind: OwnerActivityKind::Reading },
Event::OperationActivityStarted { span_id: started, activity_kind: OwnerActivityKind::Reading, action_id: Some(action) },
Event::OperationActivityCompleted {
span_id: completed,
activity_kind: OwnerActivityKind::Reading,
outcome: OwnerOperationOutcome::Succeeded,
},
] if started == completed && started.starts_with("call-1#")
outcome: OwnerOperationOutcome::Succeeded, action_id: Some(completed_action) },
] if started == completed && started.starts_with("call-1#") && action == "read_file" && completed_action == action
),
"unexpected activity: {events:?}"
);
@@ -798,9 +797,10 @@ async fn dropped_operation_span_completes_as_cancelled() {
// opened must still close, or every host leaks an active operation.
let (tx_event, mut rx_event) = mpsc::channel(16);
let span = super::tool_execution::OperationSpanGuard::start(
tx_event,
&tx_event,
"call-x",
OwnerActivityKind::Editing,
Some("edit_file".into()),
None,
)
.await;
@@ -814,6 +814,7 @@ async fn dropped_operation_span_completes_as_cancelled() {
span_id,
activity_kind: OwnerActivityKind::Editing,
outcome: OwnerOperationOutcome::Cancelled,
..
}) => assert_eq!(span_id, started),
other => panic!("unexpected event: {other:?}"),
}
crates/tui/src/core/engine/tool_execution.rs
+259 -197
View File
@@ -49,20 +49,20 @@ impl Drop for ReplaySpanSequenceGuard {
/// Pairs an observed `OperationActivityStarted` with at most one completion.
///
/// The turn loop drops an in-flight tool future when the user cancels
/// (`tokio::select!` on the cancel token, or `drop(tool_tasks)` for a parallel
/// batch), so a Completed sent inline after the await would never be sent.
/// Dropping an unfinished span sends `Completed { Cancelled }` with
/// `try_send`: best effort, like the other guards here, because `Drop` cannot
/// await a full channel. Cancellation may release a completion parked on a
/// full queue; the turn's reserved terminal observation settles its lifecycle.
pub(super) struct OperationSpanGuard {
tx: mpsc::Sender<Event>,
span: Option<(String, codewhale_protocol::engine_owner::OwnerActivityKind)>,
cancel: Option<CancellationToken>,
/// The turn loop drops an in-flight tool future when the user cancels.
/// Capacity reserved before admission keeps completion synchronous even when
/// a nested call is withdrawn while its parent continues.
pub(crate) struct OperationSpanGuard<'a> {
start: Option<mpsc::Permit<'a, Event>>,
span: Option<(
mpsc::Permit<'a, Event>,
String,
codewhale_protocol::engine_owner::OwnerActivityKind,
Option<String>,
)>,
}
impl OperationSpanGuard {
impl<'a> OperationSpanGuard<'a> {
/// A process-unique span id. The model's tool-call id is not unique:
/// gateways that elide ids fall back to `call_{block_index}`, which
/// repeats every step, and a consumer deduplicates completed spans.
@@ -83,62 +83,87 @@ impl OperationSpanGuard {
format!("{call_id}#{seq}")
}
pub(super) async fn start(
tx: mpsc::Sender<Event>,
pub(crate) async fn reserve(
tx: &'a mpsc::Sender<Event>,
call_id: &str,
activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind,
action_id: Option<String>,
cancel: Option<CancellationToken>,
) -> Self {
let span_id = Self::span_id(call_id);
// Reservation is cancel safe: if this future is dropped the event
// was either sent or not, and the guard is armed only once it was.
let sent = match super::streaming::reserve_event_capacity(
&tx,
cancel.as_ref(),
super::streaming::EventReservationPolicy::Strict,
)
.await
{
Ok(permit) => {
permit.send(Event::OperationActivityStarted {
span_id: span_id.clone(),
activity_kind,
});
true
}
Err(_) => false,
let reserve = tx.reserve_many(2);
let permits = match cancel.as_ref() {
Some(cancel) => tokio::select! {
biased;
() = cancel.cancelled() => None,
result = reserve => result.ok(),
},
None => reserve.await.ok(),
};
let Some(mut permits) = permits else {
return Self {
start: None,
span: None,
};
};
let start = permits.next().expect("reserved activity start");
let completion = permits.next().expect("reserved activity completion");
Self {
tx,
span: sent.then_some((span_id, activity_kind)),
cancel,
start: Some(start),
span: Some((completion, Self::span_id(call_id), activity_kind, action_id)),
}
}
async fn complete(mut self, outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome) {
if let Some((span_id, activity_kind)) = self.span.take()
&& let Ok(permit) = super::streaming::reserve_event_capacity(
&self.tx,
self.cancel.as_ref(),
super::streaming::EventReservationPolicy::Receipt,
)
pub(crate) fn activate(mut self) -> Self {
if let Some(start) = self.start.take()
&& let Some((_, span_id, activity_kind, action_id)) = &self.span
{
start.send(Event::OperationActivityStarted {
span_id: span_id.clone(),
activity_kind: *activity_kind,
action_id: action_id.clone(),
});
}
self
}
pub(crate) async fn start(
tx: &'a mpsc::Sender<Event>,
call_id: &str,
activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind,
action_id: Option<String>,
cancel: Option<CancellationToken>,
) -> Self {
Self::reserve(tx, call_id, activity_kind, action_id, cancel)
.await
.activate()
}
pub(crate) async fn complete(
mut self,
outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome,
) {
if self.start.is_none()
&& let Some((permit, span_id, activity_kind, action_id)) = self.span.take()
{
permit.send(Event::OperationActivityCompleted {
span_id,
activity_kind,
action_id,
outcome,
});
}
}
}
impl Drop for OperationSpanGuard {
impl Drop for OperationSpanGuard<'_> {
fn drop(&mut self) {
if let Some((span_id, activity_kind)) = self.span.take() {
let _ = self.tx.try_send(Event::OperationActivityCompleted {
if self.start.is_none()
&& let Some((permit, span_id, activity_kind, action_id)) = self.span.take()
{
permit.send(Event::OperationActivityCompleted {
span_id,
activity_kind,
action_id,
outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome::Cancelled,
});
}
@@ -590,7 +615,7 @@ impl Engine {
// Typed owner activity: classified only after every gate above has
// passed, from the same authority that dispatches the call (the MCP
// pool's resolved server map, the interpreter, or the registry plus
// the canonical action alias). Names and arguments never leave here.
// the canonical action alias). Arguments never leave here.
let activity_kind = match activity_call_id.as_ref() {
None => None,
Some(_) if McpPool::is_mcp_tool(&tool_name) => match mcp_pool.as_ref() {
@@ -605,31 +630,17 @@ impl Engine {
Some(_)
if matches!(
tool_name.as_str(),
CODE_EXECUTION_TOOL_NAME | JS_EXECUTION_TOOL_NAME
CODE_EXECUTION_TOOL_NAME | JS_EXECUTION_TOOL_NAME | EXECUTE_TOOLS_TOOL_NAME
) =>
{
Some(codewhale_protocol::engine_owner::OwnerActivityKind::Executing)
}
// The code-mode wrapper is not itself an operation.
Some(_) if tool_name == EXECUTE_TOOLS_TOOL_NAME => None,
Some(_) => registry
.filter(|registry| registry.get(&tool_name).is_some())
.and_then(|_| {
crate::tools::activity::registry_activity_kind(&tool_name, &tool_input)
}),
};
let operation_span = match (activity_call_id.as_deref(), activity_kind) {
(Some(call_id), Some(activity_kind)) => Some(
OperationSpanGuard::start(
tx_event.clone(),
call_id,
activity_kind,
cancel_token.clone(),
)
.await,
),
_ => None,
};
if cancel_token
.as_ref()
.is_some_and(CancellationToken::is_cancelled)
@@ -660,136 +671,172 @@ impl Engine {
} else {
None
};
if let Some(context) = context_override
let context = context_override
.as_ref()
.or_else(|| registry.map(|registry| registry.context()))
&& (context.acp_host.is_some() || context.child_host.is_some())
.or_else(|| registry.map(|registry| registry.context()));
let execution_receipt = if context
.is_some_and(|context| context.acp_host.is_some() || context.child_host.is_some())
&& activity_call_id.is_some()
{
let spec = registry
.and_then(|registry| registry.get(&tool_name))
.ok_or_else(|| {
ToolError::not_available("ACP call has no admitted registered tool")
})?;
crate::tools::registry::enforce_tool_authority(
&tool_name,
&tool_input,
spec.as_ref(),
context,
)?;
crate::extension_host::validate_caller_plugins(context.plugin_registry.as_deref())
.map_err(ToolError::not_available)?;
if let Some(id) = activity_call_id.as_ref() {
if let Ok(permit) = super::streaming::reserve_event_capacity(
Some(
super::streaming::reserve_event_capacity(
&tx_event,
cancel_token.as_ref(),
super::streaming::EventReservationPolicy::Receipt,
)
.await
{
// Event backpressure is an await boundary. Recheck the
// exact captured authority before claiming dispatch.
if cancel_token
.as_ref()
.is_some_and(CancellationToken::is_cancelled)
{
return Err(ToolError::cancelled("ACP dispatch admission cancelled"));
}
crate::tools::registry::enforce_tool_authority(
&tool_name,
&tool_input,
spec.as_ref(),
context,
)?;
crate::extension_host::validate_caller_plugins(
context.plugin_registry.as_deref(),
)
.map_err(ToolError::not_available)?;
permit.send(Event::ToolExecutionStarted { id: id.clone() });
} else {
return Err(ToolError::cancelled("ACP dispatch observation unavailable"));
}
.map_err(|_| ToolError::cancelled("ACP dispatch observation unavailable"))?,
)
} else {
None
};
let operation_span = match (activity_call_id.as_deref(), activity_kind) {
(Some(call_id), Some(kind)) => Some(
OperationSpanGuard::reserve(
&tx_event,
call_id,
kind,
Some(crate::tools::activity::action_id(&tool_name, &tool_input)),
cancel_token.clone(),
)
.await,
),
_ => None,
};
let on_admitted = || {
if cancel_token
.as_ref()
.is_some_and(CancellationToken::is_cancelled)
{
return Err(ToolError::cancelled("Tool activity admission cancelled."));
}
}
let outcome: Result<RichToolResult, ToolError> = if McpPool::is_mcp_tool(&tool_name) {
if let Some(pool) = mcp_pool {
let disallowed_tools = context_override
if let Some(context) = context
&& (context.acp_host.is_some() || context.child_host.is_some())
{
let spec = registry
.and_then(|registry| registry.get(&tool_name))
.ok_or_else(|| {
ToolError::not_available("ACP call has no admitted registered tool")
})?;
crate::tools::registry::enforce_tool_authority(
&tool_name,
&tool_input,
spec.as_ref(),
context,
)?;
crate::extension_host::validate_caller_plugins(context.plugin_registry.as_deref())
.map_err(ToolError::not_available)?;
}
if let Some(permit) = execution_receipt {
permit.send(Event::ToolExecutionStarted {
id: activity_call_id
.as_ref()
.expect("reserved execution identity")
.clone(),
});
}
Ok(operation_span.map(OperationSpanGuard::activate))
};
let registry_dispatch = !McpPool::is_mcp_tool(&tool_name)
&& !matches!(
tool_name.as_str(),
CODE_EXECUTION_TOOL_NAME | JS_EXECUTION_TOOL_NAME | EXECUTE_TOOLS_TOOL_NAME
);
let outcome = if registry_dispatch {
match registry {
Some(registry) => {
registry
.execute_rich_full_with_context(
&tool_name,
tool_input.clone(),
context_override.as_ref(),
on_admitted,
)
.await
}
None => Err(ToolError::not_available(format!(
"tool '{tool_name}' is not registered"
))),
}
} else {
let operation_span = on_admitted()?;
let outcome: Result<RichToolResult, ToolError> = if McpPool::is_mcp_tool(&tool_name) {
if let Some(pool) = mcp_pool {
let disallowed_tools = context_override
.as_ref()
.or_else(|| registry.map(|registry| registry.context()))
.map(|context| context.disallowed_tools.as_slice())
.unwrap_or_default();
// Only a per-call override can carry a person's decision; the
// registry's shared context never does.
let decision = context_override
.as_ref()
.and_then(|context| context.human_decision.as_ref());
Engine::execute_mcp_tool_with_pool(
pool,
&tx_event,
&tool_name,
tool_input,
disallowed_tools,
decision,
)
.await
} else {
Err(ToolError::not_available(format!(
"tool '{tool_name}' is not registered"
)))
}
} else if matches!(
tool_name.as_str(),
CODE_EXECUTION_TOOL_NAME | JS_EXECUTION_TOOL_NAME
) {
if let Some(context) = context_override
.as_ref()
.or_else(|| registry.map(|registry| registry.context()))
.map(|context| context.disallowed_tools.as_slice())
.unwrap_or_default();
// Only a per-call override can carry a person's decision; the
// registry's shared context never does.
let decision = context_override
.as_ref()
.and_then(|context| context.human_decision.as_ref());
Engine::execute_mcp_tool_with_pool(
pool,
&tx_event,
&tool_name,
tool_input,
disallowed_tools,
decision,
)
.await
} else {
Err(ToolError::not_available(format!(
"tool '{tool_name}' is not registered"
)))
}
} else if matches!(
tool_name.as_str(),
CODE_EXECUTION_TOOL_NAME | JS_EXECUTION_TOOL_NAME
) {
if let Some(context) = context_override
.as_ref()
.or_else(|| registry.map(|registry| registry.context()))
{
let result = if tool_name == CODE_EXECUTION_TOOL_NAME {
execute_code_execution_tool(&tool_input, &workspace, context).await
{
let result = if tool_name == CODE_EXECUTION_TOOL_NAME {
execute_code_execution_tool(&tool_input, &workspace, context).await
} else {
execute_js_execution_tool(&tool_input, &workspace, context).await
};
result.map(RichToolResult::plain)
} else {
execute_js_execution_tool(&tool_input, &workspace, context).await
};
result.map(RichToolResult::plain)
} else {
Err(ToolError::not_available(
"local code execution requires an effective tool context",
))
}
} else if tool_name == EXECUTE_TOOLS_TOOL_NAME {
if let Some(registry) = registry {
let context = context_override
.as_ref()
.cloned()
.unwrap_or_else(|| registry.context().clone());
crate::tools::codemode::execute_tools_tool(&tool_input, registry, &context)
.await
.map(RichToolResult::plain)
Err(ToolError::not_available(
"local code execution requires an effective tool context",
))
}
} else if tool_name == EXECUTE_TOOLS_TOOL_NAME {
if let Some(registry) = registry {
let context = context_override
.as_ref()
.cloned()
.unwrap_or_else(|| registry.context().clone());
crate::tools::codemode::execute_tools_tool(&tool_input, registry, &context)
.await
.map(RichToolResult::plain)
} else {
Err(ToolError::not_available(format!(
"tool '{tool_name}' is not registered"
)))
}
} else {
Err(ToolError::not_available(format!(
"tool '{tool_name}' is not registered"
)))
};
if let Some(operation_span) = operation_span {
operation_span
.complete(crate::tools::activity::operation_outcome(
&outcome,
cancel_token
.as_ref()
.is_some_and(CancellationToken::is_cancelled),
))
.await;
}
} else if let Some(registry) = registry {
registry
.execute_rich_full_with_context(&tool_name, tool_input, context_override.as_ref())
.await
} else {
Err(ToolError::not_available(format!(
"tool '{tool_name}' is not registered"
)))
outcome
};
if let Some(operation_span) = operation_span {
let cancelled = cancel_token
.as_ref()
.is_some_and(CancellationToken::is_cancelled);
operation_span
.complete(crate::tools::activity::operation_outcome(
&outcome, cancelled,
))
.await;
}
if outcome.as_ref().is_ok_and(|result| result.result.success)
&& let Some((authority, writes)) = child_mcp_call
{
@@ -888,9 +935,10 @@ mod tests {
let (tx, mut rx) = mpsc::channel(2);
let cancel = CancellationToken::new();
let span = OperationSpanGuard::start(
tx,
&tx,
"completed-call",
OwnerActivityKind::Tool,
Some("tool".into()),
Some(cancel.clone()),
)
.await;
@@ -901,6 +949,7 @@ mod tests {
let Event::OperationActivityStarted {
span_id,
activity_kind,
action_id,
} = rx.try_recv().unwrap()
else {
panic!("activity start");
@@ -909,6 +958,7 @@ mod tests {
span_id: completed_span,
activity_kind: completed_kind,
outcome: completed_outcome,
action_id: completed_action,
} = rx
.try_recv()
.expect("cancellation must not discard the completion")
@@ -917,6 +967,7 @@ mod tests {
};
assert_eq!(completed_span, span_id);
assert_eq!(completed_kind, activity_kind);
assert_eq!(completed_action, action_id);
assert_eq!(
completed_outcome, outcome,
"retain the observed outcome exactly"
@@ -924,30 +975,41 @@ mod tests {
assert!(rx.try_recv().is_err(), "exactly one completion");
}
let (tx, mut rx) = mpsc::channel(512);
let mut spans = Vec::new();
for i in 0..256 {
spans.push(
OperationSpanGuard::start(
&tx,
&format!("full-call-{i}"),
OwnerActivityKind::Tool,
Some("tool".into()),
None,
)
.await,
);
}
assert_eq!(tx.capacity(), 0);
drop(spans);
let mut starts = 0;
let mut cancelled = 0;
while let Ok(event) = rx.try_recv() {
match event {
Event::OperationActivityStarted { .. } => starts += 1,
Event::OperationActivityCompleted {
outcome: OwnerOperationOutcome::Cancelled,
..
} => cancelled += 1,
other => panic!("unexpected activity receipt: {other:?}"),
}
}
assert_eq!((starts, cancelled), (256, 256));
assert_eq!(tx.capacity(), 512);
let (tx, mut rx) = mpsc::channel(1);
let cancel = CancellationToken::new();
let span = OperationSpanGuard::start(
tx,
"full-call",
OwnerActivityKind::Tool,
Some(cancel.clone()),
)
.await;
cancel.cancel();
tokio::time::timeout(
Duration::from_millis(100),
span.complete(OwnerOperationOutcome::Succeeded),
)
.await
.expect("a full queue cannot retain a cancelled activity sender");
assert!(matches!(
rx.try_recv().unwrap(),
Event::OperationActivityStarted { .. }
));
assert!(
rx.try_recv().is_err(),
"no sender survives to publish after draining"
);
let span =
OperationSpanGuard::start(&tx, "too-small", OwnerActivityKind::Tool, None, None).await;
drop(span);
assert!(rx.try_recv().is_err(), "never publish an unpaired start");
}
#[tokio::test]
crates/tui/src/core/events.rs
+3 -1
View File
@@ -267,15 +267,17 @@ pub enum Event {
/// Trusted operation activity emitted only after dispatch and authority
/// gates resolve the underlying operation. The payload deliberately
/// excludes tool names, arguments, commands, and results.
/// excludes arguments, commands, and results.
OperationActivityStarted {
span_id: String,
activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind,
action_id: Option<String>,
},
OperationActivityCompleted {
span_id: String,
activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind,
outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome,
action_id: Option<String>,
},
// === Turn Lifecycle ===
crates/tui/src/core/protocol_parity.rs
+4
View File
@@ -539,21 +539,25 @@ pub fn event_to_protocol(event: &Event, ids: &ProtocolIds) -> wire::EventMsg {
Event::OperationActivityStarted {
span_id,
activity_kind,
action_id,
} => wire::EventMsg::OperationActivityStarted {
thread_id,
session_id,
span_id: span_id.clone(),
activity_kind: *activity_kind,
action_id: action_id.clone(),
},
Event::OperationActivityCompleted {
span_id,
activity_kind,
action_id,
outcome,
} => wire::EventMsg::OperationActivityCompleted {
thread_id,
session_id,
span_id: span_id.clone(),
activity_kind: *activity_kind,
action_id: action_id.clone(),
outcome: *outcome,
},
Event::TurnStarted {
crates/tui/src/runtime_api/tests.rs
+4
View File
@@ -468,6 +468,7 @@ fn messages_from_thread_detail_batches_tool_results() {
duration_ms: Some(0),
usage: None,
model_request_diagnostics: None,
operation_activity: None,
routing_settlement: false,
effective_route_usage: None,
permission_posture: Some("ask".to_string()),
@@ -9797,6 +9798,7 @@ async fn session_save_merges_thread_cost_split_and_records_coverage() -> Result<
..Usage::default()
}),
model_request_diagnostics: None,
operation_activity: None,
routing_settlement: false,
effective_route_usage: None,
permission_posture: None,
@@ -9996,6 +9998,7 @@ async fn session_save_persists_parent_cny_unpriced_reasons_without_double_count(
..Usage::default()
}),
model_request_diagnostics: None,
operation_activity: None,
routing_settlement: false,
effective_route_usage: None,
permission_posture: None,
@@ -11344,6 +11347,7 @@ fn seed_summary_search_transcript(
duration_ms: Some(0),
usage: None,
model_request_diagnostics: None,
operation_activity: None,
routing_settlement: false,
effective_route_usage: None,
permission_posture: None,
crates/tui/src/runtime_threads.rs
+133
View File
@@ -1446,6 +1446,35 @@ impl From<&crate::tool_inspection::TurnStopDiagnostics> for RuntimeTurnRequestDi
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RuntimeOperationSpan {
pub span_id: String,
pub activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub action_id: Option<String>,
pub started_at: DateTime<Utc>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RuntimeOperationCompletion {
pub span_id: String,
pub activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub action_id: Option<String>,
pub completed_at: DateTime<Utc>,
pub outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RuntimeOperationActivity {
pub observed_at: DateTime<Utc>,
pub active: Vec<RuntimeOperationSpan>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_completed: Option<RuntimeOperationCompletion>,
#[serde(default)]
pub overflowed: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TurnRecord {
/// Admitted per-request allowance. Older turns have no explicit allowance.
@@ -1601,6 +1630,8 @@ pub struct TurnRecord {
skip_serializing_if = "Option::is_none"
)]
pub model_request_diagnostics: Option<RuntimeTurnRequestDiagnostics>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub operation_activity: Option<RuntimeOperationActivity>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(default)]
@@ -1635,6 +1666,24 @@ pub struct TurnRecord {
impl TurnRecord {
fn validate_output_token_limit(&self) -> Result<()> {
if let Some(activity) = &self.operation_activity {
let safe_id =
|id: &str| !id.is_empty() && id.len() <= 256 && !id.chars().any(char::is_control);
anyhow::ensure!(
activity.active.len() <= 256
&& activity.active.iter().all(|span| {
safe_id(&span.span_id)
&& span.action_id.as_deref().is_none_or(safe_id)
&& span.started_at <= activity.observed_at
})
&& activity.last_completed.as_ref().is_none_or(|span| {
safe_id(&span.span_id)
&& span.action_id.as_deref().is_none_or(safe_id)
&& span.completed_at <= activity.observed_at
}),
"Turn operation activity exceeds its bound"
);
}
if self.decision_receipts.len() > MAX_ROUTED_USAGE_RECORDS_PER_TURN
|| self.decision_receipts.iter().any(|r| !r.is_bounded())
{
@@ -1962,6 +2011,7 @@ fn settle_unaccepted_routed_usage(
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: Some(UNACCEPTED_TURN_REASON.to_string()),
item_ids: Vec::new(),
steer_count: 0,
@@ -13244,6 +13294,7 @@ impl RuntimeThreadManager {
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids,
steer_count: 0,
@@ -13648,6 +13699,9 @@ impl RuntimeThreadManager {
.saturating_add(background_residual);
if turn.status == RuntimeTurnStatus::InProgress {
turn.status = RuntimeTurnStatus::Failed;
if let Some(activity) = &mut turn.operation_activity {
activity.active.clear();
}
turn.ended_at = Some(now);
turn.duration_ms = turn.started_at.map(|start| duration_ms(start, now));
turn.error = Some(reason.to_string());
@@ -14544,6 +14598,7 @@ impl RuntimeThreadManager {
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: Vec::new(),
steer_count: 0,
@@ -15054,6 +15109,7 @@ impl RuntimeThreadManager {
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: Vec::new(),
steer_count: 0,
@@ -16779,6 +16835,77 @@ impl RuntimeThreadManager {
)
.await?;
}
operation @ (EngineEvent::OperationActivityStarted { .. }
| EngineEvent::OperationActivityCompleted { .. }
| EngineEvent::ToolCallHeartbeat) => {
if !saw_turn_started {
continue;
}
if !matches!(operation, EngineEvent::ToolCallHeartbeat) {
saw_engine_activity = true;
}
let projection_lock = self.projection_lock(&thread_id);
let _projection = projection_lock.lock().await;
let store = self.store.clone();
let operation_turn_id = turn_id.clone();
let projected = tokio::task::spawn_blocking(move || {
let _turn_mutation = store.turn_mutation.lock();
let mut turn = store.load_turn(&operation_turn_id)?;
if turn.status != RuntimeTurnStatus::InProgress {
return Ok::<_, anyhow::Error>(None);
}
if matches!(operation, EngineEvent::ToolCallHeartbeat)
&& turn.operation_activity.as_ref().is_none_or(|a| a.active.is_empty())
{
return Ok(None);
}
let now = Utc::now();
let activity = turn.operation_activity.get_or_insert_with(|| RuntimeOperationActivity {
observed_at: now,
active: Vec::new(),
last_completed: None,
overflowed: false,
});
let now = now.max(activity.observed_at);
activity.observed_at = now;
let (name, mut payload) = match operation {
EngineEvent::OperationActivityStarted { span_id, activity_kind, action_id } => {
let span_id = format!("operation:{}", crate::hashing::sha256_hex(span_id));
let action_id = action_id.filter(|id| !id.is_empty() && id.len() <= 256 && !id.chars().any(char::is_control));
if !activity.active.iter().any(|span| span.span_id == span_id) {
if activity.active.len() < 256 {
activity.active.push(RuntimeOperationSpan {
span_id: span_id.clone(), activity_kind, action_id: action_id.clone(), started_at: now,
});
} else {
activity.overflowed = true;
}
}
("operation.activity_started", json!({"span_id":span_id,"activity_kind":activity_kind,"action_id":action_id,"observed_at":now}))
}
EngineEvent::OperationActivityCompleted { span_id, activity_kind, action_id, outcome } => {
let span_id = format!("operation:{}", crate::hashing::sha256_hex(span_id));
let action_id = action_id.filter(|id| !id.is_empty() && id.len() <= 256 && !id.chars().any(char::is_control));
activity.active.retain(|span| span.span_id != span_id);
activity.last_completed = Some(RuntimeOperationCompletion {
span_id: span_id.clone(), activity_kind, action_id: action_id.clone(), completed_at: now, outcome,
});
("operation.activity_completed", json!({"span_id":span_id,"activity_kind":activity_kind,"action_id":action_id,"outcome":outcome,"observed_at":now}))
}
EngineEvent::ToolCallHeartbeat => ("operation.heartbeat", json!({"observed_at":now})),
_ => unreachable!(),
};
if payload.get("action_id").is_some_and(Value::is_null) {
payload.as_object_mut().expect("operation payload").remove("action_id");
}
store.save_turn(&turn)?;
Ok(Some((name, payload)))
}).await.context("Runtime operation projection task failed")??;
if let Some((name, payload)) = projected {
self.emit_event(&thread_id, Some(&turn_id), None, name, payload)
.await?;
}
}
EngineEvent::ToolExecutionStarted { id } => {
if let Some(item_id) = tool_items.get(&id) {
self.emit_event(
@@ -18298,6 +18425,9 @@ impl RuntimeThreadManager {
let _turn_mutation = self.store.turn_mutation.lock();
let mut turn = self.store.load_turn(&turn_id)?;
turn.status = turn_status;
if let Some(activity) = &mut turn.operation_activity {
activity.active.clear();
}
turn.ended_at = Some(ended_at);
turn.duration_ms = turn.started_at.map(|start| duration_ms(start, ended_at));
turn.usage = turn_usage;
@@ -18720,6 +18850,9 @@ impl RuntimeThreadManager {
}
if interrupted_candidate {
turn.status = RuntimeTurnStatus::Interrupted;
if let Some(activity) = &mut turn.operation_activity {
activity.active.clear();
}
turn.error = Some(RUNTIME_RESTART_REASON.to_string());
turn.ended_at = Some(now);
if let Some(started_at) = turn.started_at {
crates/tui/src/runtime_threads/tests.rs
+156
View File
@@ -2485,6 +2485,7 @@ fn sample_turn(thread_id: &str, turn_id: &str, status: RuntimeTurnStatus) -> Tur
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: Vec::new(),
steer_count: 0,
@@ -9461,6 +9462,104 @@ async fn monitor_persists_request_snapshots_and_matching_terminal_diagnostics()
submission_id: None,
})
.await?;
harness
.tx_event
.send(EngineEvent::ToolCallHeartbeat)
.await?;
for (span_id, activity_kind, action_id) in [
(
"outer-private-call",
codewhale_protocol::engine_owner::OwnerActivityKind::Executing,
"execute_tools",
),
(
"nested-private-call",
codewhale_protocol::engine_owner::OwnerActivityKind::Reading,
"read_file",
),
] {
harness
.tx_event
.send(EngineEvent::OperationActivityStarted {
span_id: span_id.to_string(),
activity_kind,
action_id: Some(action_id.to_string()),
})
.await?;
}
harness
.tx_event
.send(EngineEvent::ToolCallHeartbeat)
.await?;
let active_detail = tokio::time::timeout(Duration::from_secs(5), async {
loop {
let detail = manager.get_thread_detail(&thread.id).await?;
if detail.turns.iter().any(|turn| {
turn.operation_activity
.as_ref()
.is_some_and(|activity| activity.active.len() == 2)
}) {
break Ok::<_, anyhow::Error>(detail);
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await??;
let active = active_detail
.turns
.iter()
.find(|turn| turn.id == first.id)
.unwrap()
.operation_activity
.as_ref()
.unwrap();
assert!(!active.overflowed);
assert_eq!(active.active[1].action_id.as_deref(), Some("read_file"));
assert!(
active
.active
.iter()
.all(|span| span.span_id.starts_with("operation:")
&& span.span_id.len() == 74
&& span.started_at <= active.observed_at)
);
assert!(!serde_json::to_string(active)?.contains("private-call"));
assert!(
manager
.events_since(&thread.id, Some(active_detail.latest_seq))?
.iter()
.all(|event| event.event != "operation.activity_started")
);
for (span_id, activity_kind, action_id) in [
(
"nested-private-call",
codewhale_protocol::engine_owner::OwnerActivityKind::Reading,
"read_file",
),
(
"outer-private-call",
codewhale_protocol::engine_owner::OwnerActivityKind::Executing,
"execute_tools",
),
] {
harness
.tx_event
.send(EngineEvent::OperationActivityCompleted {
span_id: span_id.to_string(),
activity_kind,
action_id: Some(action_id.to_string()),
outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome::Succeeded,
})
.await?;
}
harness
.tx_event
.send(EngineEvent::OperationActivityStarted {
span_id: "terminal-interruption".to_string(),
activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind::Tool,
action_id: Some("mcp_example_probe".to_string()),
})
.await?;
let mut tool = catalog_tool("mcp_computer_get_app_state");
tool.description = "Inspect app; api_key=sk-fixture-private-value".to_string();
let pre_request = crate::tool_inspection::ToolInspectionSnapshot::from_prepared_request(
@@ -9537,6 +9636,56 @@ async fn monitor_persists_request_snapshots_and_matching_terminal_diagnostics()
.await?;
let completed = wait_for_terminal_turn(&manager, &first.id).await?;
let operation_activity = completed
.operation_activity
.as_ref()
.expect("admitted operation snapshot");
assert!(operation_activity.active.is_empty());
assert_eq!(
operation_activity
.last_completed
.as_ref()
.unwrap()
.action_id
.as_deref(),
Some("execute_tools")
);
let operations = manager
.events_since(&thread.id, None)?
.into_iter()
.filter(|event| event.event.starts_with("operation."))
.collect::<Vec<_>>();
assert_eq!(
operations
.iter()
.map(|event| event.event.as_str())
.collect::<Vec<_>>(),
[
"operation.activity_started",
"operation.activity_started",
"operation.heartbeat",
"operation.activity_completed",
"operation.activity_completed",
"operation.activity_started"
]
);
assert!(operations.windows(2).all(|pair| pair[0].seq < pair[1].seq));
assert_eq!(
operations[0].payload["span_id"],
operations[4].payload["span_id"]
);
assert_eq!(
operations[1].payload["span_id"],
operations[3].payload["span_id"]
);
assert_eq!(operations[3].payload["outcome"], "succeeded");
assert!(
operations
.iter()
.all(|event| event.turn_id.as_deref() == Some(first.id.as_str())
&& event.payload["observed_at"].is_string())
);
assert!(operations[3].seq > active_detail.latest_seq);
assert_eq!(
completed.model_request_diagnostics,
Some(RuntimeTurnRequestDiagnostics {
@@ -18026,6 +18175,7 @@ fn opening_manager_recovers_stale_queued_and_in_progress_work() -> Result<()> {
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: vec![completed_item.id.clone(), in_progress_item.id.clone()],
steer_count: 0,
@@ -18065,6 +18215,7 @@ fn opening_manager_recovers_stale_queued_and_in_progress_work() -> Result<()> {
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: vec![queued_item.id.clone()],
steer_count: 0,
@@ -18262,6 +18413,7 @@ fn seed_turns_with_user_messages(
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: vec![user_item_id, asst_item_id],
steer_count: 0,
@@ -19082,6 +19234,7 @@ fn restart_rebuild_restores_tool_call_identity_from_persisted_items() -> Result<
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: vec![user_item.id.clone(), call_item.id.clone()],
steer_count: 0,
@@ -19188,6 +19341,7 @@ fn restart_rebuild_keeps_in_flight_tool_call_identity() -> Result<()> {
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: vec![call_item.id.clone()],
steer_count: 0,
@@ -19289,6 +19443,7 @@ fn restart_rebuild_skips_steers_the_engine_never_delivered() -> Result<()> {
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: vec![delivered.id.clone(), dropped.id.clone(), pending.id.clone()],
steer_count: 0,
@@ -19391,6 +19546,7 @@ fn restart_rebuild_skips_legacy_tool_items_without_identity() -> Result<()> {
routed_usage_source_ids: Vec::new(),
routed_usage_dropped_records: 0,
model_request_diagnostics: None,
operation_activity: None,
error: None,
item_ids: vec![user_item.id.clone(), legacy_tool_item.id.clone()],
steer_count: 0,
crates/tui/src/tools/activity.rs
+17 -1
View File
@@ -1,7 +1,7 @@
//! Typed operation activity for the Engine owner contract.
//!
//! The Engine reports *what kind* of work a dispatched call is doing (reading,
//! editing, desktop control, ...) without the tool name, arguments, command or
//! editing, desktop control, ...) with a bounded canonical action identity, without arguments, command or
//! result. Classification reuses the resolvers dispatch already trusts:
//! [`canonical_action_alias`] for action families and the MCP pool's resolved
//! server map for MCP tools. A call that neither resolves to a concrete
@@ -55,6 +55,16 @@ pub(crate) fn registry_activity_kind(tool_name: &str, input: &Value) -> Option<O
Some(kind_for_operation(canonical_action_alias(tool_name, input)))
}
#[must_use]
pub(crate) fn action_id(tool_name: &str, input: &Value) -> String {
let id = canonical_action_alias(tool_name, input);
if !id.is_empty() && id.len() <= 256 && !id.chars().any(char::is_control) {
id.to_owned()
} else {
"tool".to_owned()
}
}
/// Classify an MCP call by the server the pool resolved it to, not by the
/// model-facing name. Desktop-control servers report `Computer`; everything
/// else is a generic `Tool`.
@@ -105,6 +115,12 @@ mod tests {
fn primitives_and_action_families_resolve_through_the_canonical_alias() {
let none = json!({});
assert_eq!(registry_activity_kind("read", &none), Some(Kind::Reading));
assert_eq!(action_id("read", &none), "read_file");
assert_eq!(action_id("Git", &json!({"action": "diff"})), "git_diff");
assert_eq!(action_id("mcp_example_custom", &none), "mcp_example_custom");
for unsafe_id in ["", "bad\nname", &"鲸".repeat(86)] {
assert_eq!(action_id(unsafe_id, &none), "tool");
}
assert_eq!(registry_activity_kind("write", &none), Some(Kind::Editing));
assert_eq!(registry_activity_kind("edit", &none), Some(Kind::Editing));
let cases = [
crates/tui/src/tools/codemode.rs
+53
View File
@@ -761,6 +761,20 @@ impl CodemodeInvoker {
/// session pool (same dispatcher as a direct call), everything else
/// through its registry spec under the turn's authority envelope.
async fn execute(&self, name: &str, input: Value) -> Result<RichToolResult, ToolError> {
let action_id = super::activity::action_id(name, &input);
let activity_kind = if McpPool::is_mcp_tool(name) {
if let Some(pool) = self.gate.as_ref().and_then(|gate| gate.mcp_pool.as_ref()) {
pool.lock()
.await
.resolved_tool_servers()
.get(name)
.map(|server| super::activity::mcp_activity_kind(server))
} else {
None
}
} else {
super::activity::registry_activity_kind(name, &input)
};
let future: std::pin::Pin<
Box<dyn std::future::Future<Output = Result<RichToolResult, ToolError>> + Send>,
> = if McpPool::is_mcp_tool(name) {
@@ -797,12 +811,51 @@ impl CodemodeInvoker {
};
enforce_tool_authority(name, &input, spec.as_ref(), &self.context)?;
let context = self.context.clone();
crate::extension_host::validate_caller_plugins(context.plugin_registry.as_deref())
.map_err(ToolError::not_available)?;
Box::pin(async move {
crate::extension_host::validate_caller_plugins(context.plugin_registry.as_deref())
.map_err(ToolError::not_available)?;
spec.execute_rich(input, &context).await
})
};
let tx = self.gate.as_ref().map(|gate| gate.tx_event.clone());
let cancel = self.context.cancel_token.clone();
let call_id = self
.context
.origin_tool_call_id
.clone()
.unwrap_or_else(|| self.spill_prefix.clone());
let future = async move {
if cancel.as_ref().is_some_and(CancellationToken::is_cancelled) {
return Err(ToolError::cancelled("Nested dispatch admission cancelled."));
}
let span = match (tx.as_ref(), activity_kind) {
(Some(tx), Some(kind)) => Some(
crate::core::engine::tool_execution::OperationSpanGuard::start(
tx,
&call_id,
kind,
Some(action_id),
cancel.clone(),
)
.await,
),
_ => None,
};
if cancel.as_ref().is_some_and(CancellationToken::is_cancelled) {
return Err(ToolError::cancelled("Nested dispatch admission cancelled."));
}
let outcome = future.await;
if let Some(span) = span {
span.complete(super::activity::operation_outcome(
&outcome,
cancel.as_ref().is_some_and(CancellationToken::is_cancelled),
))
.await;
}
outcome
};
match self.runtime.as_ref() {
Some(runtime) => {
let mut task = AbortOnDrop(runtime.spawn(future));
crates/tui/src/tui/pet_watch/avatars.rs
+4 -2
View File
@@ -119,13 +119,15 @@ impl Avatars {
}
pub fn choices(&self) -> String {
std::iter::once("whale".to_string())
.chain(std::iter::once("whale-girl".to_string()))
.chain(["particles".to_string(), "whale-girl".to_string()])
.chain(self.catalog.iter().map(|p| p.key.clone()))
.collect::<Vec<_>>()
.join("\n")
}
pub fn select(&mut self, key: &str) -> bool {
if !matches!(key, "whale" | "whale-girl") && !self.catalog.iter().any(|p| p.key == key) {
if !matches!(key, "whale" | "whale-girl" | "particles")
&& !self.catalog.iter().any(|p| p.key == key)
{
return false;
}
self.key = key.into();
crates/tui/src/tui/pet_watch/graphics.rs
+47 -6
View File
@@ -12,6 +12,7 @@ pub fn clear(output: &mut impl Write) -> io::Result<()> {
#[derive(Default)]
pub struct Renderer {
pub colors: Vec<[u8; 3]>,
dimensions: (usize, usize),
background: Vec<u8>,
appearance: super::appearance::Appearance,
@@ -74,13 +75,23 @@ impl Renderer {
Vec::new()
};
let mut cells = vec![0u8; cols * rows];
self.colors = vec![self.appearance.background; cols * rows];
let mut strengths = vec![0.0; cols * rows];
let val = |name: &str, default: f64| pose.state[name].as_f64().unwrap_or(default);
let attention = val("attention", 0.0);
let attention = if pose.materials.is_some() {
0.0
} else {
val("attention", 0.0)
};
let flip = val("flip", 1.0);
let scale = (w as f64 * 0.52).min(h as f64 * 0.85) * (1.0 + attention * 0.07);
let scale = if pose.materials.is_some() {
w.min(h) as f64 * 0.46
} else {
(w as f64 * 0.52).min(h as f64 * 0.85) * (1.0 + attention * 0.07)
};
let ox = w as f64 * (0.5 + val("roamX", 0.0) * 0.30);
let oy = h as f64 * (0.47 + val("roamY", 0.0) * 0.30 + attention * 0.05);
let colour = [pose.style.r, pose.style.g, pose.style.b];
if view.pixels && self.appearance.environment {
for x in 16..w.saturating_sub(16) {
blend(&mut rgb, w, h, x as i32, 16, [45.0, 70.0, 81.0], 0.5);
@@ -104,6 +115,28 @@ impl Renderer {
}
let radius = (w.min(h) as f64 * 0.00285).clamp(0.68, 1.55) * self.appearance.dot_scale;
for (i, q) in pose.points.iter().enumerate() {
let depth = 0.65 + 0.35 * ((i * 37 % 101) as f64 / 100.0);
let material = pose.materials.as_ref().and_then(|m| m.get(i));
let fallback = [
pose.style.r,
pose.style.g,
pose.style.b,
pose.style.alpha * depth,
];
let mut pigment = *material.unwrap_or(&fallback);
if let Some(old) = previous
.and_then(|s| s.materials.as_ref())
.and_then(|m| m.get(i))
{
for k in 0..4 {
pigment[k] = old[k] + (pigment[k] - old[k]) * mix;
}
}
let colour = [pigment[0], pigment[1], pigment[2]];
let alpha = pigment[3].clamp(0.0, 1.0);
if alpha <= 0.0 {
continue;
}
let p = previous.and_then(|s| s.points.get(i));
let xx = p.map_or(q[0], |p| p[0] + (q[0] - p[0]) * mix);
let yy = p.map_or(q[1], |p| p[1] + (q[1] - p[1]) * mix);
@@ -113,13 +146,20 @@ impl Renderer {
let cy = (y / h as f64 * (rows * 4) as f64).floor() as i32;
if cx >= 0 && cy >= 0 && cx < (cols * 2) as i32 && cy < (rows * 4) as i32 {
let bit = [[0, 3], [1, 4], [2, 5], [6, 7]][cy as usize % 4][cx as usize % 2];
cells[(cy as usize / 4) * cols + cx as usize / 2] |= 1 << bit;
let index = (cy as usize / 4) * cols + cx as usize / 2;
cells[index] |= 1 << bit;
if alpha > strengths[index] {
strengths[index] = alpha;
for k in 0..3 {
self.colors[index][k] = (colour[k] * alpha
+ f64::from(self.appearance.background[k]) * (1.0 - alpha))
as u8;
}
}
}
if !view.pixels {
continue;
}
let depth = 0.65 + 0.35 * ((i * 37 % 101) as f64 / 100.0);
let alpha = (pose.style.alpha * depth).clamp(0.0, 1.0);
let r = radius * depth;
let glow = if !pose.style.hollow && i % 5 == 0 {
1.0 + 4.0 * self.appearance.glow
@@ -196,6 +236,7 @@ mod tests {
fn pixel_transport_is_bounded_chunked_rgb_and_braille_remains_available() {
let pose = Pose {
points: vec![[0.0, 0.0]; 980],
materials: None,
style: Style {
r: 120.,
g: 210.,
crates/tui/src/tui/pet_watch/live.rs
+19
View File
@@ -24,6 +24,8 @@ pub struct Style {
#[derive(Clone, Deserialize, Serialize)]
pub struct Pose {
pub points: Vec<[f64; 2]>,
#[serde(default)]
pub materials: Option<Vec<[f64; 4]>>,
pub style: Style,
pub state: Value,
}
@@ -46,6 +48,8 @@ pub struct Scene {
pub audio_owner: Option<String>,
pub audio_unavailable: bool,
pub points: Vec<[f64; 2]>,
#[serde(default)]
pub materials: Option<Vec<[f64; 4]>>,
pub style: Style,
pub state: Value,
pub still: Pose,
@@ -83,6 +87,18 @@ impl Scene {
&& self.time_ms.is_finite()
&& self.points.len() == 980
&& self.still.points.len() == 980
&& [&self.materials, &self.still.materials]
.into_iter()
.all(|materials| {
materials.as_ref().is_none_or(|values| {
values.len() == 980
&& values.iter().all(|m| {
m.iter().all(|v| v.is_finite())
&& m[..3].iter().all(|v| (0.0..=255.0).contains(v))
&& (0.0..=1.0).contains(&m[3])
})
})
})
&& self
.points
.iter()
@@ -96,6 +112,7 @@ pub struct Presentation {
pub client: String,
pub scene: Scene,
pub cells: Vec<u8>,
pub colors: Vec<[u8; 3]>,
pub image: Option<Vec<u8>>,
pub width: u16,
pub height: u16,
@@ -591,6 +608,7 @@ fn run(
let mut pose = if view.motion {
Pose {
points: s.points.clone(),
materials: s.materials.clone(),
style: s.style.clone(),
state: s.state.clone(),
}
@@ -619,6 +637,7 @@ fn run(
client: id.clone(),
scene: s.clone(),
cells,
colors: renderer.colors.clone(),
image,
width: view.width,
height: view.height,
crates/tui/src/tui/pet_watch/mod.rs
+36 -3
View File
@@ -259,7 +259,7 @@ impl PetWatch {
/// Existing protocol projection defines the variant names. This allowlist
/// removes payloads before the bounded worker queue; no model text escapes.
fn metadata(event: &Event, turn_id: Option<&str>) -> Option<String> {
let value = match event {
let mut value = match event {
Event::TurnStarted { turn_id, .. } => json!({"event":"turn_started","turn_id":turn_id}),
Event::TurnComplete { status, .. } => {
let mut value = json!({
@@ -290,18 +290,22 @@ fn metadata(event: &Event, turn_id: Option<&str>) -> Option<String> {
Event::OperationActivityStarted {
span_id,
activity_kind,
action_id,
} => json!({
"event":"operation_activity_started",
"span_id":span_id,
"action_id":action_id,
"activity_kind":activity_kind
}),
Event::OperationActivityCompleted {
span_id,
activity_kind,
action_id,
outcome,
} => json!({
"event":"operation_activity_completed",
"span_id":span_id,
"action_id":action_id,
"activity_kind":activity_kind,
"outcome":outcome
}),
@@ -338,6 +342,12 @@ fn metadata(event: &Event, turn_id: Option<&str>) -> Option<String> {
}),
_ => return None,
};
if value
.get("action_id")
.is_some_and(serde_json::Value::is_null)
{
value.as_object_mut()?.remove("action_id");
}
let json = serde_json::to_string(&value).ok()?;
(json.len() <= 16_384).then_some(json)
}
@@ -590,7 +600,9 @@ fn render_tank(frame: &mut Frame, area: Rect, app: &mut App) {
&& activity.freshness == codewhale_protocol::engine_owner::OwnerFreshness::Fresh
&& r.frame_changed.elapsed().as_millis() < 800
{
if let Some(kind) = activity.activity_kind {
if let Some(action) = activity.action_id.as_deref() {
text = format!("{action} · {text}");
} else if let Some(kind) = activity.activity_kind {
text = format!("{} · {}", kind.as_str(), text);
}
if activity.parallel_agent_count > 0 {
@@ -691,6 +703,23 @@ fn render_tank(frame: &mut Frame, area: Rect, app: &mut App) {
&label,
Style::default().fg(ink),
);
if let Some(raster) = raster.filter(|r| r.width == area.width && r.height == tank.height)
&& area.height >= 4
&& unicode_width::UnicodeWidthStr::width(label.as_str()) <= usize::from(area.width)
{
for y in tank.y..tank.bottom() {
for x in tank.x..tank.right() {
let index =
usize::from(y - tank.y) * usize::from(tank.width) + usize::from(x - tank.x);
if raster.cells.get(index).is_some_and(|bits| *bits != 0)
&& let Some(rgb) = raster.colors.get(index)
&& let Some(cell) = frame.buffer_mut().cell_mut((x, y))
{
cell.set_fg(Color::Rgb(rgb[0], rgb[1], rgb[2]));
}
}
}
}
if raster.is_none() {
paint_resting_whale(frame, area, Style::default().fg(ink));
}
@@ -1023,6 +1052,7 @@ mod tests {
&Event::OperationActivityStarted {
span_id: "private-internal-span".into(),
activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind::Computer,
action_id: Some("mcp_computer_use_click".into()),
},
None,
)
@@ -1032,13 +1062,15 @@ mod tests {
json!({
"event":"operation_activity_started",
"span_id":"private-internal-span",
"activity_kind":"computer"
"activity_kind":"computer",
"action_id":"mcp_computer_use_click"
})
);
let completed = metadata(
&Event::OperationActivityCompleted {
span_id: "private-internal-span".into(),
activity_kind: codewhale_protocol::engine_owner::OwnerActivityKind::Editing,
action_id: Some("edit_file".into()),
outcome: codewhale_protocol::engine_owner::OwnerOperationOutcome::Denied,
},
None,
@@ -1050,6 +1082,7 @@ mod tests {
"event":"operation_activity_completed",
"span_id":"private-internal-span",
"activity_kind":"editing",
"action_id":"edit_file",
"outcome":"denied"
})
);
crates/tui/src/tui/pet_watch/owner.rs
+31
View File
@@ -601,6 +601,37 @@ fn run_world(
pose["style"][key] = json!(value);
}
}
let hollow = pose["style"]["hollow"].as_bool().unwrap_or(true);
let lit = pose["state"]["lit"]
.as_f64()
.unwrap_or(1.0)
.clamp(0.18, 1.0);
let light = saved
.appearance
.background
.iter()
.map(|v| f64::from(*v))
.sum::<f64>()
> 510.0;
if let Some(materials) = pose["materials"].as_array_mut() {
for material in materials {
if let Some(channels) = material.as_array_mut().filter(|c| c.len() == 4) {
for k in 0..3 {
let value = if hollow {
[153., 176., 184.][k]
} else if !saved.appearance.event_colors {
f64::from(saved.appearance.particle[k])
} else {
channels[k].as_f64().unwrap_or(160.0)
};
channels[k] = json!(if light { value * 0.48 } else { value });
}
let alpha = channels[3].as_f64().unwrap_or(0.5);
channels[3] =
json!((alpha * saved.appearance.brightness * lit).clamp(0.0, 1.0));
}
}
}
pose["style"]["alpha"] = json!(
(pose["style"]["alpha"].as_f64().unwrap_or(0.5) * saved.appearance.brightness)
.clamp(0.0, 1.0)