Merge epic/8089-test-ownership into main

.github/workflows/session-mongodb.yml

@@ -55,4 +55,4 @@ jobs:
         run: .github/scripts/start-session-mongo-replset.sh 27017
 
       - name: Run MongoDB driver conformance
-        run: cargo test --manifest-path integration-proofs/run-ledger-mongodb/Cargo.toml --features mongodb-storage-tests --test session_driver_conformance -- --ignored
+        run: cargo test --locked --manifest-path integration-proofs/run-ledger-mongodb/Cargo.toml --features mongodb-storage-tests --test session_driver_conformance -- --ignored

crates/tinyagents-graph/src/testkit/conformance.rs

@@ -792,11 +792,11 @@ where
 
     let short = key("g", "n", "short");
     cache
-        .put(&short, json!(1), Some(Duration::from_millis(30)))
+        .put(&short, json!(1), Some(Duration::from_secs(2)))
         .await
         .unwrap();
     assert_eq!(cache.get(&short).await.unwrap(), Some(json!(1)));
-    tokio::time::sleep(Duration::from_millis(80)).await;
+    tokio::time::sleep(Duration::from_millis(2_100)).await;
     assert_eq!(cache.get(&short).await.unwrap(), None, "expired after TTL");
     assert_eq!(cache.get(&short).await.unwrap(), None, "stays expired");
     cache.put(&short, json!(2), None).await.unwrap();

crates/tinyagents-harness-tools/README.md

@@ -29,7 +29,8 @@ run-state middleware and its required `goals` and `todos` modules.
   `tinyagents_orchestration::subagent::SubAgentJobRegistry`.
 - `thread_locks` provides the public `ThreadLockMap` concurrency primitive.
 - The crate root exports `CurrentTimeTool`, `ResolveTimeTool`,
-  `AskClarificationTool`, `PlanExitTool`, and the `per_call` binding adapter.
+  `AskClarificationTool`, `PlanExitTool`, `WaitTool`, `WaitLoopTool`, and the
+  `per_call` binding adapter.
 
 Subagent registry policy stays in `tinyagents-orchestration`; this crate owns
 the reusable tool declarations. The harness owns tool registration,

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

@@ -12,6 +12,7 @@ mod plan_exit;
 pub mod thread_locks;
 mod time;
 mod time_parse;
+mod wait;
 
 #[cfg(feature = "goals")]
 pub mod goals;
@@ -29,6 +30,7 @@ pub mod todos;
 pub use ask_clarification::AskClarificationTool;
 pub use plan_exit::{PLAN_EXIT_MARKER, PlanExitModeLookup, PlanExitTool};
 pub use time::{CurrentTimeTool, ResolveTimeTool, register_time_tools, time_tools};
+pub use wait::{WaitLoopTool, WaitTool};
 
 #[cfg(test)]
 #[path = "time_parse_tests.rs"]

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

