use crate::output::{ActivityEvent, ActivityKind, ActivityStatus, ContextUsageSource, OutputEvent};
use serde::Serialize;
use sha2::{Digest, Sha256};
pub(crate) const ACTIVITY_EVENT: &str = "turn.activity";
pub(crate) const MAX_SUMMARY_BYTES: usize = 2048;
#[derive(Debug, Serialize)]
pub(crate) struct ActivityDto {
pub(crate) activity_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
parent_activity_id: Option<String>,
#[serde(flatten)]
data: ActivityData,
}
#[derive(Debug, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum ActivityData {
Lifecycle {
category: &'static str,
status: &'static str,
details_omitted: bool,
},
Tool {
tool_id: String,
tool_name: Option<&'static str>,
server_id: Option<String>,
transport: &'static str,
status: &'static str,
timed_out: Option<bool>,
details_omitted: bool,
},
ReasoningSummary {
text: String,
redacted: bool,
truncated: bool,
},
Context {
current_tokens: usize,
max_tokens: usize,
reasoning_tokens: Option<usize>,
source: &'static str,
request_sequence: u64,
},
Usage {
effective_input: u64,
output: u64,
cache_read: u64,
cache_known: bool,
request_sequence: u64,
final_usage: bool,
source: &'static str,
},
Compaction {
status: &'static str,
current_tokens: Option<usize>,
max_tokens: Option<usize>,
replay_required: bool,
},
}
fn identity(namespace: &str, value: &str) -> String {
format!(
"{namespace}-{}",
crate::hex::lower_hex(Sha256::digest(value.as_bytes()))
)
}
fn dto(id: &str, data: ActivityData) -> ActivityDto {
ActivityDto {
activity_id: identity("activity", id),
parent_activity_id: None,
data,
}
}
fn status(value: ActivityStatus) -> &'static str {
match value {
ActivityStatus::Queued => "queued",
ActivityStatus::Running => "running",
ActivityStatus::Writing => "writing",
ActivityStatus::Success => "completed",
ActivityStatus::Failed => "failed",
ActivityStatus::Canceled => "cancelled",
}
}
fn source(value: ContextUsageSource) -> &'static str {
match value {
ContextUsageSource::FallbackEstimate => "fallback_estimate",
ContextUsageSource::TokenizerEstimate => "tokenizer_estimate",
ContextUsageSource::FallbackProjection => "fallback_projection",
ContextUsageSource::TokenizerProjection => "tokenizer_projection",
ContextUsageSource::LastProviderUsage => "last_provider_usage",
ContextUsageSource::ProviderExact => "provider_exact",
ContextUsageSource::ProviderPartial => "provider_partial",
}
}
pub(crate) fn activity(event: ActivityEvent) -> Option<ActivityDto> {
match event {
ActivityEvent::Started {
id,
parent_id,
kind,
status: state,
..
} => {
let category = match kind {
ActivityKind::Tool => "tool",
ActivityKind::Hook => "hook",
ActivityKind::ProviderContextInjection => "context_injection",
ActivityKind::SubagentBatch => "subagent_batch",
ActivityKind::SubagentTask => "subagent_task",
ActivityKind::Assistant => "assistant",
ActivityKind::Diagnostic => "diagnostic",
ActivityKind::Compaction => "compaction",
};
let mut mapped = dto(
id.as_str(),
ActivityData::Lifecycle {
category,
status: status(state),
details_omitted: true,
},
);
mapped.parent_activity_id = parent_id.map(|id| identity("activity", id.as_str()));
Some(mapped)
}
ActivityEvent::Finished {
id, status: state, ..
} => Some(dto(
id.as_str(),
ActivityData::Lifecycle {
category: "activity",
status: status(state),
details_omitted: true,
},
)),
ActivityEvent::ToolStartedDetail { id, detail }
| ActivityEvent::ToolResultDetail { id, detail } => {
Some(tool(id.as_str(), &detail.tool_name, status(detail.status)))
}
ActivityEvent::UsageUpdate {
id,
current_tokens,
max_tokens,
reasoning_tokens,
source: origin,
request_sequence,
} => Some(dto(
id.as_str(),
ActivityData::Context {
current_tokens,
max_tokens,
reasoning_tokens,
source: source(origin),
request_sequence,
},
)),
ActivityEvent::UsageSnapshot {
id,
usage,
request_sequence,
final_usage,
} => {
let usage = usage.whole_run;
Some(dto(
id.as_str(),
ActivityData::Usage {
effective_input: usage.effective_input,
output: usage.output,
cache_read: usage.cache_read,
cache_known: usage.cache_known,
request_sequence,
final_usage,
source: "subagent_aggregate",
},
))
}
ActivityEvent::Delta { .. }
| ActivityEvent::FinalPreview { .. }
| ActivityEvent::FastObservation { .. } => None,
}
}
fn builtin_tool_name(name: &str) -> Option<&'static str> {
use crate::tools::contract::tool_name::*;
[
READ, VIEW_IMAGE, BASH, HASH_EDIT, WRITE, GREP, FIND, LIST_FILES, SUBAGENTS, WEB, AST_GREP,
]
.into_iter()
.find(|known| *known == name)
}
fn tool(id: &str, name: &str, status: &'static str) -> ActivityDto {
let mcp = (name.len() <= crate::mcp::names::MCP_QUALIFIED_NAME_MAX_BYTES)
.then(|| crate::mcp::QualifiedMcpToolName::parse(name).ok())
.flatten();
dto(
id,
ActivityData::Tool {
tool_id: identity("tool", name),
tool_name: builtin_tool_name(name),
server_id: mcp.as_ref().map(|route| identity("server", route.server())),
transport: if mcp.is_some() {
"mcp"
} else if builtin_tool_name(name).is_some() {
"local"
} else {
"unknown"
},
status,
timed_out: None,
details_omitted: true,
},
)
}
pub(crate) fn provider_summary(text: &str) -> ActivityDto {
let data = if text.len() > super::protocol::MAX_STRING_BYTES {
ActivityData::ReasoningSummary {
text: String::new(),
redacted: false,
truncated: true,
}
} else {
let mut safe = crate::output::sanitize_display_text(text);
let redacted = safe != text;
let truncated = safe.len() > MAX_SUMMARY_BYTES;
if truncated {
let mut end = MAX_SUMMARY_BYTES;
while !safe.is_char_boundary(end) {
end -= 1;
}
safe.truncate(end);
}
ActivityData::ReasoningSummary {
text: safe,
redacted,
truncated,
}
};
dto("reasoning", data)
}
pub(crate) fn output(event: OutputEvent) -> Option<ActivityDto> {
let data = match event {
OutputEvent::ToolStarted { call, .. } => {
return Some(tool(&call.id, &call.name, "running"));
}
OutputEvent::ToolResult { call, result, .. } => {
let mut mapped = tool(
&call.id,
&call.name,
if result.success {
"completed"
} else {
"failed"
},
);
if let ActivityData::Tool { timed_out, .. } = &mut mapped.data {
*timed_out = result
.metadata
.get(crate::tools::contract::metadata_key::TIMED_OUT)
.and_then(serde_json::Value::as_bool);
}
return Some(mapped);
}
OutputEvent::ContextUsage {
current_tokens,
max_tokens,
reasoning_tokens,
source: origin,
request_sequence,
} => ActivityData::Context {
current_tokens,
max_tokens,
reasoning_tokens,
source: source(origin),
request_sequence,
},
OutputEvent::UsageSnapshot {
usage,
request_sequence,
final_usage,
} => ActivityData::Usage {
effective_input: usage.effective_input,
output: usage.output,
cache_read: usage.cache_read,
cache_known: usage.cache_known,
request_sequence,
final_usage,
source: "provider_request",
},
OutputEvent::CompactionTriggered {
current_tokens,
max_tokens,
..
} => ActivityData::Compaction {
status: "triggered",
current_tokens: Some(current_tokens),
max_tokens: Some(max_tokens),
replay_required: false,
},
OutputEvent::CompactionStarted => ActivityData::Compaction {
status: "running",
current_tokens: None,
max_tokens: None,
replay_required: false,
},
OutputEvent::CompactionCompleted {
current_tokens,
max_tokens,
..
} => ActivityData::Compaction {
status: "completed",
current_tokens: Some(current_tokens),
max_tokens: Some(max_tokens),
replay_required: true,
},
OutputEvent::CompactionFailed { canceled, .. } => ActivityData::Compaction {
status: if canceled { "cancelled" } else { "failed" },
current_tokens: None,
max_tokens: None,
replay_required: true,
},
_ => return None,
};
let id = match &data {
ActivityData::ReasoningSummary { .. } => "reasoning",
ActivityData::Context { .. } | ActivityData::Usage { .. } => "usage",
ActivityData::Compaction { .. } => "compaction",
_ => unreachable!(),
};
Some(ActivityDto {
activity_id: identity("stream", id),
parent_activity_id: None,
data,
})
}
#[derive(Debug, Serialize)]
pub(crate) struct ActivityCapabilities {
events: bool,
approvals: bool,
steering: bool,
mcp_server_lifecycle: bool,
mcp_tool_activity: bool,
subagents: bool,
raw_tool_output: bool,
activity_replay: bool,
max_summary_bytes: usize,
}
impl ActivityCapabilities {
pub(crate) fn current() -> Self {
Self {
events: true,
approvals: false,
steering: false,
mcp_server_lifecycle: false,
mcp_tool_activity: true,
subagents: true,
raw_tool_output: false,
activity_replay: false,
max_summary_bytes: MAX_SUMMARY_BYTES,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::output::{ActivityId, ActivityMetadata, NormalizedUsageSnapshot};
use serde_json::{Value, json};
fn wire(event: OutputEvent) -> Value {
serde_json::to_value(output(event).unwrap()).unwrap()
}
#[test]
fn tool_projection_correlates_without_commands_results_or_untrusted_names() {
let call = crate::providers::ToolCall {
id: "secret-call-id".into(),
name: "mcp__secret-server__secret-tool".into(),
arguments: json!({"command": "secret-command", "Authorization": "secret-header"}),
};
let start = wire(OutputEvent::ToolStarted {
call: Box::new(call.clone()),
label: "secret-label".into(),
});
let result = crate::tools::ToolResult {
tool_name: call.name.clone(),
success: false,
content: "secret-output".repeat(10000),
metadata: json!({"timed_out": true, "stderr": "secret-stderr"}),
display: Default::default(),
};
let summary = crate::output::tool_display_summary(&call, &result);
let finish = wire(OutputEvent::ToolResult {
changed_paths: Vec::new(),
call: Box::new(call),
result: Box::new(result),
summary: Box::new(summary),
});
assert_eq!(start["activity_id"], finish["activity_id"]);
assert_eq!(start["tool_id"], finish["tool_id"]);
assert_eq!(finish["transport"], "mcp");
assert_eq!(finish["server_id"], identity("server", "secret-server"));
assert_eq!(finish["status"], "failed");
assert_eq!(finish["timed_out"], true);
assert_eq!(finish["details_omitted"], true);
let encoded = serde_json::to_string(&json!([start, finish])).unwrap();
assert!(!encoded.contains("secret"));
assert!(encoded.len() < 1024);
}
#[test]
fn child_lifecycle_has_safe_parent_and_cancellation_identity() {
let start = activity(ActivityEvent::Started {
id: ActivityId::new("parent/secret-child"),
parent_id: Some(ActivityId::new("parent")),
kind: ActivityKind::SubagentTask,
status: ActivityStatus::Queued,
metadata: ActivityMetadata::new("secret-prompt"),
})
.unwrap();
let finish = activity(ActivityEvent::Finished {
id: ActivityId::new("parent/secret-child"),
status: ActivityStatus::Canceled,
metadata: None,
})
.unwrap();
let start = serde_json::to_value(start).unwrap();
let finish = serde_json::to_value(finish).unwrap();
assert_eq!(start["activity_id"], finish["activity_id"]);
assert_eq!(start["parent_activity_id"], identity("activity", "parent"));
assert_eq!(start["category"], "subagent_task");
assert_eq!(finish["status"], "cancelled");
assert!(!start.to_string().contains("secret"));
assert!(
activity(ActivityEvent::Delta {
id: ActivityId::new("child"),
preview: "secret-child-output".into()
})
.is_none()
);
}
#[test]
fn complete_reasoning_is_sanitized_and_bounded_not_streamed() {
assert!(
output(OutputEvent::ThinkingSummaryDelta {
text: "Bearer partial".into()
})
.is_none()
);
for event in [
OutputEvent::ThinkingSummaryComplete {
text: "raw thinking".into(),
},
OutputEvent::ThinkingSummaryCompleteIdentified {
text: "raw thinking with a generated identity".into(),
item_id: Some("legacy-1-1".into()),
turn_id: Some("1".into()),
},
] {
assert!(output(event).is_none());
}
let summary = serde_json::to_value(provider_summary(&format!(
"\u{1b}[31mBearer abcdefghijklmnopqrstuvwxyz\n{}",
"é".repeat(2048)
)))
.unwrap();
assert_eq!(summary["redacted"], true);
assert_eq!(summary["truncated"], true);
let text = summary["text"].as_str().unwrap();
assert!(!text.contains("abcdefghijklmnopqrstuvwxyz"));
assert!(!text.contains('\u{1b}'));
assert!(text.len() <= MAX_SUMMARY_BYTES);
let oversized = serde_json::to_value(provider_summary(
&"x".repeat(super::super::protocol::MAX_STRING_BYTES + 1),
))
.unwrap();
assert_eq!(oversized["text"], "");
assert_eq!(oversized["truncated"], true);
}
#[test]
fn usage_and_compaction_project_sources_and_reconciliation_without_raw_summary() {
let context = wire(OutputEvent::ContextUsage {
current_tokens: 100,
max_tokens: 200,
reasoning_tokens: Some(12),
source: ContextUsageSource::ProviderPartial,
request_sequence: 3,
});
assert_eq!(context["source"], "provider_partial");
assert_eq!(context["reasoning_tokens"], 12);
let usage = wire(OutputEvent::UsageSnapshot {
usage: NormalizedUsageSnapshot {
effective_input: 100,
output: 30,
cache_read: 20,
cache_known: true,
},
request_sequence: 3,
final_usage: true,
});
assert_eq!(usage["effective_input"], 100);
assert_eq!(usage["source"], "provider_request");
let running = wire(OutputEvent::CompactionStarted);
let complete = wire(OutputEvent::CompactionCompleted {
current_tokens: 50,
max_tokens: 200,
summary: "secret-compaction-summary".into(),
});
let failed = wire(OutputEvent::CompactionFailed {
message: "secret-error".into(),
canceled: true,
});
assert_eq!(running["activity_id"], complete["activity_id"]);
assert_eq!(complete["replay_required"], true);
assert_eq!(failed["status"], "cancelled");
assert!(!json!([complete, failed]).to_string().contains("secret"));
}
}