feat: add config-owned direct submit runtime

Keep browser-attached workflows on the configured direct-skill path and align the Zhihu export/browser regression contracts with the current ws merge state.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
木炎
2026-04-11 15:45:42 +08:00
29 changed files with 5218 additions and 585 deletions

View File

@@ -7,15 +7,18 @@ use std::sync::{Arc, Mutex, OnceLock};
use std::thread;
use std::time::Duration;
use common::MockTransport;
use serde_json::{json, Value};
use uuid::Uuid;
use common::MockTransport;
use sgclaw::agent::{
handle_browser_message, handle_browser_message_with_context, AgentRuntimeContext,
};
use sgclaw::pipe::{AgentMessage, BrowserMessage, BrowserPipeTool, Timing};
use sgclaw::compat::runtime::CompatTaskContext;
use sgclaw::config::SgClawSettings;
use sgclaw::pipe::{Action, AgentMessage, BrowserMessage, BrowserPipeTool, Timing};
use sgclaw::security::MacPolicy;
use tungstenite::{accept, Message};
use uuid::Uuid;
use tungstenite::{accept, error::ProtocolError, Message};
fn env_lock() -> &'static Mutex<()> {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
@@ -34,6 +37,7 @@ fn write_config(
base_url: &str,
model: &str,
skills_dir: Option<&str>,
browser_ws_url: Option<&str>,
) -> PathBuf {
let config_path = root.join("sgclaw_config.json");
let mut payload = json!({
@@ -45,6 +49,9 @@ fn write_config(
if let Some(skills_dir) = skills_dir {
payload["skillsDir"] = json!(skills_dir);
}
if let Some(browser_ws_url) = browser_ws_url {
payload["browserWsUrl"] = json!(browser_ws_url);
}
fs::write(&config_path, serde_json::to_string_pretty(&payload).unwrap()).unwrap();
config_path
}
@@ -80,7 +87,10 @@ fn start_browser_ws_server() -> (String, Arc<Mutex<Vec<String>>>, thread::JoinHa
let message = match socket.read() {
Ok(message) => message,
Err(tungstenite::Error::ConnectionClosed)
| Err(tungstenite::Error::AlreadyClosed) => break,
| Err(tungstenite::Error::AlreadyClosed)
| Err(tungstenite::Error::Protocol(
ProtocolError::ResetWithoutClosingHandshake,
)) => break,
Err(err) => panic!("browser ws test server read failed: {err}"),
};
let payload = match message {
@@ -155,20 +165,567 @@ fn start_browser_ws_server() -> (String, Arc<Mutex<Vec<String>>>, thread::JoinHa
(format!("ws://{address}"), frames, handle)
}
#[test]
fn browser_ws_server_treats_reset_without_closing_handshake_as_disconnect() {
let err = tungstenite::Error::Protocol(ProtocolError::ResetWithoutClosingHandshake);
assert!(matches!(
err,
tungstenite::Error::Protocol(ProtocolError::ResetWithoutClosingHandshake)
));
}
fn provider_path_test_policy() -> MacPolicy {
policy_for_domains(&["www.baidu.com"])
}
fn direct_runtime_test_policy() -> MacPolicy {
policy_for_domains(&["95598.sgcc.com.cn"])
}
fn test_policy() -> MacPolicy {
policy_for_domains(&["www.zhihu.com"])
}
fn policy_for_domains(domains: &[&str]) -> MacPolicy {
MacPolicy::from_json_str(
r#"{
&serde_json::json!({
"version": "1.0",
"domains": { "allowed": ["www.baidu.com", "www.zhihu.com"] },
"domains": { "allowed": domains },
"pipe_actions": {
"allowed": ["click", "type", "navigate", "getText", "eval"],
"blocked": []
}
}"#,
})
.to_string(),
)
.unwrap()
}
fn build_direct_runtime_skill_root() -> PathBuf {
let root = std::env::temp_dir().join(format!(
"sgclaw-agent-runtime-skill-root-{}",
Uuid::new_v4()
));
let skill_dir = root.join("fault-details-report");
let script_dir = skill_dir.join("scripts");
fs::create_dir_all(&script_dir).unwrap();
fs::write(
skill_dir.join("SKILL.toml"),
r#"
[skill]
name = "fault-details-report"
description = "Collect 95598 fault detail data via browser eval."
version = "0.1.0"
[[tools]]
name = "collect_fault_details"
description = "Collect structured fault detail rows for a specific period."
kind = "browser_script"
command = "scripts/collect_fault_details.js"
[tools.args]
period = "YYYY-MM period to collect."
"#,
)
.unwrap();
fs::write(
script_dir.join("collect_fault_details.js"),
r#"
return {
fault_type: "outage",
observed_at: `${args.period}-15 09:00`,
affected_scope: "line-7",
expected_domain: args.expected_domain,
artifact_payload: "report artifact payload"
};
"#,
)
.unwrap();
root
}
fn write_direct_submit_config(workspace_root: &std::path::Path, skill_root: &std::path::Path) -> PathBuf {
let config_path = workspace_root.join("sgclaw_config.json");
fs::write(
&config_path,
serde_json::json!({
"providers": [],
"skillsDir": skill_root,
"directSubmitSkill": "fault-details-report.collect_fault_details"
})
.to_string(),
)
.unwrap();
config_path
}
fn direct_submit_runtime_context(skill_root: &std::path::Path) -> AgentRuntimeContext {
let workspace_root = std::env::temp_dir().join(format!(
"sgclaw-agent-runtime-workspace-{}",
Uuid::new_v4()
));
fs::create_dir_all(&workspace_root).unwrap();
let config_path = write_direct_submit_config(&workspace_root, skill_root);
AgentRuntimeContext::new(Some(config_path), workspace_root)
}
fn submit_fault_details_message() -> BrowserMessage {
BrowserMessage::SubmitTask {
instruction: "请采集 2026-03 的故障明细并返回结果".to_string(),
conversation_id: String::new(),
messages: vec![],
page_url: "https://95598.sgcc.com.cn/".to_string(),
page_title: "网上国网".to_string(),
}
}
fn direct_submit_mode_logs(sent: &[AgentMessage]) -> Vec<String> {
sent.iter()
.filter_map(|message| match message {
AgentMessage::LogEntry { level, message } if level == "mode" => Some(message.clone()),
_ => None,
})
.collect()
}
fn direct_submit_completion(sent: &[AgentMessage]) -> Option<(bool, String)> {
sent.iter().find_map(|message| match message {
AgentMessage::TaskComplete { success, summary } => Some((*success, summary.clone())),
_ => None,
})
}
fn success_browser_response(seq: u64, data: serde_json::Value) -> BrowserMessage {
BrowserMessage::Response {
seq,
success: true,
data,
aom_snapshot: vec![],
timing: Timing {
queue_ms: 1,
exec_ms: 10,
},
}
}
fn report_artifact_browser_response(
seq: u64,
status: &str,
partial_reasons: &[&str],
detail_rows: Vec<serde_json::Value>,
summary_rows: Vec<serde_json::Value>,
) -> BrowserMessage {
success_browser_response(
seq,
serde_json::json!({
"text": {
"type": "report-artifact",
"report_name": "fault-details-report",
"period": "2026-03",
"selected_range": {
"start": "2026-03-08 16:00:00",
"end": "2026-03-09 16:00:00"
},
"columns": ["qxdbh"],
"rows": detail_rows,
"sections": [{
"name": "summary-sheet",
"columns": ["index"],
"rows": summary_rows
}],
"counts": {
"detail_rows": detail_rows.len(),
"summary_rows": summary_rows.len()
},
"status": status,
"partial_reasons": partial_reasons,
"downstream": {
"export": {
"attempted": true,
"success": status != "blocked" && status != "error",
"path": "http://localhost/export.xlsx"
},
"report_log": {
"attempted": true,
"success": partial_reasons.is_empty(),
"error": partial_reasons
.first()
.copied()
.unwrap_or("")
}
}
}
}),
)
}
#[test]
fn direct_submit_runtime_executes_fault_details_skill_without_provider_path() {
let skill_root = build_direct_runtime_skill_root();
let transport = Arc::new(MockTransport::new(vec![success_browser_response(
1,
serde_json::json!({
"text": {
"fault_type": "outage",
"observed_at": "2026-03-15 09:00",
"affected_scope": "line-7"
}
}),
)]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
direct_runtime_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
let mut settings = SgClawSettings::from_legacy_deepseek_fields(
"unused-key".to_string(),
"http://127.0.0.1:9".to_string(),
"unused-model".to_string(),
Some(skill_root.clone()),
)
.unwrap();
settings.direct_submit_skill = Some("fault-details-report.collect_fault_details".to_string());
let summary = sgclaw::compat::direct_skill_runtime::execute_direct_submit_skill(
browser_tool,
"请采集 2026-03 的故障明细并返回结果",
&CompatTaskContext {
page_url: Some("https://95598.sgcc.com.cn/".to_string()),
..CompatTaskContext::default()
},
PathBuf::from(env!("CARGO_MANIFEST_DIR")).as_path(),
&settings,
)
.unwrap();
assert!(summary.success);
assert!(summary.summary.contains("fault_type"));
let sent = transport.sent_messages();
assert!(sent.iter().all(|message| !matches!(message, AgentMessage::LogEntry { level, message } if level == "info" && message.contains("DeepSeek config loaded"))));
assert!(matches!(
&sent[0],
AgentMessage::Command {
seq,
action,
params,
security,
} if *seq == 1
&& action == &Action::Eval
&& security.expected_domain == "95598.sgcc.com.cn"
&& params["script"].as_str().is_some_and(|script| script.contains("2026-03"))
));
}
#[test]
fn submit_task_uses_direct_skill_mode_without_llm_configuration() {
std::env::remove_var("DEEPSEEK_API_KEY");
std::env::remove_var("DEEPSEEK_BASE_URL");
std::env::remove_var("DEEPSEEK_MODEL");
let skill_root = build_direct_runtime_skill_root();
let runtime_context = direct_submit_runtime_context(&skill_root);
let transport = Arc::new(MockTransport::new(vec![success_browser_response(
1,
serde_json::json!({
"text": {
"fault_type": "outage",
"observed_at": "2026-03-15 09:00",
"affected_scope": "line-7",
"artifact_payload": "report artifact payload"
}
}),
)]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
direct_runtime_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
submit_fault_details_message(),
)
.unwrap();
let sent = transport.sent_messages();
let completion = direct_submit_completion(&sent).expect("task completion");
assert!(completion.0, "expected direct submit task to succeed: {sent:?}");
assert!(
completion.1.contains("report artifact payload"),
"expected report artifact payload in summary: {}",
completion.1
);
assert!(
!completion.1.contains("未配置大语言模型"),
"did not expect missing-llm summary: {}",
completion.1
);
}
#[test]
fn submit_task_rejects_invalid_direct_submit_skill_config_before_routing() {
std::env::remove_var("DEEPSEEK_API_KEY");
std::env::remove_var("DEEPSEEK_BASE_URL");
std::env::remove_var("DEEPSEEK_MODEL");
let skill_root = build_direct_runtime_skill_root();
let workspace_root = std::env::temp_dir().join(format!(
"sgclaw-invalid-direct-submit-workspace-{}",
Uuid::new_v4()
));
fs::create_dir_all(&workspace_root).unwrap();
let config_path = workspace_root.join("sgclaw_config.json");
fs::write(
&config_path,
serde_json::json!({
"providers": [],
"skillsDir": skill_root,
"directSubmitSkill": "fault-details-report"
})
.to_string(),
)
.unwrap();
let runtime_context = AgentRuntimeContext::new(Some(config_path), workspace_root);
let transport = Arc::new(MockTransport::new(vec![]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
direct_runtime_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
submit_fault_details_message(),
)
.unwrap();
let sent = transport.sent_messages();
assert!(matches!(
sent.last(),
Some(AgentMessage::TaskComplete { success, summary })
if !success && summary.contains("skill.tool")
));
assert!(direct_submit_mode_logs(&sent).is_empty());
assert!(!sent.iter().any(|message| matches!(message, AgentMessage::Command { .. })));
}
#[test]
fn submit_task_treats_partial_report_artifact_as_success_with_warning_summary() {
std::env::remove_var("DEEPSEEK_API_KEY");
std::env::remove_var("DEEPSEEK_BASE_URL");
std::env::remove_var("DEEPSEEK_MODEL");
let skill_root = build_direct_runtime_skill_root();
let runtime_context = direct_submit_runtime_context(&skill_root);
let transport = Arc::new(MockTransport::new(vec![report_artifact_browser_response(
1,
"partial",
&["report_log_failed"],
vec![serde_json::json!({ "qxdbh": "QX-1" })],
vec![serde_json::json!({ "index": 1 })],
)]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
direct_runtime_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
submit_fault_details_message(),
)
.unwrap();
let sent = transport.sent_messages();
let completion = direct_submit_completion(&sent).expect("task completion");
assert!(completion.0, "expected partial artifact to succeed: {sent:?}");
assert!(completion.1.contains("fault-details-report"));
assert!(completion.1.contains("2026-03"));
assert!(completion.1.contains("status=partial"));
assert!(completion.1.contains("detail_rows=1"));
assert!(completion.1.contains("summary_rows=1"));
assert!(completion.1.contains("report_log_failed"));
}
#[test]
fn submit_task_treats_empty_report_artifact_as_success() {
std::env::remove_var("DEEPSEEK_API_KEY");
std::env::remove_var("DEEPSEEK_BASE_URL");
std::env::remove_var("DEEPSEEK_MODEL");
let skill_root = build_direct_runtime_skill_root();
let runtime_context = direct_submit_runtime_context(&skill_root);
let transport = Arc::new(MockTransport::new(vec![report_artifact_browser_response(
1,
"empty",
&[],
vec![],
vec![],
)]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
direct_runtime_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
submit_fault_details_message(),
)
.unwrap();
let sent = transport.sent_messages();
let completion = direct_submit_completion(&sent).expect("task completion");
assert!(completion.0, "expected empty artifact to succeed: {sent:?}");
assert!(completion.1.contains("status=empty"));
assert!(completion.1.contains("detail_rows=0"));
}
#[test]
fn submit_task_treats_blocked_report_artifact_as_failure() {
std::env::remove_var("DEEPSEEK_API_KEY");
std::env::remove_var("DEEPSEEK_BASE_URL");
std::env::remove_var("DEEPSEEK_MODEL");
let skill_root = build_direct_runtime_skill_root();
let runtime_context = direct_submit_runtime_context(&skill_root);
let transport = Arc::new(MockTransport::new(vec![report_artifact_browser_response(
1,
"blocked",
&["selected_range_unavailable"],
vec![],
vec![],
)]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
direct_runtime_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
submit_fault_details_message(),
)
.unwrap();
let sent = transport.sent_messages();
let completion = direct_submit_completion(&sent).expect("task completion");
assert!(!completion.0, "expected blocked artifact to fail: {sent:?}");
assert!(completion.1.contains("status=blocked"));
assert!(completion.1.contains("selected_range_unavailable"));
}
#[test]
fn submit_task_treats_error_report_artifact_as_failure() {
std::env::remove_var("DEEPSEEK_API_KEY");
std::env::remove_var("DEEPSEEK_BASE_URL");
std::env::remove_var("DEEPSEEK_MODEL");
let skill_root = build_direct_runtime_skill_root();
let runtime_context = direct_submit_runtime_context(&skill_root);
let transport = Arc::new(MockTransport::new(vec![report_artifact_browser_response(
1,
"error",
&["detail_normalization_failed"],
vec![],
vec![],
)]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
direct_runtime_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
submit_fault_details_message(),
)
.unwrap();
let sent = transport.sent_messages();
let completion = direct_submit_completion(&sent).expect("task completion");
assert!(!completion.0, "expected error artifact to fail: {sent:?}");
assert!(completion.1.contains("status=error"));
assert!(completion.1.contains("detail_normalization_failed"));
}
#[test]
fn direct_skill_mode_logs_direct_skill_primary() {
std::env::remove_var("DEEPSEEK_API_KEY");
std::env::remove_var("DEEPSEEK_BASE_URL");
std::env::remove_var("DEEPSEEK_MODEL");
let skill_root = build_direct_runtime_skill_root();
let runtime_context = direct_submit_runtime_context(&skill_root);
let transport = Arc::new(MockTransport::new(vec![success_browser_response(
1,
serde_json::json!({
"text": {
"fault_type": "outage",
"observed_at": "2026-03-15 09:00",
"affected_scope": "line-7",
"artifact_payload": "report artifact payload"
}
}),
)]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
direct_runtime_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));
handle_browser_message_with_context(
transport.as_ref(),
&browser_tool,
&runtime_context,
submit_fault_details_message(),
)
.unwrap();
let sent = transport.sent_messages();
let mode_logs = direct_submit_mode_logs(&sent);
assert_eq!(mode_logs, vec!["direct_skill_primary".to_string()]);
assert!(
!mode_logs.iter().any(|mode| mode == "compat_llm_primary"),
"unexpected compat mode logs: {mode_logs:?}"
);
assert!(
!mode_logs
.iter()
.any(|mode| mode == "zeroclaw_process_message_primary"),
"unexpected zeroclaw mode logs: {mode_logs:?}"
);
}
#[test]
fn production_submit_task_routes_zhihu_through_ws_backend_without_helper_bootstrap() {
let _guard = env_lock().lock().unwrap_or_else(|err| err.into_inner());
@@ -179,17 +736,16 @@ fn production_submit_task_routes_zhihu_through_ws_backend_without_helper_bootstr
std::env::remove_var("DEEPSEEK_MODEL");
let workspace_root = temp_workspace_root();
let (ws_url, frames, ws_handle) = start_browser_ws_server();
let config_path = write_config(
&workspace_root,
"deepseek-test-key",
"http://127.0.0.1:9",
"deepseek-chat",
Some(real_skill_lib_root().to_str().unwrap()),
Some(&ws_url),
);
let (ws_url, frames, ws_handle) = start_browser_ws_server();
std::env::set_var("SGCLAW_BROWSER_WS_URL", &ws_url);
let transport = Arc::new(MockTransport::new(vec![]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
@@ -306,7 +862,7 @@ fn production_submit_task_does_not_route_into_legacy_runtime_without_llm_config(
let transport = Arc::new(MockTransport::new(vec![]));
let browser_tool = BrowserPipeTool::new(
transport.clone(),
test_policy(),
provider_path_test_policy(),
vec![1, 2, 3, 4, 5, 6, 7, 8],
)
.with_response_timeout(Duration::from_secs(1));