huskies: merge 531_story_mcp_tool_to_read_agent_session_logs_from_disk_not_just_live_stream
This commit is contained in:
dave
+227
-123
@@ -96,111 +96,97 @@ pub(super) fn tool_list_agents(ctx: &AppContext) -> Result<String, String> {
|
||||
.map_err(|e| format!("Serialization error: {e}"))
|
||||
}
|
||||
|
||||
pub(super) async fn tool_get_agent_output_poll(args: &Value, ctx: &AppContext) -> Result<String, String> {
|
||||
let story_id = args
|
||||
.get("story_id")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("Missing required argument: story_id")?;
|
||||
let agent_name = args
|
||||
.get("agent_name")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("Missing required argument: agent_name")?;
|
||||
|
||||
// Try draining in-memory events first.
|
||||
match ctx.agents.drain_events(story_id, agent_name) {
|
||||
Ok(drained) => {
|
||||
let done = drained.iter().any(|e| {
|
||||
matches!(
|
||||
e,
|
||||
crate::agents::AgentEvent::Done { .. }
|
||||
| crate::agents::AgentEvent::Error { .. }
|
||||
)
|
||||
});
|
||||
|
||||
let events: Vec<serde_json::Value> = drained
|
||||
.into_iter()
|
||||
.filter_map(|e| serde_json::to_value(&e).ok())
|
||||
.collect();
|
||||
|
||||
serde_json::to_string_pretty(&json!({
|
||||
"events": events,
|
||||
"done": done,
|
||||
"event_count": events.len(),
|
||||
"message": if done { "Agent stream ended." } else if events.is_empty() { "No new events. Call again to continue." } else { "Events returned. Call again to continue." }
|
||||
}))
|
||||
.map_err(|e| format!("Serialization error: {e}"))
|
||||
}
|
||||
Err(_) => {
|
||||
// Agent not in memory — fall back to persistent log file.
|
||||
get_agent_output_from_log(story_id, agent_name, ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Fall back to reading agent output from the persistent log file on disk.
|
||||
/// Read agent session logs from disk and return a human-readable timeline.
|
||||
///
|
||||
/// Tries to find the log file via the agent's stored log_session_id first,
|
||||
/// then falls back to `find_latest_log` scanning the log directory.
|
||||
pub(super) fn get_agent_output_from_log(
|
||||
story_id: &str,
|
||||
agent_name: &str,
|
||||
/// Stitches all session log files for the story together in chronological
|
||||
/// order. If `agent_name` is omitted, logs from every agent are included.
|
||||
/// Supports `lines` (tail the last N lines) and `filter` (substring match).
|
||||
/// If a named agent is currently running, its buffered in-memory events are
|
||||
/// appended as "live" output so the caller sees everything in one call.
|
||||
pub(super) async fn tool_get_agent_output(
|
||||
args: &Value,
|
||||
ctx: &AppContext,
|
||||
) -> Result<String, String> {
|
||||
use crate::agent_log;
|
||||
|
||||
let story_id = args
|
||||
.get("story_id")
|
||||
.and_then(|v| v.as_str())
|
||||
.ok_or("Missing required argument: story_id")?;
|
||||
let agent_name_filter = args.get("agent_name").and_then(|v| v.as_str());
|
||||
let tail = args
|
||||
.get("lines")
|
||||
.and_then(|v| v.as_u64())
|
||||
.map(|n| n as usize);
|
||||
let filter = args.get("filter").and_then(|v| v.as_str());
|
||||
|
||||
let project_root = ctx.agents.get_project_root(&ctx.state)?;
|
||||
|
||||
// Try to find the log file: first from in-memory agent info, then by scanning.
|
||||
let log_path = ctx
|
||||
.agents
|
||||
.get_log_info(story_id, agent_name)
|
||||
.map(|(session_id, root)| agent_log::log_file_path(&root, story_id, agent_name, &session_id))
|
||||
.filter(|p| p.exists())
|
||||
.or_else(|| agent_log::find_latest_log(&project_root, story_id, agent_name));
|
||||
// Collect all matching log files, oldest first.
|
||||
let log_files =
|
||||
agent_log::list_story_log_files(&project_root, story_id, agent_name_filter);
|
||||
|
||||
let log_path = match log_path {
|
||||
Some(p) => p,
|
||||
None => {
|
||||
return serde_json::to_string_pretty(&json!({
|
||||
"events": [],
|
||||
"done": true,
|
||||
"event_count": 0,
|
||||
"message": format!("No agent '{agent_name}' for story '{story_id}' and no log file found."),
|
||||
"source": "none",
|
||||
}))
|
||||
.map_err(|e| format!("Serialization error: {e}"));
|
||||
let mut all_lines: Vec<String> = Vec::new();
|
||||
|
||||
for path in &log_files {
|
||||
let file_name = path
|
||||
.file_name()
|
||||
.and_then(|n| n.to_str())
|
||||
.unwrap_or("?");
|
||||
all_lines.push(format!("=== {} ===", file_name.trim_end_matches(".log")));
|
||||
match agent_log::read_log_as_readable_lines(path) {
|
||||
Ok(lines) => all_lines.extend(lines),
|
||||
Err(e) => all_lines.push(format!("[ERROR reading log: {e}]")),
|
||||
}
|
||||
all_lines.push(String::new()); // blank line between sessions
|
||||
}
|
||||
|
||||
// Append buffered live events when a specific agent is requested.
|
||||
if let Some(agent_name) = agent_name_filter
|
||||
&& let Ok(live_events) = ctx.agents.drain_events(story_id, agent_name)
|
||||
&& !live_events.is_empty()
|
||||
{
|
||||
all_lines.push(format!("=== {agent_name} (live) ==="));
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
for event in &live_events {
|
||||
if let Ok(event_value) = serde_json::to_value(event)
|
||||
&& let Some(line) =
|
||||
agent_log::format_log_entry_as_text(&now, &event_value)
|
||||
{
|
||||
all_lines.push(line);
|
||||
}
|
||||
}
|
||||
all_lines.push(String::new());
|
||||
}
|
||||
|
||||
if log_files.is_empty() {
|
||||
let agent_hint = agent_name_filter
|
||||
.map(|n| format!(" agent '{n}'"))
|
||||
.unwrap_or_default();
|
||||
return Ok(format!(
|
||||
"No log files found for story '{story_id}'{agent_hint}."
|
||||
));
|
||||
}
|
||||
|
||||
// Apply substring filter (always keep section headers).
|
||||
let filtered: Vec<String> = if let Some(f) = filter {
|
||||
all_lines
|
||||
.into_iter()
|
||||
.filter(|l| l.starts_with("===") || l.contains(f))
|
||||
.collect()
|
||||
} else {
|
||||
all_lines
|
||||
};
|
||||
|
||||
match agent_log::read_log(&log_path) {
|
||||
Ok(entries) => {
|
||||
let events: Vec<serde_json::Value> = entries
|
||||
.into_iter()
|
||||
.map(|e| {
|
||||
let mut val = e.event;
|
||||
if let serde_json::Value::Object(ref mut map) = val {
|
||||
map.insert(
|
||||
"timestamp".to_string(),
|
||||
serde_json::Value::String(e.timestamp),
|
||||
);
|
||||
}
|
||||
val
|
||||
})
|
||||
.collect();
|
||||
// Apply tail (last N lines).
|
||||
let output = if let Some(n) = tail {
|
||||
let start = filtered.len().saturating_sub(n);
|
||||
filtered[start..].join("\n")
|
||||
} else {
|
||||
filtered.join("\n")
|
||||
};
|
||||
|
||||
let count = events.len();
|
||||
serde_json::to_string_pretty(&json!({
|
||||
"events": events,
|
||||
"done": true,
|
||||
"event_count": count,
|
||||
"message": "Events loaded from persistent log file.",
|
||||
"source": "log_file",
|
||||
"log_file": log_path.to_string_lossy(),
|
||||
}))
|
||||
.map_err(|e| format!("Serialization error: {e}"))
|
||||
}
|
||||
Err(e) => Err(format!("Failed to read log file: {e}")),
|
||||
}
|
||||
Ok(output)
|
||||
}
|
||||
|
||||
pub(super) fn tool_get_agent_config(ctx: &AppContext) -> Result<String, String> {
|
||||
@@ -396,61 +382,179 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_get_agent_output_poll_missing_story_id() {
|
||||
async fn tool_get_agent_output_missing_story_id() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = test_ctx(tmp.path());
|
||||
let result = tool_get_agent_output_poll(&json!({"agent_name": "bot"}), &ctx).await;
|
||||
let result = tool_get_agent_output(&json!({}), &ctx).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("story_id"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_get_agent_output_poll_missing_agent_name() {
|
||||
async fn tool_get_agent_output_no_logs_returns_not_found_message() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = test_ctx(tmp.path());
|
||||
let result =
|
||||
tool_get_agent_output_poll(&json!({"story_id": "1_test"}), &ctx).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("agent_name"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_get_agent_output_poll_no_agent_falls_back_to_empty_log() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = test_ctx(tmp.path());
|
||||
// No agent registered, no log file → returns empty response from log fallback
|
||||
let result = tool_get_agent_output_poll(
|
||||
// No agent registered, no log file → returns "no log files found" message
|
||||
let result = tool_get_agent_output(
|
||||
&json!({"story_id": "99_nope", "agent_name": "bot"}),
|
||||
&ctx,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let parsed: Value = serde_json::from_str(&result).unwrap();
|
||||
assert_eq!(parsed["done"], true);
|
||||
assert_eq!(parsed["event_count"], 0);
|
||||
assert!(
|
||||
parsed["message"].as_str().unwrap_or("").contains("No agent"),
|
||||
"expected 'No agent' message: {parsed}"
|
||||
result.contains("No log files found"),
|
||||
"expected 'No log files found' message: {result}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_get_agent_output_poll_with_running_agent_returns_empty_events() {
|
||||
async fn tool_get_agent_output_agent_name_is_optional() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = test_ctx(tmp.path());
|
||||
// Inject a running agent — no events broadcast yet
|
||||
ctx.agents
|
||||
.inject_test_agent("10_story", "worker", crate::agents::AgentStatus::Running);
|
||||
let result = tool_get_agent_output_poll(
|
||||
&json!({"story_id": "10_story", "agent_name": "worker"}),
|
||||
// No agent_name provided — should succeed (no error)
|
||||
let result = tool_get_agent_output(
|
||||
&json!({"story_id": "99_nope"}),
|
||||
&ctx,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let parsed: Value = serde_json::from_str(&result).unwrap();
|
||||
assert_eq!(parsed["done"], false);
|
||||
assert_eq!(parsed["event_count"], 0);
|
||||
assert!(parsed["events"].is_array());
|
||||
assert!(result.contains("No log files found"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_get_agent_output_reads_from_disk() {
|
||||
use crate::agent_log::AgentLogWriter;
|
||||
use crate::agents::AgentEvent;
|
||||
use crate::store::StoreOps;
|
||||
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = test_ctx(tmp.path());
|
||||
// Point the store at the tmp root so the tool can find log files.
|
||||
ctx.store
|
||||
.set("project_root", json!(tmp.path().to_string_lossy().as_ref()));
|
||||
|
||||
// Write a log file
|
||||
let mut writer = AgentLogWriter::new(
|
||||
tmp.path(),
|
||||
"42_story_foo",
|
||||
"coder-1",
|
||||
"sess-test",
|
||||
)
|
||||
.unwrap();
|
||||
writer
|
||||
.write_event(&AgentEvent::Output {
|
||||
story_id: "42_story_foo".to_string(),
|
||||
agent_name: "coder-1".to_string(),
|
||||
text: "My readable output line".to_string(),
|
||||
})
|
||||
.unwrap();
|
||||
writer
|
||||
.write_event(&AgentEvent::Done {
|
||||
story_id: "42_story_foo".to_string(),
|
||||
agent_name: "coder-1".to_string(),
|
||||
session_id: Some("sess-test".to_string()),
|
||||
})
|
||||
.unwrap();
|
||||
drop(writer);
|
||||
|
||||
let result = tool_get_agent_output(
|
||||
&json!({"story_id": "42_story_foo", "agent_name": "coder-1"}),
|
||||
&ctx,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(
|
||||
result.contains("My readable output line"),
|
||||
"expected output text in result: {result}"
|
||||
);
|
||||
assert!(result.contains("DONE"), "expected DONE marker: {result}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_get_agent_output_tail_limits_lines() {
|
||||
use crate::agent_log::AgentLogWriter;
|
||||
use crate::agents::AgentEvent;
|
||||
use crate::store::StoreOps;
|
||||
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = test_ctx(tmp.path());
|
||||
ctx.store
|
||||
.set("project_root", json!(tmp.path().to_string_lossy().as_ref()));
|
||||
|
||||
let mut writer = AgentLogWriter::new(
|
||||
tmp.path(),
|
||||
"42_story_bar",
|
||||
"coder-1",
|
||||
"sess-tail",
|
||||
)
|
||||
.unwrap();
|
||||
for i in 0..10 {
|
||||
writer
|
||||
.write_event(&AgentEvent::Output {
|
||||
story_id: "42_story_bar".to_string(),
|
||||
agent_name: "coder-1".to_string(),
|
||||
text: format!("line {i}"),
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
drop(writer);
|
||||
|
||||
let result = tool_get_agent_output(
|
||||
&json!({"story_id": "42_story_bar", "agent_name": "coder-1", "lines": 3}),
|
||||
&ctx,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Should contain "line 7", "line 8", "line 9" but NOT "line 0"
|
||||
assert!(result.contains("line 9"), "should contain last line: {result}");
|
||||
assert!(!result.contains("line 0"), "should not contain early lines: {result}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_get_agent_output_filter_narrows_results() {
|
||||
use crate::agent_log::AgentLogWriter;
|
||||
use crate::agents::AgentEvent;
|
||||
use crate::store::StoreOps;
|
||||
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = test_ctx(tmp.path());
|
||||
ctx.store
|
||||
.set("project_root", json!(tmp.path().to_string_lossy().as_ref()));
|
||||
|
||||
let mut writer = AgentLogWriter::new(
|
||||
tmp.path(),
|
||||
"42_story_baz",
|
||||
"coder-1",
|
||||
"sess-filter",
|
||||
)
|
||||
.unwrap();
|
||||
writer
|
||||
.write_event(&AgentEvent::Output {
|
||||
story_id: "42_story_baz".to_string(),
|
||||
agent_name: "coder-1".to_string(),
|
||||
text: "needle line here".to_string(),
|
||||
})
|
||||
.unwrap();
|
||||
writer
|
||||
.write_event(&AgentEvent::Output {
|
||||
story_id: "42_story_baz".to_string(),
|
||||
agent_name: "coder-1".to_string(),
|
||||
text: "haystack only".to_string(),
|
||||
})
|
||||
.unwrap();
|
||||
drop(writer);
|
||||
|
||||
let result = tool_get_agent_output(
|
||||
&json!({"story_id": "42_story_baz", "agent_name": "coder-1", "filter": "needle"}),
|
||||
&ctx,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(result.contains("needle"), "filter should keep matching lines: {result}");
|
||||
assert!(!result.contains("haystack"), "filter should remove non-matching lines: {result}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
+38
-14
@@ -134,7 +134,17 @@ pub async fn mcp_post_handler(req: &Request, body: Body, ctx: Data<&Arc<AppConte
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("");
|
||||
if tool_name == "get_agent_output" {
|
||||
return handle_agent_output_sse(rpc.id, &rpc.params, &ctx);
|
||||
// Only use live SSE streaming when agent_name is explicitly provided.
|
||||
// Without agent_name, fall through to the disk-based handler below.
|
||||
let has_agent_name = rpc
|
||||
.params
|
||||
.get("arguments")
|
||||
.and_then(|a| a.get("agent_name"))
|
||||
.and_then(|v| v.as_str())
|
||||
.is_some();
|
||||
if has_agent_name {
|
||||
return handle_agent_output_sse(rpc.id, &rpc.params, &ctx);
|
||||
}
|
||||
}
|
||||
if tool_name == "run_command" {
|
||||
return shell_tools::handle_run_command_sse(rpc.id, &rpc.params, &ctx);
|
||||
@@ -502,24 +512,28 @@ fn handle_tools_list(id: Option<Value>) -> JsonRpcResponse {
|
||||
},
|
||||
{
|
||||
"name": "get_agent_output",
|
||||
"description": "Poll recent output from a running agent. Subscribes to the agent's event stream and collects events for up to 2 seconds. Returns text output and status events. Call repeatedly to follow progress.",
|
||||
"description": "Read agent session logs from disk as a human-readable timeline. Stitches all sessions for the story together in chronological order — text output, tool calls, tool results, errors. Works for both running and completed agents. If agent_name is omitted, returns logs from every agent that worked on the story. If the named agent is currently running, live buffered events are appended.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"story_id": {
|
||||
"type": "string",
|
||||
"description": "Story identifier"
|
||||
"description": "Story identifier (e.g. '42_story_my_feature')"
|
||||
},
|
||||
"agent_name": {
|
||||
"type": "string",
|
||||
"description": "Agent name"
|
||||
"description": "Optional: filter to a specific agent (e.g. 'mergemaster', 'coder-1'). Omit to see all agents."
|
||||
},
|
||||
"timeout_ms": {
|
||||
"lines": {
|
||||
"type": "integer",
|
||||
"description": "How long to wait for events in milliseconds (default: 2000, max: 10000)"
|
||||
"description": "Optional: return only the last N lines (tail). Useful for large logs."
|
||||
},
|
||||
"filter": {
|
||||
"type": "string",
|
||||
"description": "Optional: return only lines containing this substring (e.g. 'ERROR', 'TOOL:', a function name)."
|
||||
}
|
||||
},
|
||||
"required": ["story_id", "agent_name"]
|
||||
"required": ["story_id"]
|
||||
}
|
||||
},
|
||||
{
|
||||
@@ -1300,7 +1314,7 @@ async fn handle_tools_call(
|
||||
"list_agents" => agent_tools::tool_list_agents(ctx),
|
||||
"get_agent_config" => agent_tools::tool_get_agent_config(ctx),
|
||||
"reload_agent_config" => agent_tools::tool_get_agent_config(ctx),
|
||||
"get_agent_output" => agent_tools::tool_get_agent_output_poll(&args, ctx).await,
|
||||
"get_agent_output" => agent_tools::tool_get_agent_output(&args, ctx).await,
|
||||
"wait_for_agent" => agent_tools::tool_wait_for_agent(&args, ctx).await,
|
||||
// Worktree tools
|
||||
"create_worktree" => agent_tools::tool_create_worktree(&args, ctx).await,
|
||||
@@ -1756,7 +1770,11 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mcp_post_sse_get_agent_output_missing_agent_name() {
|
||||
async fn mcp_post_sse_get_agent_output_without_agent_name_returns_disk_content() {
|
||||
// Without agent_name the SSE live-streaming intercept is skipped and
|
||||
// the disk-based handler runs. The transport still wraps the result in
|
||||
// SSE format (data: …\n\n) because the client sent Accept: text/event-stream,
|
||||
// but the content should be a valid JSON-RPC result, not a subscribe error.
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let ctx = std::sync::Arc::new(test_ctx(tmp.path()));
|
||||
let cli = poem::test::TestClient::new(test_mcp_app(ctx));
|
||||
@@ -1767,11 +1785,17 @@ mod tests {
|
||||
.body(r#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_agent_output","arguments":{"story_id":"1_test"}}}"#)
|
||||
.send()
|
||||
.await;
|
||||
assert_eq!(
|
||||
resp.0.headers().get("content-type").unwrap(),
|
||||
"text/event-stream",
|
||||
"expected SSE content-type"
|
||||
);
|
||||
let body = resp.0.into_body().into_string().await.unwrap();
|
||||
// Body is SSE-wrapped: "data: {…}\n\n" — strip the prefix and verify it's
|
||||
// a valid JSON-RPC result (not an error about missing agent_name).
|
||||
let json_part = body.trim_start_matches("data: ").trim_end_matches("\n\n").trim();
|
||||
let parsed: serde_json::Value = serde_json::from_str(json_part)
|
||||
.unwrap_or_else(|_| panic!("expected JSON-RPC in SSE body, got: {body}"));
|
||||
assert!(parsed.get("result").is_some(),
|
||||
"expected JSON-RPC result (disk-based handler ran): {parsed}");
|
||||
// Must NOT be an error about missing agent_name (agent_name is now optional)
|
||||
assert!(parsed.get("error").is_none(),
|
||||
"unexpected error when agent_name omitted: {parsed}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
Reference in New Issue
Block a user