diff --git a/crates/tinyagents-harness/src/run_queue/forwarder.rs b/crates/tinyagents-harness/src/run_queue/forwarder.rs index f617fe2d4..abfc7f14b 100644 --- a/crates/tinyagents-harness/src/run_queue/forwarder.rs +++ b/crates/tinyagents-harness/src/run_queue/forwarder.rs @@ -391,7 +391,7 @@ impl Drop for SteeringForwarderGuard { .map(|(text, _lane, id)| (Some(id.clone()), Some(text.clone()))) .unwrap_or((None, None)); - // `RunQueue` is async (tokio `Mutex`); `Drop` is synchronous. Put + // `RunQueue::push` is async (tokio `Mutex`); `Drop` is synchronous. Push // the recovered steers back on a detached task so they land in the // session queue and become the next turn's input. The forwarder poll // loop keeps re-draining the queue, so even if the requeue completes @@ -400,19 +400,14 @@ impl Drop for SteeringForwarderGuard { Ok(rt) => { let label = thread_label.clone(); rt.spawn(async move { - // These were delivered before anything still queued, so - // they go back at the front of their lane, in order. - let mut steers = Vec::new(); - let mut collects = Vec::new(); + // Appended, not prepended: a later guard's residuals must + // not jump ahead of these, and successive drops each add + // theirs behind the last. for (text, lane, id) in requeue_items { - let item = T::requeued(id, text, &label, now_ms()); - match lane { - QueueLane::Collect => collects.push(item), - _ => steers.push(item), - } + queue + .push(lane, T::requeued(id, text, &label, now_ms())) + .await; } - queue.requeue_front(QueueLane::Steer, steers).await; - queue.requeue_front(QueueLane::Collect, collects).await; }); } Err(_) => { diff --git a/crates/tinyagents-harness/src/tools/ask_clarification.rs b/crates/tinyagents-harness/src/tools/ask_clarification.rs index e85715458..de5969c3d 100644 --- a/crates/tinyagents-harness/src/tools/ask_clarification.rs +++ b/crates/tinyagents-harness/src/tools/ask_clarification.rs @@ -83,12 +83,20 @@ impl Tool for AskClarificationTool { .filter(|q| !q.is_empty()) .unwrap_or("Could you clarify?"); - let options = args.get("options").and_then(|v| v.as_array()).map(|arr| { - arr.iter() - .filter_map(|v| v.as_str()) - .collect::>() - .join(", ") - }); + // Only non-empty string options are shown; with none, the question + // stands alone rather than ending on a dangling "Options:" line. + let options = args + .get("options") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str()) + .map(str::trim) + .filter(|opt| !opt.is_empty()) + .collect::>() + .join(", ") + }) + .filter(|opts| !opts.is_empty()); // Plain question text, no marker: this output IS the message the user // reads. The early-exit hook captures it verbatim as the pause question, diff --git a/crates/tinyagents-harness/src/tools/ask_clarification_test.rs b/crates/tinyagents-harness/src/tools/ask_clarification_test.rs index 615eeefaf..158fff0f4 100644 --- a/crates/tinyagents-harness/src/tools/ask_clarification_test.rs +++ b/crates/tinyagents-harness/src/tools/ask_clarification_test.rs @@ -134,3 +134,15 @@ fn wire_schema_is_stable() { }) ); } + +#[tokio::test] +async fn options_without_usable_strings_add_no_options_line() { + let tool = AskClarificationTool::new(); + for options in [json!([]), json!([1, 2]), json!(["", " "])] { + let result = tool + .execute(json!({ "question": "Which one?", "options": options })) + .await + .unwrap(); + assert_eq!(result.output(), "Which one?"); + } +} diff --git a/crates/tinyagents-harness/src/tools/wait.rs b/crates/tinyagents-harness/src/tools/wait.rs index 256901aed..5cf93d8af 100644 --- a/crates/tinyagents-harness/src/tools/wait.rs +++ b/crates/tinyagents-harness/src/tools/wait.rs @@ -243,15 +243,17 @@ fn parse_wait_request(args: &serde_json::Value) -> Result { fn format_wait_tick(request: &WaitRequest, loop_mode: bool) -> String { let seconds = request.duration_ms as f64 / MILLIS_PER_SEC as f64; let loop_instruction = loop_mode.then(|| { - json!({ - "tool": "wait_loop", - "arguments": { - "message": request.message, - "duration_ms": request.duration_ms, - "loop_key": request.loop_key, - "iteration": request.iteration.saturating_add(1) - } - }) + let mut arguments = json!({ + "message": request.message, + "duration_ms": request.duration_ms, + "iteration": request.iteration.saturating_add(1) + }); + // The schema allows `loop_key` only as a string: omit it rather than + // hand back a ready-to-call instruction carrying `null`. + if let Some(loop_key) = &request.loop_key { + arguments["loop_key"] = json!(loop_key); + } + json!({ "tool": "wait_loop", "arguments": arguments }) }); let payload = json!({ "status": "elapsed", diff --git a/crates/tinyagents-harness/src/tools/wait_test.rs b/crates/tinyagents-harness/src/tools/wait_test.rs index 731dafbd9..434bb81f0 100644 --- a/crates/tinyagents-harness/src/tools/wait_test.rs +++ b/crates/tinyagents-harness/src/tools/wait_test.rs @@ -66,6 +66,24 @@ fn wait_loop_tick_at_max_iteration_does_not_overflow() { assert!(output.contains(&format!("\"iteration\":{}", u64::MAX))); } +#[test] +fn wait_loop_tick_omits_an_absent_loop_key() { + let request = parse_wait_request(&json!({ "message": "poll", "duration_ms": 10 })).unwrap(); + let output = format_wait_tick(&request, true); + let json = output + .split("[wait_tick]\n") + .nth(1) + .and_then(|rest| rest.split("\n[/wait_tick]").next()) + .expect("wait_tick payload"); + let payload: serde_json::Value = serde_json::from_str(json).unwrap(); + let arguments = &payload["instructions"]["repeat"]["arguments"]; + assert!(arguments.is_object()); + assert!( + arguments.get("loop_key").is_none(), + "the repeat call must not carry a null loop_key: {arguments}" + ); +} + #[tokio::test] async fn wait_execute_returns_callback_message() { let res = WaitTool::new() diff --git a/crates/tinyagents-session/src/transcript/spend.rs b/crates/tinyagents-session/src/transcript/spend.rs index 35e06c937..e3412b337 100644 --- a/crates/tinyagents-session/src/transcript/spend.rs +++ b/crates/tinyagents-session/src/transcript/spend.rs @@ -111,7 +111,16 @@ fn spend_from_usages<'a>( /// `request_id` is the compaction's and no appended line already counted /// spend under that `request_id`. A compaction written without a /// `request_id` falls back to the usage on its final assistant row. -fn file_spend(path: &Path) -> Option<(TranscriptSpend, TranscriptMeta)> { +/// +/// `earlier_requests` holds the `request_id`s that earlier generation files of +/// the same conversation already counted. A successor generation starts by +/// writing the rows it retained from its parent, and those rows keep their +/// original `request_id`, so a row whose `request_id` is in that set is +/// carried history, not new spend. The ids this file counts are added to it. +fn file_spend( + path: &Path, + earlier_requests: &mut HashSet, +) -> Option<(TranscriptSpend, TranscriptMeta)> { let display = match read_transcript_display(path) { Ok(display) => display, Err(err) => { @@ -130,7 +139,11 @@ fn file_spend(path: &Path) -> Option<(TranscriptSpend, TranscriptMeta)> { for record in &display.records { match record { DisplayRecord::Message(message) if !message.interrupted => { - if let Some(usage) = message.message.turn_usage.as_ref() { + let carried = message + .request_id + .as_deref() + .is_some_and(|request_id| earlier_requests.contains(request_id)); + if !carried && let Some(usage) = message.message.turn_usage.as_ref() { usages.push(usage); if let Some(request_id) = message.request_id.as_deref() { counted_requests.insert(request_id); @@ -141,7 +154,9 @@ fn file_spend(path: &Path) -> Option<(TranscriptSpend, TranscriptMeta)> { DisplayRecord::Message(_) => {} DisplayRecord::Compaction(marker) => { match marker.request_id.as_deref() { - Some(turn) if !counted_requests.contains(turn) => { + Some(turn) + if !counted_requests.contains(turn) && !earlier_requests.contains(turn) => + { let fresh: Vec<&TurnUsage> = marker .replacement .iter() @@ -180,6 +195,7 @@ fn file_spend(path: &Path) -> Option<(TranscriptSpend, TranscriptMeta)> { } } let spend = spend_from_usages(usages, display.meta.model.as_ref()); + earlier_requests.extend(counted_requests.into_iter().map(str::to_owned)); Some((spend, display.meta)) } @@ -246,9 +262,12 @@ pub struct ThreadSpend { pub fn thread_spend(workspace_dir: &Path, thread_id: &str) -> ThreadSpend { let mut out = ThreadSpend::default(); let roots = find_root_transcripts_for_thread(workspace_dir, thread_id); + // Shared across the thread's root files (oldest first), so rows a later + // generation carried over from an earlier one are not counted twice. + let mut root_requests = HashSet::new(); for root in &roots { out.found_transcript = true; - if let Some((spend, meta)) = file_spend(root) { + if let Some((spend, meta)) = file_spend(root, &mut root_requests) { out.root.input_tokens = out.root.input_tokens.saturating_add(spend.input_tokens); out.root.output_tokens = out.root.output_tokens.saturating_add(spend.output_tokens); out.root.cached_input_tokens = out @@ -272,7 +291,7 @@ pub fn thread_spend(workspace_dir: &Path, thread_id: &str) -> ThreadSpend { out.updated = Some(meta.updated); } for child in descendant_transcripts(root) { - let Some((spend, child_meta)) = file_spend(&child) else { + let Some((spend, child_meta)) = file_spend(&child, &mut HashSet::new()) else { continue; }; out.found_transcript = true; diff --git a/crates/tinyagents-session/src/transcript/spend_test.rs b/crates/tinyagents-session/src/transcript/spend_test.rs index 5359b1518..018bfdef8 100644 --- a/crates/tinyagents-session/src/transcript/spend_test.rs +++ b/crates/tinyagents-session/src/transcript/spend_test.rs @@ -524,3 +524,45 @@ fn request_less_compaction_with_usage_identical_to_history_still_counts() { assert_eq!(spend.root.input_tokens, 10_000); assert_eq!(spend.root.turns, 2); } + +/// A successor generation starts with the rows it retained from its parent. +/// Those rows keep their original `request_id`, and their usage was already +/// counted in the sealed generation. +#[test] +fn retained_rows_in_a_successor_generation_are_not_counted_twice() { + use crate::transcript::{FileTranscriptLocator, SessionRef, TranscriptLocator, TruncateCut}; + let tmp = tempfile::tempdir().expect("tempdir"); + let thread = "thread-generations"; + let locator = FileTranscriptLocator::new(tmp.path()); + let session = SessionRef::scoped(thread, "orchestrator"); + let first = locator + .open_session(&session, meta("orchestrator", "root", Some(thread))) + .expect("open session"); + first + .append(TranscriptMessage::new("user", "q0")) + .expect("append user"); + let mut answer = TranscriptMessage::assistant("a0"); + answer.turn_usage = Some(turn_usage(4_000, 40, 0)); + answer.request_id = Some("req-0".into()); + answer.preserve_request_id = true; + first.append(answer).expect("append answer"); + + let before = thread_spend(tmp.path(), thread); + assert_eq!(before.root.input_tokens, 4_000); + + locator + .truncate_into_next_generation( + &session, + TruncateCut::BeforeIndex(2), + meta("orchestrator", "root", Some(thread)), + ) + .expect("next generation"); + assert_eq!( + find_root_transcripts_for_thread(tmp.path(), thread).len(), + 2, + "both generations are root transcripts of the thread" + ); + let after = thread_spend(tmp.path(), thread); + assert_eq!(after.root.input_tokens, 4_000, "retained row counted twice"); + assert_eq!(after.root.turns, 1); +}