fix: wave 4 follow-ups (generation spend, clarification options, wait_loop args, residual order)

crates/tinyagents-harness/src/run_queue/forwarder.rs

@@ -391,7 +391,7 @@ impl<T: QueuedMessage> Drop for SteeringForwarderGuard<T> {
             .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<T: QueuedMessage> Drop for SteeringForwarderGuard<T> {
             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(_) => {

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::<Vec<_>>()
-                .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::<Vec<_>>()
+                    .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,

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?");
+    }
+}

crates/tinyagents-harness/src/tools/wait.rs

@@ -243,15 +243,17 @@ fn parse_wait_request(args: &serde_json::Value) -> Result<WaitRequest, String> {
 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",

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()

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<String>,
+) -> 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;

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);
+}