huskies: merge 1209 story Gateway lifecycle & telemetry MCP: gateway_info, restart_gateway, gateway_logs, start_story, chat_telemetry
This commit is contained in:
@@ -33,6 +33,12 @@ const GATEWAY_TOOLS: &[&str] = &[
|
||||
"fleet_resources",
|
||||
// Read sled identity pins vs. live signed identity, and TOFU re-pin.
|
||||
"fleet_identity",
|
||||
// Gateway process pid/build/version/uptime/configured model (story 1209).
|
||||
"gateway_info",
|
||||
// Bounce the gateway process itself, never touching project containers (story 1209).
|
||||
"restart_gateway",
|
||||
// Tail/grep the gateway's own in-process log, distinct from sled get_server_logs (story 1209).
|
||||
"gateway_logs",
|
||||
];
|
||||
|
||||
/// Gateway tool definitions.
|
||||
@@ -201,6 +207,39 @@ pub(crate) fn gateway_tool_definitions() -> Vec<Value> {
|
||||
}
|
||||
}
|
||||
}),
|
||||
json!({
|
||||
"name": "gateway_info",
|
||||
"description": "Return the running gateway process's pid, build hash, version, uptime in seconds, and configured Claude Code model (from the gateway's own bot.toml).",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {}
|
||||
}
|
||||
}),
|
||||
json!({
|
||||
"name": "restart_gateway",
|
||||
"description": "Safely bounce the gateway process itself (flush persisted state, then exit so Docker's restart policy relaunches the container). Never touches or removes any registered project's container — use project_rebuild for that.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {}
|
||||
}
|
||||
}),
|
||||
json!({
|
||||
"name": "gateway_logs",
|
||||
"description": "Tail/grep the gateway process's own in-process log (matrix bot, sled-uplink, poller activity) — distinct from get_server_logs, which (when proxied through the gateway) returns the active project's own log instead.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"lines": {
|
||||
"type": "integer",
|
||||
"description": "Number of lines to return (default 100, max 1000)."
|
||||
},
|
||||
"filter": {
|
||||
"type": "string",
|
||||
"description": "Optional substring filter applied to each log line (e.g. 'matrix', 'uplink', 'poller')."
|
||||
}
|
||||
}
|
||||
}
|
||||
}),
|
||||
]
|
||||
}
|
||||
|
||||
@@ -536,6 +575,9 @@ async fn handle_gateway_tool(
|
||||
"project_rebuild" => handle_project_rebuild_tool(params, state, id).await,
|
||||
"fleet_resources" => handle_fleet_resources_tool(params, state, id).await,
|
||||
"fleet_identity" => handle_fleet_identity_tool(params, state, id).await,
|
||||
"gateway_info" => handle_gateway_info_tool(state, id).await,
|
||||
"restart_gateway" => handle_restart_gateway_tool(state, id).await,
|
||||
"gateway_logs" => handle_gateway_logs_tool(params, id),
|
||||
_ => JsonRpcResponse::error(id, -32601, format!("Unknown gateway tool: {tool_name}")),
|
||||
}
|
||||
}
|
||||
@@ -1158,6 +1200,92 @@ async fn handle_fleet_identity_tool(
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle the `gateway_info` gateway tool (story 1209).
|
||||
///
|
||||
/// Returns the running gateway process's pid, build hash, version, uptime,
|
||||
/// and configured Claude Code model — read from the gateway's own `bot.toml`,
|
||||
/// not any registered project's.
|
||||
async fn handle_gateway_info_tool(state: &GatewayState, id: Option<Value>) -> JsonRpcResponse {
|
||||
let build_hash = option_env!("BUILD_GIT_HASH").unwrap_or("unknown");
|
||||
let fields = gateway::io::read_bot_config_raw(&state.config_dir);
|
||||
let info = json!({
|
||||
"pid": std::process::id(),
|
||||
"version": env!("CARGO_PKG_VERSION"),
|
||||
"build_hash": build_hash,
|
||||
"uptime_secs": gateway::gateway_uptime_secs(),
|
||||
"configured_model": fields.model,
|
||||
});
|
||||
JsonRpcResponse::success(
|
||||
id,
|
||||
json!({
|
||||
"content": [{
|
||||
"type": "text",
|
||||
"text": serde_json::to_string_pretty(&info).unwrap_or_default()
|
||||
}],
|
||||
"info": info,
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
/// Handle the `restart_gateway` gateway tool (story 1209).
|
||||
///
|
||||
/// Notifies any bot channels that the gateway is going offline for a
|
||||
/// restart, then bounces the gateway process itself: flushes persisted
|
||||
/// state and exits so Docker's restart policy relaunches the container.
|
||||
/// Deliberately touches only the gateway's own state — `state.projects`
|
||||
/// (and therefore any project container) is never read or written here, so
|
||||
/// no project container is ever removed or swapped by this tool.
|
||||
///
|
||||
/// The actual exit is spawned in the background so the JSON-RPC response
|
||||
/// reaches the caller first; consequently, like `upgrade_and_reexec`, the
|
||||
/// exit itself is not unit-tested (`std::process::exit` would kill the test
|
||||
/// process) — only the notify step and response shape are covered.
|
||||
async fn handle_restart_gateway_tool(state: &GatewayState, id: Option<Value>) -> JsonRpcResponse {
|
||||
if let Some(tx) = state.bot_shutdown_tx.lock().await.as_ref() {
|
||||
let _ = tx.send(Some(crate::rebuild::ShutdownReason::Rebuild));
|
||||
}
|
||||
|
||||
let config_dir = state.config_dir.clone();
|
||||
tokio::spawn(async move {
|
||||
// Give the bot task a moment to post its "going offline" message
|
||||
// before the process flushes state and exits.
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
crate::rebuild::drain_and_exit(&config_dir, "restart_gateway").await
|
||||
});
|
||||
|
||||
JsonRpcResponse::success(
|
||||
id,
|
||||
json!({
|
||||
"content": [{
|
||||
"type": "text",
|
||||
"text": "Gateway restart triggered. The gateway container will restart momentarily; no project containers are affected."
|
||||
}]
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
/// Handle the `gateway_logs` gateway tool (story 1209).
|
||||
///
|
||||
/// Reads directly from the gateway process's own in-process log ring buffer
|
||||
/// (`log_buffer::global()`) — the same buffer the gateway's own `slog!`
|
||||
/// calls (matrix bot, sled-uplink, poller) write to. This is distinct from
|
||||
/// `get_server_logs`, which is not a gateway tool and therefore proxies to
|
||||
/// the active project's own log buffer instead.
|
||||
fn handle_gateway_logs_tool(params: &Value, id: Option<Value>) -> JsonRpcResponse {
|
||||
let args = params.get("arguments").unwrap_or(params);
|
||||
let lines_count = args
|
||||
.get("lines")
|
||||
.and_then(|v| v.as_u64())
|
||||
.map(|n| n.min(1000) as usize)
|
||||
.unwrap_or(100);
|
||||
let filter = args.get("filter").and_then(|v| v.as_str());
|
||||
|
||||
let recent = crate::log_buffer::global().get_recent(lines_count, filter, None);
|
||||
let text = recent.join("\n");
|
||||
|
||||
JsonRpcResponse::success(id, json!({ "content": [{ "type": "text", "text": text }] }))
|
||||
}
|
||||
|
||||
/// Handle the `pipeline.get` read-RPC — returns per-project item lists in the
|
||||
/// shape expected by the gateway web UI:
|
||||
/// `{ "active": "...", "projects": { "name": { "active": [...], "backlog_count": N } } }`.
|
||||
@@ -1403,6 +1531,89 @@ mod tests {
|
||||
assert!(GATEWAY_TOOLS.contains(&"fleet_identity"));
|
||||
}
|
||||
|
||||
// ── gateway_info / restart_gateway / gateway_logs (story 1209) ──────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn gateway_info_tool_returns_pid_version_and_uptime() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let state = make_test_state(dir.path());
|
||||
let resp = handle_gateway_info_tool(&state, Some(json!(1))).await;
|
||||
assert!(resp.error.is_none(), "expected success: {:?}", resp.error);
|
||||
let info = &resp.result.unwrap()["info"];
|
||||
assert_eq!(info["pid"], std::process::id());
|
||||
assert_eq!(info["version"], env!("CARGO_PKG_VERSION"));
|
||||
assert!(info["uptime_secs"].is_u64());
|
||||
assert!(info["build_hash"].is_string());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn gateway_info_tool_reports_configured_model_from_bot_toml() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let huskies_dir = dir.path().join(".huskies");
|
||||
std::fs::create_dir_all(&huskies_dir).unwrap();
|
||||
std::fs::write(
|
||||
huskies_dir.join("bot.toml"),
|
||||
"model = \"claude-opus-4-8\"\n",
|
||||
)
|
||||
.unwrap();
|
||||
let state = make_test_state(dir.path());
|
||||
let resp = handle_gateway_info_tool(&state, Some(json!(1))).await;
|
||||
let info = &resp.result.unwrap()["info"];
|
||||
assert_eq!(info["configured_model"], "claude-opus-4-8");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn restart_gateway_is_registered_as_a_gateway_tool() {
|
||||
// The handler triggers `std::process::exit` on a background task after
|
||||
// a short delay — it is deliberately never invoked from a test (doing
|
||||
// so risks terminating the entire test binary if the spawned task
|
||||
// outlives the test's runtime). Only registration/schema wiring is
|
||||
// verified here, matching the existing lack of coverage for
|
||||
// `upgrade_and_reexec` (same shape, same reason).
|
||||
assert!(GATEWAY_TOOLS.contains(&"restart_gateway"));
|
||||
let defs = gateway_tool_definitions();
|
||||
assert!(
|
||||
defs.iter().any(|d| d["name"] == "restart_gateway"),
|
||||
"restart_gateway must appear in gateway_tool_definitions()"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gateway_logs_is_in_gateway_tools() {
|
||||
assert!(GATEWAY_TOOLS.contains(&"gateway_logs"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gateway_logs_tool_returns_recent_lines() {
|
||||
crate::slog!("gateway_logs_tool_marker_alpha");
|
||||
let resp = handle_gateway_logs_tool(&json!({"arguments": {"lines": 500}}), Some(json!(1)));
|
||||
assert!(resp.error.is_none(), "expected success: {:?}", resp.error);
|
||||
let text = resp.result.unwrap()["content"][0]["text"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_string();
|
||||
assert!(
|
||||
text.contains("gateway_logs_tool_marker_alpha"),
|
||||
"expected marker line in output: {text}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gateway_logs_tool_applies_filter() {
|
||||
crate::slog!("gateway_logs_tool_marker_beta");
|
||||
crate::slog!("unrelated_other_line");
|
||||
let resp = handle_gateway_logs_tool(
|
||||
&json!({"arguments": {"lines": 500, "filter": "gateway_logs_tool_marker_beta"}}),
|
||||
Some(json!(1)),
|
||||
);
|
||||
let text = resp.result.unwrap()["content"][0]["text"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_string();
|
||||
assert!(text.contains("gateway_logs_tool_marker_beta"));
|
||||
assert!(!text.contains("unrelated_other_line"));
|
||||
}
|
||||
|
||||
// ── explicit per-call `project` targeting (story 1208 AC 1) ─────────────
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -46,6 +46,33 @@ pub(crate) async fn tool_start_agent(args: &Value, ctx: &AppContext) -> Result<S
|
||||
.map_err(|e| format!("Serialization error: {e}"))
|
||||
}
|
||||
|
||||
/// MCP equivalent of the chat `start N` command (story 1209).
|
||||
///
|
||||
/// Unlike `start_agent`, which requires an exact `story_id` and has no
|
||||
/// busy-queue messaging, this accepts a bare story number, resolves it
|
||||
/// across all pipeline stages, supports the `coder-{hint}` agent-name
|
||||
/// shorthand, and reports "queued" (not "failed") when every coder is busy —
|
||||
/// by delegating to the same `handle_start` the Matrix bot's `start N`
|
||||
/// command uses.
|
||||
pub(crate) async fn tool_start_story(args: &Value, ctx: &AppContext) -> Result<String, String> {
|
||||
let story_number = args
|
||||
.get("story_number")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("Missing required argument: story_number")?;
|
||||
let agent_hint = args.get("agent_hint").and_then(|v| v.as_str());
|
||||
|
||||
let project_root = ctx.services.agents.get_project_root(&ctx.state)?;
|
||||
|
||||
Ok(crate::chat::transport::matrix::start::handle_start(
|
||||
"mcp",
|
||||
story_number,
|
||||
agent_hint,
|
||||
&project_root,
|
||||
&ctx.services.agents,
|
||||
)
|
||||
.await)
|
||||
}
|
||||
|
||||
pub(crate) async fn tool_stop_agent(args: &Value, ctx: &AppContext) -> Result<String, String> {
|
||||
let story_id = args
|
||||
.get("story_id")
|
||||
@@ -258,6 +285,42 @@ stage = "coder"
|
||||
}
|
||||
}
|
||||
|
||||
// -- tool_start_story (story 1209) ---------------------------------------
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_start_story_missing_story_number() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = test_ctx(tmp.path());
|
||||
let result = tool_start_story(&json!({}), &ctx).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("story_number"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_start_story_unknown_number_reports_not_found() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let project_root = tmp.path();
|
||||
for stage in &[
|
||||
"1_backlog",
|
||||
"2_current",
|
||||
"3_qa",
|
||||
"4_merge",
|
||||
"5_done",
|
||||
"6_archived",
|
||||
] {
|
||||
std::fs::create_dir_all(project_root.join(".huskies").join("work").join(stage))
|
||||
.unwrap();
|
||||
}
|
||||
let ctx = test_ctx(project_root);
|
||||
let result = tool_start_story(&json!({"story_number": "999999"}), &ctx)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
result.contains("No story") && result.contains("999999"),
|
||||
"unexpected response: {result}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn wait_for_agent_tool_missing_story_id() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
|
||||
@@ -10,7 +10,7 @@ pub(crate) use inspection::{
|
||||
tool_get_agent_config, tool_get_agent_output, tool_get_agent_remaining_turns_and_budget,
|
||||
};
|
||||
pub(crate) use lifecycle::{
|
||||
tool_list_agents, tool_start_agent, tool_stop_agent, tool_wait_for_agent,
|
||||
tool_list_agents, tool_start_agent, tool_start_story, tool_stop_agent, tool_wait_for_agent,
|
||||
};
|
||||
pub(crate) use worktree::{
|
||||
tool_cleanup_worktrees, tool_create_worktree, tool_get_editor_command, tool_list_worktrees,
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
//! MCP `chat_telemetry` tool — recent per-turn chat duration/ttft/cache/cost.
|
||||
|
||||
use serde_json::{Value, json};
|
||||
|
||||
/// Return up to `limit` (default 20, max 200) recent chat-turn telemetry
|
||||
/// records from the in-memory ring buffer, without any log-scraping.
|
||||
pub(crate) fn tool_chat_telemetry(args: &Value) -> Result<String, String> {
|
||||
let limit = args
|
||||
.get("limit")
|
||||
.and_then(|v| v.as_u64())
|
||||
.map(|n| n.min(200) as usize)
|
||||
.unwrap_or(20);
|
||||
|
||||
let turns = crate::service::chat_telemetry::recent(limit);
|
||||
serde_json::to_string_pretty(&json!({ "turns": turns }))
|
||||
.map_err(|e| format!("Serialization error: {e}"))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::service::chat_telemetry::{ChatTurnTelemetry, record};
|
||||
|
||||
#[test]
|
||||
fn tool_chat_telemetry_returns_recorded_turns() {
|
||||
record(ChatTurnTelemetry {
|
||||
timestamp: "2026-01-01T00:00:00Z".to_string(),
|
||||
persona: "test_chat_telemetry_tool_marker".to_string(),
|
||||
duration_ms: 250,
|
||||
ttft_ms: Some(90),
|
||||
input_tokens: 10,
|
||||
output_tokens: 20,
|
||||
cache_creation_input_tokens: 5,
|
||||
cache_read_input_tokens: 15,
|
||||
total_cost_usd: 0.05,
|
||||
});
|
||||
|
||||
let result = tool_chat_telemetry(&json!({"limit": 200})).unwrap();
|
||||
let parsed: Value = serde_json::from_str(&result).unwrap();
|
||||
let turns = parsed["turns"].as_array().unwrap();
|
||||
assert!(
|
||||
turns
|
||||
.iter()
|
||||
.any(|t| t["persona"] == "test_chat_telemetry_tool_marker"),
|
||||
"expected recorded turn to be present: {turns:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tool_chat_telemetry_default_limit_is_20() {
|
||||
for i in 0..25 {
|
||||
record(ChatTurnTelemetry {
|
||||
timestamp: "2026-01-01T00:00:00Z".to_string(),
|
||||
persona: format!("test_default_limit_marker_{i}"),
|
||||
duration_ms: 1,
|
||||
ttft_ms: None,
|
||||
input_tokens: 0,
|
||||
output_tokens: 0,
|
||||
cache_creation_input_tokens: 0,
|
||||
cache_read_input_tokens: 0,
|
||||
total_cost_usd: 0.0,
|
||||
});
|
||||
}
|
||||
let result = tool_chat_telemetry(&json!({})).unwrap();
|
||||
let parsed: Value = serde_json::from_str(&result).unwrap();
|
||||
assert_eq!(parsed["turns"].as_array().unwrap().len(), 20);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tool_chat_telemetry_empty_buffer_returns_empty_array() {
|
||||
// Not asserting emptiness (the shared global buffer may hold entries
|
||||
// from other tests) — just that the call succeeds and shape is right.
|
||||
let result = tool_chat_telemetry(&json!({"limit": 0})).unwrap();
|
||||
let parsed: Value = serde_json::from_str(&result).unwrap();
|
||||
assert_eq!(parsed["turns"].as_array().unwrap().len(), 0);
|
||||
}
|
||||
}
|
||||
@@ -5,9 +5,11 @@ use crate::http::context::AppContext;
|
||||
use crate::log_buffer;
|
||||
use serde_json::{Value, json};
|
||||
|
||||
mod chat_telemetry;
|
||||
mod permission;
|
||||
mod usage;
|
||||
|
||||
pub(crate) use chat_telemetry::tool_chat_telemetry;
|
||||
pub(crate) use permission::tool_prompt_permission;
|
||||
pub(crate) use usage::tool_get_token_usage;
|
||||
|
||||
|
||||
@@ -29,6 +29,7 @@ pub async fn dispatch_tool_call(
|
||||
"ensure_acceptance" => story_tools::tool_ensure_acceptance(&args, ctx),
|
||||
// Agent tools (async)
|
||||
"start_agent" => agent_tools::tool_start_agent(&args, ctx).await,
|
||||
"start_story" => agent_tools::tool_start_story(&args, ctx).await,
|
||||
"stop_agent" => agent_tools::tool_stop_agent(&args, ctx).await,
|
||||
"list_agents" => agent_tools::tool_list_agents(ctx).await,
|
||||
"get_agent_config" => agent_tools::tool_get_agent_config(ctx).await,
|
||||
@@ -87,6 +88,8 @@ pub async fn dispatch_tool_call(
|
||||
"prompt_permission" => diagnostics::tool_prompt_permission(&args, ctx).await,
|
||||
// Token usage
|
||||
"get_token_usage" => diagnostics::tool_get_token_usage(&args, ctx),
|
||||
// Chat turn telemetry (story 1209)
|
||||
"chat_telemetry" => diagnostics::tool_chat_telemetry(&args),
|
||||
// Delete story
|
||||
"delete_story" => story_tools::tool_delete_story(&args, ctx).await,
|
||||
// Purge story (CRDT tombstone — story 521)
|
||||
|
||||
@@ -23,6 +23,24 @@ pub(super) fn agent_tools() -> Vec<Value> {
|
||||
"required": ["story_id"]
|
||||
}
|
||||
}),
|
||||
json!({
|
||||
"name": "start_story",
|
||||
"description": "MCP equivalent of the chat `start N` command (story 1209): promotes a backlog item into active work and assigns a coder. Accepts a bare story number (resolved across all pipeline stages), an optional agent-name hint (e.g. 'opus' resolves to 'coder-opus'), and reports a friendly 'queued' message rather than an error when every coder is busy. Unlike start_agent, does not require an exact story_id.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"story_number": {
|
||||
"type": "string",
|
||||
"description": "The numeric story identifier (e.g. '331'), matched by number prefix across all pipeline stages."
|
||||
},
|
||||
"agent_hint": {
|
||||
"type": "string",
|
||||
"description": "Optional agent name hint (e.g. 'opus' resolves to 'coder-opus'). Omit to use the default coder agent."
|
||||
}
|
||||
},
|
||||
"required": ["story_number"]
|
||||
}
|
||||
}),
|
||||
json!({
|
||||
"name": "stop_agent",
|
||||
"description": "Stop a running agent. Worktree is preserved for inspection.",
|
||||
|
||||
@@ -45,6 +45,7 @@ mod tests {
|
||||
assert!(names.contains(&"record_tests"));
|
||||
assert!(names.contains(&"ensure_acceptance"));
|
||||
assert!(names.contains(&"start_agent"));
|
||||
assert!(names.contains(&"start_story"));
|
||||
assert!(names.contains(&"stop_agent"));
|
||||
assert!(names.contains(&"list_agents"));
|
||||
assert!(names.contains(&"get_agent_config"));
|
||||
@@ -118,7 +119,8 @@ mod tests {
|
||||
assert!(names.contains(&"edit"));
|
||||
assert!(names.contains(&"write"));
|
||||
assert!(names.contains(&"gc"));
|
||||
assert_eq!(tools.len(), 86);
|
||||
assert!(names.contains(&"chat_telemetry"));
|
||||
assert_eq!(tools.len(), 88);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -73,6 +73,19 @@ pub(super) fn system_tools() -> Vec<Value> {
|
||||
}
|
||||
}
|
||||
}),
|
||||
json!({
|
||||
"name": "chat_telemetry",
|
||||
"description": "Return recent per-turn chat telemetry (duration, time-to-first-token, cache tokens, and cost) for the Claude Code chat provider, sourced from an in-memory ring buffer rather than log-scraping. Only turns run against the Claude Code provider are recorded.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"description": "Maximum number of recent turns to return (default 20, max 200)"
|
||||
}
|
||||
}
|
||||
}
|
||||
}),
|
||||
json!({
|
||||
"name": "run_command",
|
||||
"description": "Execute a shell command in an agent's worktree directory. The working_dir must be inside .huskies/worktrees/. Returns stdout, stderr, exit_code, and timed_out. Supports SSE streaming (send Accept: text/event-stream) for long-running commands. Dangerous commands (rm -rf /, sudo, etc.) are blocked.",
|
||||
|
||||
@@ -199,11 +199,14 @@ where
|
||||
.get_project_root()
|
||||
.unwrap_or_else(|_| std::path::PathBuf::from("."));
|
||||
|
||||
let turn_start = std::time::Instant::now();
|
||||
let mut first_token_at: Option<std::time::Instant> = None;
|
||||
|
||||
let provider = ClaudeCodeProvider::new();
|
||||
let ClaudeCodeResult {
|
||||
messages: cc_messages,
|
||||
session_id,
|
||||
..
|
||||
usage,
|
||||
} = provider
|
||||
.chat_stream(
|
||||
&user_message,
|
||||
@@ -212,13 +215,34 @@ where
|
||||
None,
|
||||
None,
|
||||
&mut cancel_rx,
|
||||
|token| on_token(token),
|
||||
|token| {
|
||||
if first_token_at.is_none() {
|
||||
first_token_at = Some(std::time::Instant::now());
|
||||
}
|
||||
on_token(token)
|
||||
},
|
||||
|thinking| on_thinking(thinking),
|
||||
|tool_name| on_activity(tool_name),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| format!("Claude Code Error: {e}"))?;
|
||||
|
||||
if let Some(usage) = &usage {
|
||||
crate::service::chat_telemetry::record(
|
||||
crate::service::chat_telemetry::ChatTurnTelemetry {
|
||||
timestamp: Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(),
|
||||
persona: persona.to_string(),
|
||||
duration_ms: turn_start.elapsed().as_millis() as u64,
|
||||
ttft_ms: first_token_at.map(|t| (t - turn_start).as_millis() as u64),
|
||||
input_tokens: usage.input_tokens,
|
||||
output_tokens: usage.output_tokens,
|
||||
cache_creation_input_tokens: usage.cache_creation_input_tokens,
|
||||
cache_read_input_tokens: usage.cache_read_input_tokens,
|
||||
total_cost_usd: usage.total_cost_usd,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
// Build the final message history: user messages + Claude Code's turns.
|
||||
// If the session produced no structured messages (e.g. empty response),
|
||||
// fall back to an empty assistant message so the UI stops loading.
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
//! Chat turn telemetry — bounded in-memory ring buffer of per-turn
|
||||
//! duration/ttft/cache/cost metrics for the Claude Code chat provider.
|
||||
//!
|
||||
//! Populated by `llm::chat::run::chat()` on every completed Claude Code
|
||||
//! turn and surfaced via the `chat_telemetry` MCP tool so callers don't
|
||||
//! have to scrape `[pty-debug]` log lines to answer "how slow/expensive
|
||||
//! was that turn". Only the Claude Code provider path is instrumented —
|
||||
//! the Anthropic/Ollama tool-loop path in the same `chat()` function does
|
||||
//! not currently surface a comparable usage struct from its `chat_stream`
|
||||
//! return value.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
|
||||
/// Maximum number of recent turns retained in the ring buffer.
|
||||
const CAPACITY: usize = 200;
|
||||
|
||||
/// Timing and token/cost metrics for one completed chat turn.
|
||||
#[derive(Debug, Clone, serde::Serialize)]
|
||||
pub struct ChatTurnTelemetry {
|
||||
/// ISO 8601 UTC timestamp when the turn completed.
|
||||
pub timestamp: String,
|
||||
/// Persona name the turn was run against (e.g. `"timmy"`).
|
||||
pub persona: String,
|
||||
/// Wall-clock time from turn start to completion, in milliseconds.
|
||||
pub duration_ms: u64,
|
||||
/// Wall-clock time from turn start to the first streamed token, in
|
||||
/// milliseconds. `None` if no token was ever streamed.
|
||||
pub ttft_ms: Option<u64>,
|
||||
pub input_tokens: u64,
|
||||
pub output_tokens: u64,
|
||||
pub cache_creation_input_tokens: u64,
|
||||
pub cache_read_input_tokens: u64,
|
||||
pub total_cost_usd: f64,
|
||||
}
|
||||
|
||||
static BUFFER: OnceLock<Mutex<VecDeque<ChatTurnTelemetry>>> = OnceLock::new();
|
||||
|
||||
fn buffer() -> &'static Mutex<VecDeque<ChatTurnTelemetry>> {
|
||||
BUFFER.get_or_init(|| Mutex::new(VecDeque::with_capacity(CAPACITY)))
|
||||
}
|
||||
|
||||
/// Record a completed chat turn, evicting the oldest entry when at capacity.
|
||||
pub fn record(entry: ChatTurnTelemetry) {
|
||||
if let Ok(mut buf) = buffer().lock() {
|
||||
if buf.len() >= CAPACITY {
|
||||
buf.pop_front();
|
||||
}
|
||||
buf.push_back(entry);
|
||||
}
|
||||
}
|
||||
|
||||
/// Return up to `count` most recent turns, oldest first.
|
||||
pub fn recent(count: usize) -> Vec<ChatTurnTelemetry> {
|
||||
let buf = match buffer().lock() {
|
||||
Ok(b) => b,
|
||||
Err(_) => return vec![],
|
||||
};
|
||||
let start = buf.len().saturating_sub(count);
|
||||
buf.iter().skip(start).cloned().collect()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn sample(persona: &str) -> ChatTurnTelemetry {
|
||||
ChatTurnTelemetry {
|
||||
timestamp: "2026-01-01T00:00:00Z".to_string(),
|
||||
persona: persona.to_string(),
|
||||
duration_ms: 100,
|
||||
ttft_ms: Some(50),
|
||||
input_tokens: 10,
|
||||
output_tokens: 20,
|
||||
cache_creation_input_tokens: 0,
|
||||
cache_read_input_tokens: 0,
|
||||
total_cost_usd: 0.01,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn record_and_recent_round_trip() {
|
||||
// Use unique personas so this test is independent of the shared
|
||||
// global buffer's contents from other tests running concurrently.
|
||||
let marker = "test_round_trip_marker";
|
||||
record(sample(marker));
|
||||
let turns = recent(1000);
|
||||
assert!(turns.iter().any(|t| t.persona == marker));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recent_returns_most_recent_last() {
|
||||
let marker_a = "test_order_marker_a";
|
||||
let marker_b = "test_order_marker_b";
|
||||
record(sample(marker_a));
|
||||
record(sample(marker_b));
|
||||
let turns = recent(1000);
|
||||
let pos_a = turns.iter().position(|t| t.persona == marker_a);
|
||||
let pos_b = turns.iter().position(|t| t.persona == marker_b);
|
||||
if let (Some(a), Some(b)) = (pos_a, pos_b) {
|
||||
assert!(a < b, "marker_a must have been recorded before marker_b");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recent_respects_count_limit() {
|
||||
for _ in 0..5 {
|
||||
record(sample("test_limit_marker"));
|
||||
}
|
||||
let turns = recent(2);
|
||||
assert_eq!(turns.len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn buffer_evicts_oldest_past_capacity() {
|
||||
for i in 0..(CAPACITY + 10) {
|
||||
record(sample(&format!("test_evict_marker_{i}")));
|
||||
}
|
||||
let turns = recent(CAPACITY + 10);
|
||||
assert!(
|
||||
turns.len() <= CAPACITY,
|
||||
"buffer must never exceed CAPACITY entries, got {}",
|
||||
turns.len()
|
||||
);
|
||||
assert!(
|
||||
!turns.iter().any(|t| t.persona == "test_evict_marker_0"),
|
||||
"oldest entry must have been evicted"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -79,6 +79,7 @@ pub fn read_bot_config_raw(config_dir: &Path) -> BotConfigFields {
|
||||
password: s("password"),
|
||||
slack_bot_token: s("slack_bot_token"),
|
||||
slack_signing_secret: s("slack_signing_secret"),
|
||||
model: s("model"),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -91,6 +92,8 @@ pub struct BotConfigFields {
|
||||
pub password: Option<String>,
|
||||
pub slack_bot_token: Option<String>,
|
||||
pub slack_signing_secret: Option<String>,
|
||||
/// Claude Code model override configured for this bot (`gateway_info` MCP tool, story 1209).
|
||||
pub model: Option<String>,
|
||||
}
|
||||
|
||||
/// Write a `bot.toml` from the given content string.
|
||||
|
||||
@@ -38,13 +38,32 @@ use io::Client;
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::OnceLock;
|
||||
use std::sync::atomic::{AtomicI64, Ordering};
|
||||
use std::time::Instant;
|
||||
use tokio::sync::Mutex as TokioMutex;
|
||||
use tokio::sync::RwLock;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
pub use crate::crdt_state::NodePresenceView;
|
||||
|
||||
// ── Uptime (gateway_info MCP tool, story 1209) ──────────────────────────────
|
||||
|
||||
/// Instant the gateway process started, set once by `GatewayState::new`.
|
||||
static GATEWAY_START_TIME: OnceLock<Instant> = OnceLock::new();
|
||||
|
||||
/// Seconds elapsed since the gateway process started.
|
||||
///
|
||||
/// Lazily initialises the start time on first call so this never panics, but
|
||||
/// `GatewayState::new` calls it once at startup so the timer reflects actual
|
||||
/// process start rather than first `gateway_info` call in normal operation.
|
||||
pub fn gateway_uptime_secs() -> u64 {
|
||||
GATEWAY_START_TIME
|
||||
.get_or_init(Instant::now)
|
||||
.elapsed()
|
||||
.as_secs()
|
||||
}
|
||||
|
||||
// ── Status event broadcaster ────────────────────────────────────────────────
|
||||
|
||||
/// Capacity of the gateway status event broadcast channel.
|
||||
@@ -272,6 +291,7 @@ impl GatewayState {
|
||||
config_dir: PathBuf,
|
||||
port: u16,
|
||||
) -> Result<Self, String> {
|
||||
GATEWAY_START_TIME.get_or_init(Instant::now);
|
||||
let first_from_config = config::validate_config(&gateway_config)?;
|
||||
// Restore active project from CRDT if the stored value is still valid.
|
||||
let first = crate::crdt_state::read_gateway_active_project()
|
||||
@@ -838,6 +858,17 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gateway_uptime_secs_is_zero_or_positive_immediately_after_start() {
|
||||
// Just ensure it doesn't panic and returns a sane (small) value —
|
||||
// the static is process-global so we can't assert it's exactly 0.
|
||||
let uptime = gateway_uptime_secs();
|
||||
assert!(
|
||||
uptime < 3600,
|
||||
"uptime should be small in a fresh test run, got {uptime}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gateway_state_rejects_empty_config() {
|
||||
let config = GatewayConfig {
|
||||
|
||||
@@ -11,6 +11,9 @@ pub mod agents;
|
||||
pub mod anthropic;
|
||||
/// Bot command dispatch — parses and executes slash commands.
|
||||
pub mod bot_command;
|
||||
/// Chat turn telemetry — bounded in-memory ring buffer of per-turn
|
||||
/// duration/ttft/cache/cost metrics for the Claude Code chat provider.
|
||||
pub mod chat_telemetry;
|
||||
/// Shared pure helpers used across service modules.
|
||||
pub mod common;
|
||||
/// Diagnostics — server logs, CRDT dump, and permission management.
|
||||
|
||||
Reference in New Issue
Block a user