Skip to content

Commit 1f24dbe

Browse files
authored
Merge pull request #570 from M3gA-Mind/feat/port-openhuman-pin-wait-tools-agent-loop
Merge epic/8089-test-ownership into main
2 parents 19e7853 + 4006cb5 commit 1f24dbe

12 files changed

Lines changed: 590 additions & 33 deletions

File tree

‎.github/workflows/session-mongodb.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,4 +55,4 @@ jobs:
5555
run: .github/scripts/start-session-mongo-replset.sh 27017
5656

5757
- name: Run MongoDB driver conformance
58-
run: cargo test --manifest-path integration-proofs/run-ledger-mongodb/Cargo.toml --features mongodb-storage-tests --test session_driver_conformance -- --ignored
58+
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‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -792,11 +792,11 @@ where
792792

793793
let short = key("g", "n", "short");
794794
cache
795-
.put(&short, json!(1), Some(Duration::from_millis(30)))
795+
.put(&short, json!(1), Some(Duration::from_secs(2)))
796796
.await
797797
.unwrap();
798798
assert_eq!(cache.get(&short).await.unwrap(), Some(json!(1)));
799-
tokio::time::sleep(Duration::from_millis(80)).await;
799+
tokio::time::sleep(Duration::from_millis(2_100)).await;
800800
assert_eq!(cache.get(&short).await.unwrap(), None, "expired after TTL");
801801
assert_eq!(cache.get(&short).await.unwrap(), None, "stays expired");
802802
cache.put(&short, json!(2), None).await.unwrap();

‎crates/tinyagents-harness-tools/README.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,8 @@ run-state middleware and its required `goals` and `todos` modules.
2929
`tinyagents_orchestration::subagent::SubAgentJobRegistry`.
3030
- `thread_locks` provides the public `ThreadLockMap` concurrency primitive.
3131
- The crate root exports `CurrentTimeTool`, `ResolveTimeTool`,
32-
`AskClarificationTool`, `PlanExitTool`, and the `per_call` binding adapter.
32+
`AskClarificationTool`, `PlanExitTool`, `WaitTool`, `WaitLoopTool`, and the
33+
`per_call` binding adapter.
3334

3435
Subagent registry policy stays in `tinyagents-orchestration`; this crate owns
3536
the reusable tool declarations. The harness owns tool registration,

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ mod plan_exit;
1212
pub mod thread_locks;
1313
mod time;
1414
mod time_parse;
15+
mod wait;
1516

1617
#[cfg(feature = "goals")]
1718
pub mod goals;
@@ -29,6 +30,7 @@ pub mod todos;
2930
pub use ask_clarification::AskClarificationTool;
3031
pub use plan_exit::{PLAN_EXIT_MARKER, PlanExitModeLookup, PlanExitTool};
3132
pub use time::{CurrentTimeTool, ResolveTimeTool, register_time_tools, time_tools};
33+
pub use wait::{WaitLoopTool, WaitTool};
3234

