feat: add tool orchestration, retry, streaming, spawning, and context compaction

[Desicool]

Summary

• Tool system — Tool trait + ToolRegistry with ReadOnly/Exclusive concurrency; ctx.tool_step() drives a full LLM-tool-use loop automatically
• Error recovery — RetryPolicy, ResilientLlmProvider with fallback providers, ctx.step_with_retry()
• Streaming LLM — StreamingLlmProvider trait + LlmStreamEvent; OpenAiProvider implements SSE streaming via reqwest bytes stream
• Sub-workflow spawning — ctx.spawn_child() / ctx.await_children(); parent suspends as Waiting with zero resource hold until children finish
• Token management — TokenCounter, ContextConfig, ManagedConversation with TruncationCompaction and LlmSummaryCompaction
• Observability — ToolCallRecord and RetryRecord added to Observer trait
• Docs — README.md and README.zh.md updated with new capability sections and comparison table rows

New files (15)

File	Purpose
src/tool.rs	Tool trait + ToolRegistry
src/tool_loop.rs	run_tool_loop() — LLM ↔ tool execution loop
src/retry.rs	RetryPolicy, ErrorClass, with_retry()
src/llm_resilient.rs	ResilientLlmProvider with fallback + retry
src/spawn.rs	SpawnConfig, ChildHandle, ChildStatus, ChildrenResult
src/token.rs	TokenCounter, CharTokenCounter, ContextConfig
src/compact.rs	ManagedConversation, TruncationCompaction, LlmSummaryCompaction
src/sse.rs (openai)	SSE parser for OpenAI streaming completions
src/streaming.rs (openai)	StreamingLlmProvider impl for OpenAiProvider
tests/tool_tests.rs	8 tests
tests/tool_loop_tests.rs	4 tests
tests/retry_tests.rs	7 tests
tests/token_tests.rs	9 tests
tests/spawn_tests.rs + context_spawn_tests.rs	12 tests
tests/streaming_tests.rs (openai)	11 tests

Test plan

• All 211 workspace tests pass (cargo test --workspace)
• Zero clippy warnings (cargo clippy --workspace)
• TDD: test files written before implementation for all new modules

🤖 Generated with Claude Code

[Desicool]

Code review

Found 2 issues:

1. ToolCallRecord.duration_ms reports batch wall time for every individual tool, not per-tool execution time

In run_tool_loop, a single timer spans the entire concurrent batch, and the same batch_duration_ms value is stamped on every ToolCallRecord emitted in the loop. For batches of ReadOnly tools that run in parallel, each record claims the full wall time of the batch rather than its own execution time. The ToolCallRecord doc comment says "Record of a single tool call execution", implying individual timing.

Tianshu-rs/crates/workflow_engine/src/tool_loop.rs

Lines 90 to 116 in d2decd8

	// Execute tool calls with concurrency
	let start = Instant::now();
	let results = tools
	.execute_with_concurrency(&calls, config.max_concurrency)
	.await;
	let batch_duration_ms = start.elapsed().as_millis() as u64;
	// Emit observer events
	if let Some(obs) = observer {
	for (idx, result) in results.iter().enumerate() {
	let call = &calls[idx];
	let input: serde_json::Value =
	serde_json::from_str(&call.arguments).unwrap_or_default();
	let record = ToolCallRecord {
	case_key: String::new(),
	step_name: None,
	tool_name: call.name.clone(),
	call_id: call.id.clone(),
	input,
	output: Some(result.content.clone()),
	is_error: result.is_error,
	duration_ms: batch_duration_ms,
	timestamp: Utc::now(),
	};
	obs.on_tool_call(&record).await;
	}

2. ResilientLlmProvider re-tries already-failed fallbacks in the post-loop sweep

After with_retry exhausts max_attempts, the code at lines 71–78 sweeps all fallbacks unconditionally. But on ProviderOverloaded errors, fallbacks were already attempted inside the retry loop (one per attempt via fallback_idx = attempt - 1). The post-loop sweep re-tries those same providers, giving each fallback two chances while silently discarding their errors in the loop pass. This can mask the root cause: if fallbacks[0] fails fatally inside the loop, the sweep still tries it again outside.

Tianshu-rs/crates/workflow_engine/src/llm_resilient.rs

Lines 52 to 78 in d2decd8

	let result = with_retry(&self.retry_policy, |ctx: RetryContext| {
	let req = current_request.clone();
	let primary = self.primary.clone();
	let fallbacks = self.fallbacks.clone();
	async move {
	// On ProviderOverloaded from a previous attempt, try fallbacks
	if ctx.last_error_class == Some(ErrorClass::ProviderOverloaded) {
	let fallback_idx = (ctx.attempt as usize).saturating_sub(1);
	if fallback_idx < fallbacks.len() {
	return fallbacks[fallback_idx].complete(req).await;
	}
	}
	primary.complete(req).await
	}
	})
	.await;
	// If the retry loop exhausted with ProviderOverloaded and we have
	// untried fallbacks, give them a shot outside the loop
	if let Err(ref e) = result {
	let class = (self.retry_policy.classify)(e);
	if class == ErrorClass::ProviderOverloaded {
	for fb in &self.fallbacks {
	if let Ok(resp) = fb.complete(current_request.clone()).await {
	return Ok(resp);
	}
	}

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}