fix(runtime): deep audit fixes — clarification loop termination + callback alignment
Some checks failed
CI / Lint & TypeCheck (push) Has been cancelled
CI / Unit Tests (push) Has been cancelled
CI / Build Frontend (push) Has been cancelled
CI / Rust Check (push) Has been cancelled
CI / Security Scan (push) Has been cancelled
CI / E2E Tests (push) Has been cancelled
Some checks failed
CI / Lint & TypeCheck (push) Has been cancelled
CI / Unit Tests (push) Has been cancelled
CI / Build Frontend (push) Has been cancelled
CI / Rust Check (push) Has been cancelled
CI / Security Scan (push) Has been cancelled
CI / E2E Tests (push) Has been cancelled
CRITICAL: - ask_clarification now terminates Agent Loop in both run() and run_streaming() paths, preventing the LLM from continuing after requesting user clarification HIGH: - SaaS relay now forwards plan_mode and subagent_enabled to backend - GatewayClient.chatStream now supports onThinkingDelta, onSubtaskStatus, and token-bearing onComplete — aligned with kernel-types StreamCallbacks - ZclawStreamEvent type extended with thinking_delta, subtask_status variants and input_tokens/output_tokens fields for token tracking via Gateway path
This commit is contained in:
@@ -402,6 +402,7 @@ impl AgentLoop {
|
|||||||
let tool_context = self.create_tool_context(session_id.clone());
|
let tool_context = self.create_tool_context(session_id.clone());
|
||||||
let mut circuit_breaker_triggered = false;
|
let mut circuit_breaker_triggered = false;
|
||||||
let mut abort_result: Option<AgentLoopResult> = None;
|
let mut abort_result: Option<AgentLoopResult> = None;
|
||||||
|
let mut clarification_result: Option<AgentLoopResult> = None;
|
||||||
for (id, name, input) in tool_calls {
|
for (id, name, input) in tool_calls {
|
||||||
// Check if loop was already aborted
|
// Check if loop was already aborted
|
||||||
if abort_result.is_some() {
|
if abort_result.is_some() {
|
||||||
@@ -475,6 +476,32 @@ impl AgentLoop {
|
|||||||
Err(e) => serde_json::json!({ "error": e.to_string() }),
|
Err(e) => serde_json::json!({ "error": e.to_string() }),
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// Check if this is a clarification response — terminate loop immediately
|
||||||
|
// so the LLM waits for user input instead of continuing to generate.
|
||||||
|
if name == "ask_clarification"
|
||||||
|
&& tool_result.get("status").and_then(|v| v.as_str()) == Some("clarification_needed")
|
||||||
|
{
|
||||||
|
tracing::info!("[AgentLoop] Clarification requested, terminating loop");
|
||||||
|
let question = tool_result.get("question")
|
||||||
|
.and_then(|v| v.as_str())
|
||||||
|
.unwrap_or("需要更多信息")
|
||||||
|
.to_string();
|
||||||
|
messages.push(Message::tool_result(
|
||||||
|
id,
|
||||||
|
zclaw_types::ToolId::new(&name),
|
||||||
|
tool_result,
|
||||||
|
false,
|
||||||
|
));
|
||||||
|
self.memory.append_message(&session_id, &Message::assistant(&question)).await?;
|
||||||
|
clarification_result = Some(AgentLoopResult {
|
||||||
|
response: question,
|
||||||
|
input_tokens: total_input_tokens,
|
||||||
|
output_tokens: total_output_tokens,
|
||||||
|
iterations,
|
||||||
|
});
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
// Add tool result to messages
|
// Add tool result to messages
|
||||||
messages.push(Message::tool_result(
|
messages.push(Message::tool_result(
|
||||||
id,
|
id,
|
||||||
@@ -491,6 +518,11 @@ impl AgentLoop {
|
|||||||
break result;
|
break result;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// If clarification was requested, return immediately
|
||||||
|
if let Some(result) = clarification_result {
|
||||||
|
break result;
|
||||||
|
}
|
||||||
|
|
||||||
// If circuit breaker was triggered, terminate immediately
|
// If circuit breaker was triggered, terminate immediately
|
||||||
if circuit_breaker_triggered {
|
if circuit_breaker_triggered {
|
||||||
let msg = "检测到工具调用循环,已自动终止";
|
let msg = "检测到工具调用循环,已自动终止";
|
||||||
@@ -972,6 +1004,35 @@ impl AgentLoop {
|
|||||||
(error_output, true)
|
(error_output, true)
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// Check if this is a clarification response — break outer loop
|
||||||
|
if name == "ask_clarification"
|
||||||
|
&& result.get("status").and_then(|v| v.as_str()) == Some("clarification_needed")
|
||||||
|
{
|
||||||
|
tracing::info!("[AgentLoop] Streaming: Clarification requested, terminating loop");
|
||||||
|
let question = result.get("question")
|
||||||
|
.and_then(|v| v.as_str())
|
||||||
|
.unwrap_or("需要更多信息")
|
||||||
|
.to_string();
|
||||||
|
messages.push(Message::tool_result(
|
||||||
|
id,
|
||||||
|
zclaw_types::ToolId::new(&name),
|
||||||
|
result,
|
||||||
|
is_error,
|
||||||
|
));
|
||||||
|
// Send the question as final delta so the user sees it
|
||||||
|
let _ = tx.send(LoopEvent::Delta(question.clone())).await;
|
||||||
|
let _ = tx.send(LoopEvent::Complete(AgentLoopResult {
|
||||||
|
response: question.clone(),
|
||||||
|
input_tokens: total_input_tokens,
|
||||||
|
output_tokens: total_output_tokens,
|
||||||
|
iterations: iteration,
|
||||||
|
})).await;
|
||||||
|
if let Err(e) = memory.append_message(&session_id_clone, &Message::assistant(&question)).await {
|
||||||
|
tracing::warn!("[AgentLoop] Failed to save clarification message: {}", e);
|
||||||
|
}
|
||||||
|
break 'outer;
|
||||||
|
}
|
||||||
|
|
||||||
// Add tool result to message history
|
// Add tool result to message history
|
||||||
tracing::debug!("[AgentLoop] Adding tool_result to history: id={}, name={}, is_error={}", id, name, is_error);
|
tracing::debug!("[AgentLoop] Adding tool_result to history: id={}, name={}, is_error={}", id, name, is_error);
|
||||||
messages.push(Message::tool_result(
|
messages.push(Message::tool_result(
|
||||||
|
|||||||
@@ -175,9 +175,11 @@ export class GatewayClient {
|
|||||||
private deviceKeysPromise: Promise<DeviceKeys>;
|
private deviceKeysPromise: Promise<DeviceKeys>;
|
||||||
private streamCallbacks = new Map<string, {
|
private streamCallbacks = new Map<string, {
|
||||||
onDelta: (delta: string) => void;
|
onDelta: (delta: string) => void;
|
||||||
|
onThinkingDelta?: (delta: string) => void;
|
||||||
onTool?: (tool: string, input: string, output: string) => void;
|
onTool?: (tool: string, input: string, output: string) => void;
|
||||||
onHand?: (name: string, status: string, result?: unknown) => void;
|
onHand?: (name: string, status: string, result?: unknown) => void;
|
||||||
onComplete: () => void;
|
onSubtaskStatus?: (description: string, status: string, detail?: string) => void;
|
||||||
|
onComplete: (inputTokens?: number, outputTokens?: number) => void;
|
||||||
onError: (error: string) => void;
|
onError: (error: string) => void;
|
||||||
}>();
|
}>();
|
||||||
|
|
||||||
@@ -465,9 +467,11 @@ export class GatewayClient {
|
|||||||
message: string,
|
message: string,
|
||||||
callbacks: {
|
callbacks: {
|
||||||
onDelta: (delta: string) => void;
|
onDelta: (delta: string) => void;
|
||||||
|
onThinkingDelta?: (delta: string) => void;
|
||||||
onTool?: (tool: string, input: string, output: string) => void;
|
onTool?: (tool: string, input: string, output: string) => void;
|
||||||
onHand?: (name: string, status: string, result?: unknown) => void;
|
onHand?: (name: string, status: string, result?: unknown) => void;
|
||||||
onComplete: () => void;
|
onSubtaskStatus?: (description: string, status: string, detail?: string) => void;
|
||||||
|
onComplete: (inputTokens?: number, outputTokens?: number) => void;
|
||||||
onError: (error: string) => void;
|
onError: (error: string) => void;
|
||||||
},
|
},
|
||||||
opts?: {
|
opts?: {
|
||||||
@@ -638,10 +642,26 @@ export class GatewayClient {
|
|||||||
}
|
}
|
||||||
break;
|
break;
|
||||||
|
|
||||||
|
case 'thinking_delta':
|
||||||
|
// Extended thinking delta
|
||||||
|
if (data.content && callbacks.onThinkingDelta) {
|
||||||
|
callbacks.onThinkingDelta(data.content);
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
|
||||||
|
case 'subtask_status':
|
||||||
|
// Sub-agent task status update
|
||||||
|
if (callbacks.onSubtaskStatus && data.description) {
|
||||||
|
callbacks.onSubtaskStatus(data.description, data.status || '', data.detail);
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
|
||||||
case 'phase':
|
case 'phase':
|
||||||
// Phase change: streaming | done
|
// Phase change: streaming | done
|
||||||
if (data.phase === 'done') {
|
if (data.phase === 'done') {
|
||||||
callbacks.onComplete();
|
const inputTokens = typeof data.input_tokens === 'number' ? data.input_tokens : undefined;
|
||||||
|
const outputTokens = typeof data.output_tokens === 'number' ? data.output_tokens : undefined;
|
||||||
|
callbacks.onComplete(inputTokens, outputTokens);
|
||||||
this.streamCallbacks.delete(runId);
|
this.streamCallbacks.delete(runId);
|
||||||
if (this.zclawWs) {
|
if (this.zclawWs) {
|
||||||
this.zclawWs.close(1000, 'Stream complete');
|
this.zclawWs.close(1000, 'Stream complete');
|
||||||
@@ -657,7 +677,11 @@ export class GatewayClient {
|
|||||||
callbacks.onDelta(data.content);
|
callbacks.onDelta(data.content);
|
||||||
}
|
}
|
||||||
// Mark complete if phase done wasn't sent
|
// Mark complete if phase done wasn't sent
|
||||||
callbacks.onComplete();
|
{
|
||||||
|
const inputTokens = typeof data.input_tokens === 'number' ? data.input_tokens : undefined;
|
||||||
|
const outputTokens = typeof data.output_tokens === 'number' ? data.output_tokens : undefined;
|
||||||
|
callbacks.onComplete(inputTokens, outputTokens);
|
||||||
|
}
|
||||||
this.streamCallbacks.delete(runId);
|
this.streamCallbacks.delete(runId);
|
||||||
if (this.zclawWs) {
|
if (this.zclawWs) {
|
||||||
this.zclawWs.close(1000, 'Stream complete');
|
this.zclawWs.close(1000, 'Stream complete');
|
||||||
|
|||||||
@@ -68,7 +68,7 @@ export interface AgentStreamDelta {
|
|||||||
|
|
||||||
/** ZCLAW WebSocket stream event types */
|
/** ZCLAW WebSocket stream event types */
|
||||||
export interface ZclawStreamEvent {
|
export interface ZclawStreamEvent {
|
||||||
type: 'text_delta' | 'phase' | 'response' | 'typing' | 'tool_call' | 'tool_result' | 'hand' | 'workflow' | 'error' | 'connected' | 'agents_updated';
|
type: 'text_delta' | 'thinking_delta' | 'subtask_status' | 'phase' | 'response' | 'typing' | 'tool_call' | 'tool_result' | 'hand' | 'workflow' | 'error' | 'connected' | 'agents_updated';
|
||||||
content?: string;
|
content?: string;
|
||||||
phase?: 'streaming' | 'done';
|
phase?: 'streaming' | 'done';
|
||||||
state?: 'start' | 'stop';
|
state?: 'start' | 'stop';
|
||||||
@@ -87,6 +87,13 @@ export interface ZclawStreamEvent {
|
|||||||
code?: string;
|
code?: string;
|
||||||
agent_id?: string;
|
agent_id?: string;
|
||||||
agents?: Array<{ id: string; name: string; status: string }>;
|
agents?: Array<{ id: string; name: string; status: string }>;
|
||||||
|
// Subtask status fields
|
||||||
|
description?: string;
|
||||||
|
status?: string;
|
||||||
|
detail?: string;
|
||||||
|
// Token tracking fields (present in phase=done and response events)
|
||||||
|
input_tokens?: number;
|
||||||
|
output_tokens?: number;
|
||||||
}
|
}
|
||||||
|
|
||||||
// === Connection State ===
|
// === Connection State ===
|
||||||
|
|||||||
@@ -125,6 +125,8 @@ export function createSaaSRelayGatewayClient(
|
|||||||
if (opts?.agentId) body['agent_id'] = opts.agentId;
|
if (opts?.agentId) body['agent_id'] = opts.agentId;
|
||||||
if (opts?.thinking_enabled) body['thinking_enabled'] = true;
|
if (opts?.thinking_enabled) body['thinking_enabled'] = true;
|
||||||
if (opts?.reasoning_effort) body['reasoning_effort'] = opts.reasoning_effort;
|
if (opts?.reasoning_effort) body['reasoning_effort'] = opts.reasoning_effort;
|
||||||
|
if (opts?.plan_mode) body['plan_mode'] = true;
|
||||||
|
if (opts?.subagent_enabled) body['subagent_enabled'] = true;
|
||||||
|
|
||||||
const response = await saasClient.chatCompletion(body);
|
const response = await saasClient.chatCompletion(body);
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user