@@ -0,0 +1,302 @@
+//! Tool: `wait` / `wait_loop` - delayed callback ticks for the orchestrator.
+//!
+//! These tools intentionally do not own scheduling state. They sleep for a
+//! bounded duration, then return the caller-provided message as a tool result so
+//! the orchestrator can decide whether to act, poll async sub-agents, or call the
+//! loop variant again.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use serde_json::json;
+use tinytools::{PermissionLevel, Tool, ToolResult, ToolTimeout};
+
+const DEFAULT_DURATION_SECS: u64 = 5;
+const MAX_DURATION_SECS: u64 = 600;
+const MILLIS_PER_SEC: u64 = 1_000;
+
+/// One-shot delayed callback tool for the orchestrator.
+pub struct WaitTool;
+/// Repeatable delayed callback tool for orchestrator-controlled polling loops.
+pub struct WaitLoopTool;
+
+impl WaitTool {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl Default for WaitTool {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl WaitLoopTool {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl Default for WaitLoopTool {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl Tool for WaitTool {
+    fn name(&self) -> &str {
+        "wait"
+    }
+
+    fn description(&self) -> &str {
+        "Wait for a bounded duration, then return the provided callback message \
+         to the orchestrator. Use this to create a delayed tick before checking \
+         async work or retrying a condition."
+    }
+
+    fn parameters_schema(&self) -> serde_json::Value {
+        wait_schema(false)
+    }
+
+    fn permission_level(&self) -> PermissionLevel {
+        PermissionLevel::None
+    }
+
+    fn timeout_policy(&self, args: &serde_json::Value) -> ToolTimeout {
+        timeout_policy_for_wait(args)
+    }
+
+    async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
+        execute_wait(args, false).await
+    }
+}
+
+#[async_trait]
+impl Tool for WaitLoopTool {
+    fn name(&self) -> &str {
+        "wait_loop"
+    }
+
+    fn description(&self) -> &str {
+        "Wait for a bounded duration, then return the same callback message plus \
+         a ready-to-call wait_loop instruction. Use this for deliberate polling \
+         loops where the orchestrator decides after each tick whether to repeat."
+    }
+
+    fn parameters_schema(&self) -> serde_json::Value {
+        wait_schema(true)
+    }
+
+    fn permission_level(&self) -> PermissionLevel {
+        PermissionLevel::None
+    }
+
+    fn timeout_policy(&self, args: &serde_json::Value) -> ToolTimeout {
+        timeout_policy_for_wait(args)
+    }
+
+    async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
+        execute_wait(args, true).await
+    }
+}
+
+/// Sleep for the requested duration, then return the callback tick payload.
+async fn execute_wait(args: serde_json::Value, loop_mode: bool) -> anyhow::Result<ToolResult> {
+    let request = match parse_wait_request(&args) {
+        Ok(request) => request,
+        Err(message) => return Ok(ToolResult::error(message)),
+    };
+    let tool_name = if loop_mode { "wait_loop" } else { "wait" };
+    tracing::info!(
+        "[wait] tool={} duration_ms={} iteration={} loop_key={} message_chars={}",
+        tool_name,
+        request.duration_ms,
+        request.iteration,
+        request.loop_key.as_deref().unwrap_or("none"),
+        request.message.chars().count()
+    );
+
+    tokio::time::sleep(Duration::from_millis(request.duration_ms)).await;
+
+    tracing::debug!(
+        "[wait] elapsed tool={} duration_ms={} iteration={} loop_key={}",
+        tool_name,
+        request.duration_ms,
+        request.iteration,
+        request.loop_key.as_deref().unwrap_or("none")
+    );
+
+    Ok(ToolResult::success(format_wait_tick(&request, loop_mode)))
+}
+
+/// Build the JSON schema shared by `wait` and `wait_loop`.
+fn wait_schema(loop_mode: bool) -> serde_json::Value {
+    let mut properties = serde_json::Map::new();
+    properties.insert(
+        "message".to_string(),
+        json!({
+            "type": "string",
+            "description": "Callback message to return to the orchestrator after the wait elapses."
+        }),
+    );
+    properties.insert(
+        "duration_secs".to_string(),
+        json!({
+            "type": "integer",
+            "minimum": 1,
+            "maximum": MAX_DURATION_SECS,
+            "description": "Seconds to wait before returning the callback. Default 5. Ignored when duration_ms is supplied."
+        }),
+    );
+    properties.insert(
+        "duration_ms".to_string(),
+        json!({
+            "type": "integer",
+            "minimum": 1,
+            "maximum": MAX_DURATION_SECS * MILLIS_PER_SEC,
+            "description": "Milliseconds to wait before returning the callback. Use only for short test-sized waits."
+        }),
+    );
+
+    if loop_mode {
+        properties.insert(
+            "loop_key".to_string(),
+            json!({
+                "type": "string",
+                "description": "Optional caller-defined key for the polling loop."
+            }),
+        );
+        properties.insert(
+            "iteration".to_string(),
+            json!({
+                "type": "integer",
+                "minimum": 1,
+                "description": "Current loop iteration. Defaults to 1; the returned instruction increments it."
+            }),
+        );
+    }
+
+    json!({
+        "type": "object",
+        "required": ["message"],
+        "properties": serde_json::Value::Object(properties)
+    })
+}
+
+/// Give the harness a deadline that covers the requested wait plus grace.
+fn timeout_policy_for_wait(args: &serde_json::Value) -> ToolTimeout {
+    let duration_ms = duration_ms_from_args(args).unwrap_or(DEFAULT_DURATION_SECS * MILLIS_PER_SEC);
+    let rounded_secs = duration_ms.saturating_add(MILLIS_PER_SEC - 1) / MILLIS_PER_SEC;
+    ToolTimeout::Millis(
+        rounded_secs
+            .saturating_add(1)
+            .saturating_mul(MILLIS_PER_SEC),
+    )
+}
+
+/// Extract and clamp the caller's requested wait duration in milliseconds.
+fn duration_ms_from_args(args: &serde_json::Value) -> Option<u64> {
+    if let Some(duration_ms) = args.get("duration_ms").and_then(|v| v.as_u64()) {
+        return Some(duration_ms.clamp(1, MAX_DURATION_SECS * MILLIS_PER_SEC));
+    }
+    args.get("duration_secs")
+        .and_then(|v| v.as_u64())
+        .map(|secs| secs.clamp(1, MAX_DURATION_SECS) * MILLIS_PER_SEC)
+}
+
+/// Validate tool arguments and normalize them into an internal request.
+fn parse_wait_request(args: &serde_json::Value) -> Result<WaitRequest, String> {
+    let message = args
+        .get("message")
+        .and_then(|v| v.as_str())
+        .unwrap_or("")
+        .trim()
+        .to_string();
+    if message.is_empty() {
+        return Err("wait: `message` is required".to_string());
+    }
+
+    let duration_ms = duration_ms_from_args(args).unwrap_or(DEFAULT_DURATION_SECS * MILLIS_PER_SEC);
+    let iteration = args
+        .get("iteration")
+        .and_then(|v| v.as_u64())
+        .unwrap_or(1)
+        .max(1);
+    let loop_key = args
+        .get("loop_key")
+        .and_then(|v| v.as_str())
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .map(str::to_string);
+
+    Ok(WaitRequest {
+        message,
+        duration_ms,
+        iteration,
+        loop_key,
+    })
+}
+
+/// Render the delayed callback as prose plus a machine-readable `[wait_tick]`.
+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(|| {
+        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",
+        "message": request.message,
+        "duration_ms": request.duration_ms,
+        "durationMs": request.duration_ms,
+        "duration_secs": seconds,
+        "durationSecs": seconds,
+        "loop": loop_mode,
+        "loop_key": request.loop_key,
+        "loopKey": request.loop_key,
+        "iteration": request.iteration,
+        "instructions": {
+            "callback_message": request.message,
+            "repeat": loop_instruction
+        }
+    });
+
+    let prefix = if loop_mode {
+        format!(
+            "Loop tick {} elapsed after {}ms.",
+            request.iteration, request.duration_ms
+        )
+    } else {
+        format!("Wait elapsed after {}ms.", request.duration_ms)
+    };
+
+    format!(
+        "{prefix}\n\n[wait_tick]\n{}\n[/wait_tick]\n\n{}",
+        serde_json::to_string(&payload).unwrap_or_else(|_| "{}".to_string()),
+        request.message
+    )
+}
+
+#[derive(Debug)]
+/// Normalized wait arguments used by both wait tools.
+struct WaitRequest {
+    message: String,
+    duration_ms: u64,
+    iteration: u64,
+    loop_key: Option<String>,
+}
+
+#[cfg(test)]
+#[path = "wait_tests.rs"]
+mod tests;