3335
#[cfg(test)]
3436
#[path = "time_parse_tests.rs"]
Lines changed: 302 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,302 @@
1+
//! Tool: `wait` / `wait_loop` - delayed callback ticks for the orchestrator.
2+
//!
3+
//! These tools intentionally do not own scheduling state. They sleep for a
4+
//! bounded duration, then return the caller-provided message as a tool result so
5+
//! the orchestrator can decide whether to act, poll async sub-agents, or call the
6+
//! loop variant again.
7+
8+
use std::time::Duration;
9+
10+
use async_trait::async_trait;
11+
use serde_json::json;
12+
use tinytools::{PermissionLevel, Tool, ToolResult, ToolTimeout};
13+
14+
const DEFAULT_DURATION_SECS: u64 = 5;
15+
const MAX_DURATION_SECS: u64 = 600;
16+
const MILLIS_PER_SEC: u64 = 1_000;
17+
18+
/// One-shot delayed callback tool for the orchestrator.
19+
pub struct WaitTool;
20+
/// Repeatable delayed callback tool for orchestrator-controlled polling loops.
21+
pub struct WaitLoopTool;
22+
23+
impl WaitTool {
24+
pub fn new() -> Self {
25+
Self
26+
}
27+
}
28+
29+
impl Default for WaitTool {
30+
fn default() -> Self {
31+
Self::new()
32+
}
33+
}
34+
35+
impl WaitLoopTool {
36+
pub fn new() -> Self {
37+
Self
38+
}
39+
}
40+
41+
impl Default for WaitLoopTool {
42+
fn default() -> Self {
43+
Self::new()
44+
}
45+
}
46+
47+
#[async_trait]
48+
impl Tool for WaitTool {
49+
fn name(&self) -> &str {
50+
"wait"
51+
}
52+
53+
fn description(&self) -> &str {
54+
"Wait for a bounded duration, then return the provided callback message \
55+
to the orchestrator. Use this to create a delayed tick before checking \
56+
async work or retrying a condition."
57+
}
58+
59+
fn parameters_schema(&self) -> serde_json::Value {
60+
wait_schema(false)
61+
}
62+
63+
fn permission_level(&self) -> PermissionLevel {
64+
PermissionLevel::None
65+
}
66+
67+
fn timeout_policy(&self, args: &serde_json::Value) -> ToolTimeout {
68+
timeout_policy_for_wait(args)
69+
}
70+
71+
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
72+
execute_wait(args, false).await
73+
}
74+
}
75+
76+
#[async_trait]
77+
impl Tool for WaitLoopTool {
78+
fn name(&self) -> &str {
79+
"wait_loop"
80+
}
81+
82+
fn description(&self) -> &str {
83+
"Wait for a bounded duration, then return the same callback message plus \
84+
a ready-to-call wait_loop instruction. Use this for deliberate polling \
85+
loops where the orchestrator decides after each tick whether to repeat."
86+
}
87+
88+
fn parameters_schema(&self) -> serde_json::Value {
89+
wait_schema(true)
90+
}
91+
92+
fn permission_level(&self) -> PermissionLevel {
93+
PermissionLevel::None
94+
}
95+
96+
fn timeout_policy(&self, args: &serde_json::Value) -> ToolTimeout {
97+
timeout_policy_for_wait(args)
98+
}
99+
100+
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
101+
execute_wait(args, true).await
102+
}
103+
}
104+
105+
/// Sleep for the requested duration, then return the callback tick payload.
106+
async fn execute_wait(args: serde_json::Value, loop_mode: bool) -> anyhow::Result<ToolResult> {
107+
let request = match parse_wait_request(&args) {
108+
Ok(request) => request,
109+
Err(message) => return Ok(ToolResult::error(message)),
110+
};
111+
let tool_name = if loop_mode { "wait_loop" } else { "wait" };
112+
tracing::info!(
113+
"[wait] tool={} duration_ms={} iteration={} loop_key={} message_chars={}",
114+
tool_name,
115+
request.duration_ms,
116+
request.iteration,
117+
request.loop_key.as_deref().unwrap_or("none"),
118+
request.message.chars().count()
119+
);
120+
121+
tokio::time::sleep(Duration::from_millis(request.duration_ms)).await;
122+
123+
tracing::debug!(
124+
"[wait] elapsed tool={} duration_ms={} iteration={} loop_key={}",
125+
tool_name,
126+
request.duration_ms,
127+
request.iteration,
128+
request.loop_key.as_deref().unwrap_or("none")
129+
);
130+
131+
Ok(ToolResult::success(format_wait_tick(&request, loop_mode)))
132+
}
133+
134+
/// Build the JSON schema shared by `wait` and `wait_loop`.
135+
fn wait_schema(loop_mode: bool) -> serde_json::Value {
136+
let mut properties = serde_json::Map::new();
137+
properties.insert(
138+
"message".to_string(),
139+
json!({
140+
"type": "string",
141+
"description": "Callback message to return to the orchestrator after the wait elapses."
142+
}),
143+
);
144+
properties.insert(
145+
"duration_secs".to_string(),
146+
json!({
147+
"type": "integer",
148+
"minimum": 1,
149+
"maximum": MAX_DURATION_SECS,
150+
"description": "Seconds to wait before returning the callback. Default 5. Ignored when duration_ms is supplied."
151+
}),
152+
);
153+
properties.insert(
154+
"duration_ms".to_string(),
155+
json!({
156+
"type": "integer",
157+
"minimum": 1,
158+
"maximum": MAX_DURATION_SECS * MILLIS_PER_SEC,
159+
"description": "Milliseconds to wait before returning the callback. Use only for short test-sized waits."
160+
}),
161+
);
162+
163+
if loop_mode {
164+
properties.insert(
165+
"loop_key".to_string(),
166+
json!({
167+
"type": "string",
168+
"description": "Optional caller-defined key for the polling loop."
169+
}),
170+
);
171+
properties.insert(
172+
"iteration".to_string(),
173+
json!({
174+
"type": "integer",
175+
"minimum": 1,
176+
"description": "Current loop iteration. Defaults to 1; the returned instruction increments it."
177+
}),
178+
);
179+
}
180+
181+
json!({
182+
"type": "object",
183+
"required": ["message"],
184+
"properties": serde_json::Value::Object(properties)
185+
})
186+
}
187+
188+
/// Give the harness a deadline that covers the requested wait plus grace.
189+
fn timeout_policy_for_wait(args: &serde_json::Value) -> ToolTimeout {
190+
let duration_ms = duration_ms_from_args(args).unwrap_or(DEFAULT_DURATION_SECS * MILLIS_PER_SEC);
191+
let rounded_secs = duration_ms.saturating_add(MILLIS_PER_SEC - 1) / MILLIS_PER_SEC;
192+
ToolTimeout::Millis(
193+
rounded_secs
194+
.saturating_add(1)
195+
.saturating_mul(MILLIS_PER_SEC),
196+
)
197+
}
198+
199+
/// Extract and clamp the caller's requested wait duration in milliseconds.
200+
fn duration_ms_from_args(args: &serde_json::Value) -> Option<u64> {
201+
if let Some(duration_ms) = args.get("duration_ms").and_then(|v| v.as_u64()) {
202+
return Some(duration_ms.clamp(1, MAX_DURATION_SECS * MILLIS_PER_SEC));
203+
}
204+
args.get("duration_secs")
205+
.and_then(|v| v.as_u64())
206+
.map(|secs| secs.clamp(1, MAX_DURATION_SECS) * MILLIS_PER_SEC)
207+
}
208+
209+
/// Validate tool arguments and normalize them into an internal request.
210+
fn parse_wait_request(args: &serde_json::Value) -> Result<WaitRequest, String> {
211+
let message = args
212+
.get("message")
213+
.and_then(|v| v.as_str())
214+
.unwrap_or("")
215+
.trim()
216+
.to_string();
217+
if message.is_empty() {
218+
return Err("wait: `message` is required".to_string());
219+
}
220+
221+
let duration_ms = duration_ms_from_args(args).unwrap_or(DEFAULT_DURATION_SECS * MILLIS_PER_SEC);
222+
let iteration = args
223+
.get("iteration")
224+
.and_then(|v| v.as_u64())
225+
.unwrap_or(1)
226+
.max(1);
227+
let loop_key = args
228+
.get("loop_key")
229+
.and_then(|v| v.as_str())
230+
.map(str::trim)
231+
.filter(|s| !s.is_empty())
232+
.map(str::to_string);
233+
234+
Ok(WaitRequest {
235+
message,
236+
duration_ms,
237+
iteration,
238+
loop_key,
239+
})
240+
}
241+
242+
/// Render the delayed callback as prose plus a machine-readable `[wait_tick]`.
243+
fn format_wait_tick(request: &WaitRequest, loop_mode: bool) -> String {
244+
let seconds = request.duration_ms as f64 / MILLIS_PER_SEC as f64;
245+
let loop_instruction = loop_mode.then(|| {
246+
let mut arguments = json!({
247+
"message": request.message,
248+
"duration_ms": request.duration_ms,
249+
"iteration": request.iteration.saturating_add(1)
250+
});
251+
// The schema allows `loop_key` only as a string: omit it rather than
252+
// hand back a ready-to-call instruction carrying `null`.
253+
if let Some(loop_key) = &request.loop_key {
254+
arguments["loop_key"] = json!(loop_key);
255+
}
256+
json!({ "tool": "wait_loop", "arguments": arguments })
257+
});
258+
let payload = json!({
259+
"status": "elapsed",
260+
"message": request.message,
261+
"duration_ms": request.duration_ms,
262+
"durationMs": request.duration_ms,
263+
"duration_secs": seconds,
264+
"durationSecs": seconds,
265+
"loop": loop_mode,
266+
"loop_key": request.loop_key,
267+
"loopKey": request.loop_key,
268+
"iteration": request.iteration,
269+
"instructions": {
270+
"callback_message": request.message,
271+
"repeat": loop_instruction
272+
}
273+
});
274+
275+
let prefix = if loop_mode {
276+
format!(
277+
"Loop tick {} elapsed after {}ms.",
278+
request.iteration, request.duration_ms
279+
)
280+
} else {
281+
format!("Wait elapsed after {}ms.", request.duration_ms)
282+
};
283+
284+
format!(
285+
"{prefix}\n\n[wait_tick]\n{}\n[/wait_tick]\n\n{}",
286+
serde_json::to_string(&payload).unwrap_or_else(|_| "{}".to_string()),
287+
request.message
288+
)
289+
}
290+
291+
#[derive(Debug)]
292+
/// Normalized wait arguments used by both wait tools.
293+
struct WaitRequest {
294+
message: String,
295+
duration_ms: u64,
296+
iteration: u64,
297+
loop_key: Option<String>,
298+
}
299+
300+
#[cfg(test)]
301+
#[path = "wait_tests.rs"]
302+
mod tests;

0 commit comments

Comments
 (0)