Files
Genarrative/apps/ai-game-creator-shell/src-tauri/src/runner/client.rs
T
kdletters 0594a90bdd
Project CI / AI game creator shell Rust shard 1/4 (push) Successful in 4m15s
Project CI / AI game creator shell Rust smoke (push) Successful in 1m32s
Project CI / AI game creator shell Rust shard 3/4 (push) Successful in 4m58s
Project CI / AI game creator shell Rust shard 2/4 (push) Successful in 6m13s
Project CI / AI game creator shell Rust shard 4/4 (push) Successful in 5m3s
Project CI / AI game creator shell Rust crates (push) Successful in 3m1s
Project CI / Frontend tests (push) Failing after 4m24s
Project CI / Repository checks (push) Successful in 4m1s
Project CI / Native shell tests (push) Successful in 7m45s
Project CI / AI game creator shell web tests (push) Failing after 3m4s
Project CI / Backend tests (push) Successful in 8m20s
合入平台会话身份与凭据分离
平台会话快照拆分身份代次与写入 revision,凭据续期轮换不再中断在途生成
AGC Runner 请求参数新增 platform_auth_revision,并按 revision 做单调写入判定
解决与多窗口共享 Runner 的冲突,保留 claim adopt/publish 与界面参与锁语义
同步刷新轮换竞争、appSurface 鉴权用例与客户端 API 测试
补录平台会话身份与凭据分离的实施计划、共享记忆与决策记录
2026-09-16 17:42:00 +08:00

2004 lines
75 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
use super::{dispatch::*, endpoint::*, project_owner::*, protocol::*, state::*};
use crate::{AgentRuntimeContextCompactionResult, GameCreatorManifestInvalidationEventSink};
use serde_json::Value;
use sha2::{Digest as _, Sha256};
use std::ffi::OsString;
use std::fs;
use std::io::{self, BufRead, BufReader, Read, Write};
use std::net::{Ipv4Addr, SocketAddrV4, TcpStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::{Mutex, OnceLock};
use std::thread;
use std::time::{Duration, Instant};
const AGENT_RUNNER_LOG_FILE_NAME: &str = "agent-runner.log";
const AGENT_RUNNER_LOG_INPUT_LINE_MAX_BYTES: usize = 8 * 1024;
const AGENT_RUNNER_LOG_OUTPUT_MAX_CHARS: usize = 1_024;
const AGENT_RUNNER_CLIENT_EXIT_TIMEOUT: Duration = Duration::from_secs(15);
#[derive(Default)]
pub(super) struct ExternalAgentRunnerGuiOwnerAttachmentState {
generation: u64,
registration: Option<ExternalAgentRunnerGuiOwnerRegistration>,
}
struct ExternalAgentRunnerGuiOwnerRegistration {
generation: u64,
claim_mode: ExternalAgentRunnerGuiOwnerClaimMode,
config_dir: PathBuf,
params: ExternalAgentRunnerRequestParams,
attached_boot_id: Option<String>,
}
/// claim 解析模式。
///
/// `Adopt` 用于窗口启动:沿用现有 durable claim,只有 claim 缺失或不可读时才发布。
/// `Publish` 用于本窗口改动了平台登录态:发布新 epoch,成为新的登录态权威。
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum ExternalAgentRunnerGuiOwnerClaimMode {
Adopt,
Publish,
}
static EXTERNAL_AGENT_RUNNER_GUI_OWNER_ATTACHMENT_STATE: OnceLock<
Mutex<ExternalAgentRunnerGuiOwnerAttachmentState>,
> = OnceLock::new();
static EXTERNAL_AGENT_RUNNER_GUI_PARTICIPANT_LOCK: OnceLock<
Mutex<Option<ExternalAgentRunnerGuiParticipantLock>>,
> = OnceLock::new();
fn external_agent_runner_gui_owner_attachment_state(
) -> &'static Mutex<ExternalAgentRunnerGuiOwnerAttachmentState> {
EXTERNAL_AGENT_RUNNER_GUI_OWNER_ATTACHMENT_STATE
.get_or_init(|| Mutex::new(ExternalAgentRunnerGuiOwnerAttachmentState::default()))
}
fn external_agent_runner_gui_participant_lock(
) -> &'static Mutex<Option<ExternalAgentRunnerGuiParticipantLock>> {
EXTERNAL_AGENT_RUNNER_GUI_PARTICIPANT_LOCK.get_or_init(|| Mutex::new(None))
}
/// 取得并持有本窗口的界面参与锁,直到窗口退出。
///
/// 参与锁是共享句柄锁:同一 AppData 的多个窗口可以同时持有,Runner 用独占探测
/// 判断是否仍有窗口存活,因此这个锁同时也是“Runner 不能先退出”的存活凭据。
pub(crate) fn hold_external_agent_runner_gui_participant_lock(
config_dir: &Path,
) -> Result<(), String> {
let lock = acquire_external_agent_runner_gui_participant_lock(config_dir)?;
*lock_unpoisoned(external_agent_runner_gui_participant_lock()) = Some(lock);
Ok(())
}
fn release_external_agent_runner_gui_participant_lock() {
drop(lock_unpoisoned(external_agent_runner_gui_participant_lock()).take());
}
/// 登记本窗口的 owner claim 与 attach 参数。
///
/// 这里不做 claim 文件 IO:claim 由调用方按 `claim_mode` 解析后写进 `params`,
/// 因此该函数可以在没有真实 AppData 的单元测试里使用。
pub(super) fn register_external_agent_runner_gui_owner_attachment(
state: &Mutex<ExternalAgentRunnerGuiOwnerAttachmentState>,
config_dir: &Path,
claim_mode: ExternalAgentRunnerGuiOwnerClaimMode,
mut params: ExternalAgentRunnerRequestParams,
) -> Result<(), String> {
let mut state = lock_unpoisoned(state);
state.generation = state.generation.wrapping_add(1);
let generation = state.generation;
if params.gui_owner_session_revision.is_none() {
params.gui_owner_session_revision = Some(generation);
}
state.registration = Some(ExternalAgentRunnerGuiOwnerRegistration {
generation,
claim_mode,
config_dir: config_dir.to_path_buf(),
params,
attached_boot_id: None,
});
Ok(())
}
pub(super) fn reserve_external_agent_runner_gui_owner_claim_revision(
state: &Mutex<ExternalAgentRunnerGuiOwnerAttachmentState>,
) -> u64 {
let mut state = lock_unpoisoned(state);
state.generation = state.generation.wrapping_add(1);
state.generation
}
pub(super) fn resolve_external_agent_runner_gui_owner_claim(
config_dir: &Path,
claim_mode: ExternalAgentRunnerGuiOwnerClaimMode,
session_revision: u64,
) -> Result<ExternalAgentRunnerGuiOwnerClaim, String> {
match claim_mode {
ExternalAgentRunnerGuiOwnerClaimMode::Adopt => {
adopt_or_publish_external_agent_runner_gui_owner_claim(config_dir, session_revision)
}
ExternalAgentRunnerGuiOwnerClaimMode::Publish => {
publish_external_agent_runner_gui_owner_claim(config_dir, session_revision)
}
}
}
pub(super) fn attach_registered_external_agent_runner_gui_owner_if_needed_with<F>(
state: &Mutex<ExternalAgentRunnerGuiOwnerAttachmentState>,
config_dir: &Path,
endpoint: &ExternalAgentRunnerEndpoint,
attach: F,
) -> Result<(), String>
where
F: Fn(&ExternalAgentRunnerEndpoint, ExternalAgentRunnerRequestParams) -> Result<(), String>,
{
const ATTACH_CLAIM_RETRY_LIMIT: usize = 3;
let mut last_claim_error = None;
for attempt in 0..ATTACH_CLAIM_RETRY_LIMIT {
let Some((generation, params, claim_mode)) = ({
let state = lock_unpoisoned(state);
state.registration.as_ref().and_then(|registration| {
(registration.config_dir == config_dir
&& registration.attached_boot_id.as_deref() != Some(endpoint.boot_id.as_str()))
.then(|| {
(
registration.generation,
registration.params.clone(),
registration.claim_mode,
)
})
})
}) else {
return Ok(());
};
match attach(endpoint, params) {
Ok(()) => {
let mut state = lock_unpoisoned(state);
if let Some(registration) = state.registration.as_mut() {
if registration.generation == generation
&& registration.config_dir == config_dir
{
registration.attached_boot_id = Some(endpoint.boot_id.clone());
}
}
return Ok(());
}
Err(error) if attempt + 1 < ATTACH_CLAIM_RETRY_LIMIT && error.contains("claim") => {
// 另一个窗口在本次 attach 前后发布了新 claim:按最新 claim 重新解析后重试。
last_claim_error = Some(error);
refresh_registered_external_agent_runner_gui_owner_claim(
state, config_dir, claim_mode,
)?;
}
Err(error) => return Err(error),
}
}
Err(last_claim_error.unwrap_or_else(|| "Agent Runner attach 重试后仍然失败".to_string()))
}
fn refresh_registered_external_agent_runner_gui_owner_claim(
state: &Mutex<ExternalAgentRunnerGuiOwnerAttachmentState>,
config_dir: &Path,
claim_mode: ExternalAgentRunnerGuiOwnerClaimMode,
) -> Result<(), String> {
let session_revision = reserve_external_agent_runner_gui_owner_claim_revision(state);
let claim =
resolve_external_agent_runner_gui_owner_claim(config_dir, claim_mode, session_revision)?;
let mut state = lock_unpoisoned(state);
let Some(registration) = state.registration.as_mut() else {
return Ok(());
};
if registration.config_dir != config_dir {
return Ok(());
}
registration.claim_mode = claim_mode;
registration.params.gui_owner_epoch = Some(claim.owner_epoch);
registration.params.gui_owner_session_revision = Some(claim.session_revision);
registration.attached_boot_id = None;
Ok(())
}
fn redact_url_queries(line: &str) -> String {
line.split_whitespace()
.map(|token| {
if (token.starts_with("http://") || token.starts_with("https://"))
&& token.contains('?')
{
let base = token.split_once('?').map(|(base, _)| base).unwrap_or(token);
format!("{base}?<query-redacted>")
} else {
token.to_string()
}
})
.collect::<Vec<_>>()
.join(" ")
}
pub(super) fn sanitize_agent_runner_output(line: &str, config_dir: &Path) -> String {
let lowercase = line.to_ascii_lowercase();
if [
"authorization",
"bearer ",
"api_key",
"apikey",
"api key",
"x-api-key",
"token=",
"token:",
"credential",
"password",
"cookie",
"set-cookie",
"secret",
"access_token",
"accesstoken",
"platform_access_token",
"platformaccesstoken",
"refresh_token",
"\"token\"",
"'token'",
]
.iter()
.any(|marker| lowercase.contains(marker))
{
return "<sensitive runner output redacted>".to_string();
}
if lowercase.contains("panic") {
return "<runner panic details redacted>".to_string();
}
let safe_internal_detail =
lowercase.starts_with("agent.runner.failed:") || lowercase.starts_with("runner.");
if !safe_internal_detail {
let summary = if ["error", "failed", "failure", "失败", "错误", "异常"]
.iter()
.any(|marker| lowercase.contains(marker))
{
"<runner error details omitted>"
} else if ["warning", "warn:"]
.iter()
.any(|marker| lowercase.contains(marker))
{
"<runner warning details omitted>"
} else {
"<non-diagnostic runner output omitted>"
};
return summary.to_string();
}
crate::sanitize_diagnostic_message(&redact_url_queries(line), Some(config_dir))
.chars()
.take(AGENT_RUNNER_LOG_OUTPUT_MAX_CHARS)
.collect()
}
fn read_bounded_agent_runner_line<R: BufRead>(
reader: &mut R,
) -> io::Result<Option<(String, bool)>> {
let mut content = Vec::new();
let mut truncated = false;
let mut saw_bytes = false;
loop {
let available = reader.fill_buf()?;
if available.is_empty() {
return if saw_bytes {
Ok(Some((
String::from_utf8_lossy(&content).into_owned(),
truncated,
)))
} else {
Ok(None)
};
}
saw_bytes = true;
let newline = available.iter().position(|byte| *byte == b'\n');
let consumed = newline.map(|index| index + 1).unwrap_or(available.len());
let payload_len = newline.unwrap_or(available.len());
let remaining = AGENT_RUNNER_LOG_INPUT_LINE_MAX_BYTES.saturating_sub(content.len());
let copied = payload_len.min(remaining);
content.extend_from_slice(&available[..copied]);
if copied < payload_len {
truncated = true;
}
reader.consume(consumed);
if newline.is_some() {
return Ok(Some((
String::from_utf8_lossy(&content).into_owned(),
truncated,
)));
}
}
}
fn spawn_agent_runner_log_pump<R>(
stream: R,
stream_name: &'static str,
log_path: PathBuf,
config_dir: PathBuf,
) where
R: Read + Send + 'static,
{
let _ = thread::Builder::new()
.name(format!("agent-runner-{stream_name}-log"))
.spawn(move || {
let mut reader = BufReader::new(stream);
loop {
match read_bounded_agent_runner_line(&mut reader) {
Ok(None) => break,
Ok(Some((line, truncated))) => {
let line = sanitize_agent_runner_output(line.trim(), &config_dir);
let _ = crate::append_bounded_diagnostic_line(
&log_path,
&format!(
"runner.{stream_name} truncated={} {line}",
if truncated { "true" } else { "false" }
),
);
}
Err(_) => {
let _ = crate::append_bounded_diagnostic_line(
&log_path,
&format!("runner.{stream_name}.read-failed details=redacted"),
);
break;
}
}
}
});
}
pub(super) struct LaunchedExternalAgentRunner {
child: Child,
}
pub(super) fn launch_external_agent_runner(
config_dir: &Path,
) -> Result<LaunchedExternalAgentRunner, String> {
let executable = std::env::current_exe()
.map_err(|error| format!("读取 Agent Runner 当前二进制失败:{error}"))?;
let runner_log_path = config_dir.join(AGENT_RUNNER_LOG_FILE_NAME);
let _ = crate::append_bounded_diagnostic_line(&runner_log_path, "runner.launch.begin");
let mut command = Command::new(executable);
let gui_owner_required =
EXTERNAL_AGENT_RUNNER_GUI_OWNER_REQUIRED_CLIENT.load(std::sync::atomic::Ordering::Acquire);
command
.args(external_agent_runner_launch_arguments(
config_dir,
gui_owner_required,
))
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
// SAFETY: the closure only calls the async-signal-safe setsid syscall before exec.
unsafe {
command.pre_exec(|| {
if libc::setsid() == -1 {
Err(io::Error::last_os_error())
} else {
Ok(())
}
});
}
}
#[cfg(windows)]
{
crate::configure_windows_background_std_command(&mut command, true);
}
let mut child = command
.spawn()
.map_err(|error| format!("启动外部 Agent Runner 失败:{error}"))?;
if let Some(stdout) = child.stdout.take() {
spawn_agent_runner_log_pump(
stdout,
"stdout",
runner_log_path.clone(),
config_dir.to_path_buf(),
);
}
if let Some(stderr) = child.stderr.take() {
spawn_agent_runner_log_pump(
stderr,
"stderr",
runner_log_path.clone(),
config_dir.to_path_buf(),
);
}
let _ = crate::append_bounded_diagnostic_line(&runner_log_path, "runner.launch.spawned");
Ok(LaunchedExternalAgentRunner { child })
}
pub(super) fn external_agent_runner_launch_arguments(
config_dir: &Path,
gui_owner_required: bool,
) -> Vec<OsString> {
let mut arguments = vec![
OsString::from("--agent-runner"),
OsString::from("--config-dir"),
config_dir.as_os_str().to_os_string(),
];
if gui_owner_required {
arguments.push(OsString::from("--gui-owner-required"));
}
arguments
}
pub(super) fn send_external_agent_runner_request_with_protocol_and_id(
endpoint: &ExternalAgentRunnerEndpoint,
protocol_version: u32,
request_id: String,
method: &str,
params: ExternalAgentRunnerRequestParams,
) -> Result<Value, String> {
send_external_agent_runner_request_with_protocol_and_id_and_timeouts(
endpoint,
protocol_version,
request_id,
method,
params,
EXTERNAL_AGENT_RUNNER_CONNECT_TIMEOUT,
external_agent_runner_client_read_timeout(method),
EXTERNAL_AGENT_RUNNER_IO_TIMEOUT,
)
}
fn send_external_agent_runner_request_with_protocol_and_id_and_timeouts(
endpoint: &ExternalAgentRunnerEndpoint,
protocol_version: u32,
request_id: String,
method: &str,
params: ExternalAgentRunnerRequestParams,
connect_timeout: Duration,
read_timeout: Duration,
write_timeout: Duration,
) -> Result<Value, String> {
if endpoint.protocol_version != protocol_version {
return Err("Agent Runner endpoint 协议版本不兼容".to_string());
}
let request = ExternalAgentRunnerRequest {
protocol_version,
request_id: request_id.clone(),
token: endpoint.token.clone(),
method: method.to_string(),
params,
};
let payload =
serde_json::to_vec(&request).map_err(|_| "序列化 Agent Runner 请求失败".to_string())?;
let address = SocketAddrV4::new(Ipv4Addr::LOCALHOST, endpoint.port).into();
let mut stream = TcpStream::connect_timeout(&address, connect_timeout)
.map_err(|error| format!("连接 Agent Runner 失败:{error}"))?;
stream
.set_read_timeout(Some(read_timeout))
.and_then(|_| stream.set_write_timeout(Some(write_timeout)))
.map_err(|error| format!("配置 Agent Runner 客户端超时失败:{error}"))?;
write_external_agent_runner_frame(&mut stream, &payload)
.map_err(|error| format!("写入 Agent Runner 请求失败:{error}"))?;
stream
.flush()
.map_err(|error| format!("刷新 Agent Runner 请求失败:{error}"))?;
let response_payload = read_external_agent_runner_frame(&mut stream)
.map_err(|error| format!("读取 Agent Runner 响应失败:{error}"))?;
let response = serde_json::from_slice::<ExternalAgentRunnerResponse>(&response_payload)
.map_err(|_| "解析 Agent Runner 响应失败".to_string())?;
if response.protocol_version != protocol_version {
return Err("Agent Runner 响应协议版本不兼容".to_string());
}
if response.request_id != request_id {
return Err("Agent Runner 响应 requestId 不匹配".to_string());
}
if response.ok {
return Ok(response.result.unwrap_or(Value::Null));
}
let error = response.error.unwrap_or(ExternalAgentRunnerProtocolError {
code: "runner-error".to_string(),
message: "Agent Runner 请求失败".to_string(),
});
Err(redact_runner_secret(
&format!("{}: {}", error.code, error.message),
&endpoint.token,
))
}
pub(super) fn external_agent_runner_client_read_timeout(method: &str) -> Duration {
match method {
"runtime.compact" => EXTERNAL_AGENT_RUNNER_CONTEXT_COMPACTION_IO_TIMEOUT,
_ => EXTERNAL_AGENT_RUNNER_IO_TIMEOUT,
}
}
pub(super) fn send_external_agent_runner_request_with_id(
endpoint: &ExternalAgentRunnerEndpoint,
request_id: String,
method: &str,
params: ExternalAgentRunnerRequestParams,
) -> Result<Value, String> {
send_external_agent_runner_request_with_protocol_and_id(
endpoint,
EXTERNAL_AGENT_RUNNER_PROTOCOL_VERSION,
request_id,
method,
params,
)
}
pub(super) fn send_external_agent_runner_request(
endpoint: &ExternalAgentRunnerEndpoint,
method: &str,
params: ExternalAgentRunnerRequestParams,
) -> Result<Value, String> {
let request_id = random_identifier(b"genarrative-agent-runner-request-id")?;
send_external_agent_runner_request_with_id(endpoint, request_id, method, params)
}
pub(super) fn ping_external_agent_runner(
endpoint: &ExternalAgentRunnerEndpoint,
) -> Result<(), String> {
send_external_agent_runner_request(
endpoint,
"runner.ping",
ExternalAgentRunnerRequestParams::default(),
)
.map(|_| ())
}
pub(super) fn ping_external_agent_runner_with_timeout(
endpoint: &ExternalAgentRunnerEndpoint,
timeout: Duration,
) -> Result<(), String> {
if timeout.is_zero() {
return Err("Agent Runner ping 超时预算已耗尽".to_string());
}
send_external_agent_runner_request_with_protocol_and_id_and_timeouts(
endpoint,
EXTERNAL_AGENT_RUNNER_PROTOCOL_VERSION,
random_identifier(b"genarrative-agent-runner-start-ping-id")?,
"runner.ping",
ExternalAgentRunnerRequestParams::default(),
EXTERNAL_AGENT_RUNNER_CONNECT_TIMEOUT.min(timeout),
timeout,
timeout,
)
.map(|_| ())
}
pub(super) fn retire_incompatible_external_agent_runner(
endpoint_path: &Path,
endpoint: &ExternalAgentRunnerEndpoint,
allow_force_busy_migration: bool,
) -> Result<(), String> {
if !request_external_agent_runner_shutdown_if_idle_at(endpoint_path, endpoint)? {
if !allow_force_busy_migration {
return Err(
"Agent Runner 版本与当前客户端不一致,但旧 Runner 仍有任务,暂不能重启".to_string(),
);
}
force_terminate_external_agent_runner_and_cleanup(endpoint_path, endpoint, true)
.map_err(|error| format!("旧 Agent Runner busy 且安全迁移失败:{error}"))?;
}
Ok(())
}
pub(super) fn verify_external_agent_runner_ping_identity(
endpoint: &ExternalAgentRunnerEndpoint,
) -> Result<(), String> {
let result = send_external_agent_runner_request_with_protocol_and_id(
endpoint,
endpoint.protocol_version,
random_identifier(b"genarrative-agent-runner-identity-ping-id")?,
"runner.ping",
ExternalAgentRunnerRequestParams::default(),
)?;
if result.get("pid").and_then(Value::as_u64) != Some(endpoint.pid as u64)
|| result.get("bootId").and_then(Value::as_str) != Some(endpoint.boot_id.as_str())
{
return Err("Agent Runner ping 身份与 endpoint 不匹配".to_string());
}
Ok(())
}
fn request_external_agent_runner_shutdown_if_idle_at(
endpoint_path: &Path,
endpoint: &ExternalAgentRunnerEndpoint,
) -> Result<bool, String> {
let request_id = random_identifier(b"genarrative-agent-runner-upgrade-request-id")?;
let result = send_external_agent_runner_request_with_protocol_and_id(
endpoint,
endpoint.protocol_version,
request_id,
"runner.shutdown_if_idle",
ExternalAgentRunnerRequestParams::default(),
)?;
let idle = result
.get("idle")
.and_then(Value::as_bool)
.ok_or_else(|| "Agent Runner shutdown_if_idle 响应缺少 idle".to_string())?;
if !idle {
return Ok(false);
}
wait_for_external_agent_runner_boot_exit(
endpoint_path,
endpoint,
EXTERNAL_AGENT_RUNNER_START_TIMEOUT,
"旧 Agent Runner 未在版本切换期限内退出",
)?;
Ok(true)
}
fn wait_for_external_agent_runner_boot_exit(
endpoint_path: &Path,
endpoint: &ExternalAgentRunnerEndpoint,
timeout: Duration,
timeout_error: &str,
) -> Result<(), String> {
let deadline = Instant::now() + timeout;
let lock_path = endpoint_path
.parent()
.map(external_agent_runner_lock_path)
.ok_or_else(|| "Agent Runner endpoint 缺少 AppData 父目录".to_string())?;
loop {
match read_external_agent_runner_endpoint(endpoint_path) {
Ok(current) if current.boot_id == endpoint.boot_id => {}
Ok(_) => return Ok(()),
Err(_) => {
if let Some(lock) =
try_open_external_agent_runner_lock(&lock_path, "Agent Runner 单实例锁")?
{
drop(lock);
return Ok(());
}
}
}
if Instant::now() >= deadline {
return Err(timeout_error.to_string());
}
thread::sleep(Duration::from_millis(50));
}
}
fn read_external_agent_runner_endpoint_for_shutdown(
config_dir: &Path,
) -> Result<Option<(PathBuf, ExternalAgentRunnerEndpoint)>, String> {
let endpoint_path = external_agent_runner_endpoint_path(config_dir);
let lock_path = external_agent_runner_lock_path(config_dir);
let deadline = Instant::now() + EXTERNAL_AGENT_RUNNER_START_TIMEOUT;
loop {
match fs::symlink_metadata(&endpoint_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err("Agent Runner endpoint 不允许符号链接".to_string());
}
Ok(_) => {
let endpoint = read_external_agent_runner_endpoint(&endpoint_path)?;
return Ok(Some((endpoint_path, endpoint)));
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {
if let Some(lock) =
try_open_external_agent_runner_lock(&lock_path, "Agent Runner 单实例锁")?
{
drop(lock);
return Ok(None);
}
if Instant::now() >= deadline {
return Err(
"Agent Runner 启动锁仍被占用,但 endpoint 未在期限内就绪".to_string()
);
}
thread::sleep(Duration::from_millis(50));
}
Err(error) => {
return Err(format!(
"读取 Agent Runner endpoint 元数据失败:{}: {error}",
endpoint_path.display()
));
}
}
}
}
pub(super) fn shutdown_external_agent_runner_if_idle_at(config_dir: &Path) -> Result<bool, String> {
let endpoint_path = external_agent_runner_endpoint_path(config_dir);
let lock_path = external_agent_runner_lock_path(config_dir);
let deadline = Instant::now() + EXTERNAL_AGENT_RUNNER_START_TIMEOUT;
loop {
match fs::symlink_metadata(&endpoint_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err("Agent Runner endpoint 不允许符号链接".to_string());
}
Ok(_) => break,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
if let Some(lock) =
try_open_external_agent_runner_lock(&lock_path, "Agent Runner 单实例锁")?
{
drop(lock);
return Ok(true);
}
if Instant::now() >= deadline {
return Err(
"Agent Runner 启动锁仍被占用,但 endpoint 未在期限内就绪".to_string()
);
}
thread::sleep(Duration::from_millis(50));
}
Err(error) => {
return Err(format!(
"读取 Agent Runner endpoint 元数据失败:{}: {error}",
endpoint_path.display()
));
}
}
}
let endpoint = read_external_agent_runner_endpoint(&endpoint_path)?;
request_external_agent_runner_shutdown_if_idle_at(&endpoint_path, &endpoint)
}
pub(crate) fn shutdown_external_agent_runner_if_idle() -> Result<bool, String> {
let config_dir = external_agent_runner_config_dir()
.ok_or_else(|| "外部 Agent Runner 尚未配置 AppData;请显式传入 --config-dir".to_string())?;
shutdown_external_agent_runner_if_idle_at(&config_dir)
}
pub(super) fn shutdown_external_agent_runner_at(config_dir: &Path) -> Result<(), String> {
let endpoint_path = external_agent_runner_endpoint_path(config_dir);
let lock_path = external_agent_runner_lock_path(config_dir);
let endpoint_deadline = Instant::now() + EXTERNAL_AGENT_RUNNER_GUI_SHUTDOWN_IO_TIMEOUT;
loop {
match fs::symlink_metadata(&endpoint_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err("Agent Runner endpoint 不允许符号链接".to_string());
}
Ok(_) => break,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
if let Some(lock) =
try_open_external_agent_runner_lock(&lock_path, "Agent Runner 单实例锁")?
{
drop(lock);
return Ok(());
}
if Instant::now() >= endpoint_deadline {
return Err(
"Agent Runner 实例锁仍被占用,但 endpoint 未在 GUI 关闭期限内出现"
.to_string(),
);
}
thread::sleep(Duration::from_millis(25));
}
Err(error) => {
return Err(format!(
"读取 Agent Runner endpoint 元数据失败:{}: {error}",
endpoint_path.display()
));
}
}
}
let endpoint = read_external_agent_runner_endpoint(&endpoint_path)?;
let graceful_shutdown = send_external_agent_runner_request_with_protocol_and_id_and_timeouts(
&endpoint,
endpoint.protocol_version,
random_identifier(b"genarrative-agent-runner-gui-shutdown-id")?,
"runner.shutdown",
ExternalAgentRunnerRequestParams::default(),
EXTERNAL_AGENT_RUNNER_GUI_SHUTDOWN_CONNECT_TIMEOUT,
EXTERNAL_AGENT_RUNNER_GUI_SHUTDOWN_IO_TIMEOUT,
EXTERNAL_AGENT_RUNNER_GUI_SHUTDOWN_IO_TIMEOUT,
)
.and_then(|result| {
result
.get("willShutdown")
.and_then(Value::as_bool)
.unwrap_or(false)
.then_some(())
.ok_or_else(|| "Agent Runner shutdown 响应未确认退出".to_string())
});
match graceful_shutdown {
Ok(()) => {
let deadline = Instant::now() + EXTERNAL_AGENT_RUNNER_GUI_SHUTDOWN_EXIT_TIMEOUT;
loop {
match read_external_agent_runner_endpoint(&endpoint_path) {
Ok(current) if current.boot_id == endpoint.boot_id => {}
Ok(_) => return Ok(()),
Err(_) => {
if let Some(lock) = try_open_external_agent_runner_lock(
&lock_path,
"Agent Runner 单实例锁",
)? {
drop(lock);
return Ok(());
}
}
}
if Instant::now() >= deadline {
return force_terminate_external_agent_runner_and_cleanup(
&endpoint_path,
&endpoint,
false,
);
}
thread::sleep(Duration::from_millis(25));
}
}
Err(graceful_error) => {
force_terminate_external_agent_runner_and_cleanup(&endpoint_path, &endpoint, true)
.map_err(|force_error| {
format!("{graceful_error};强制终止 Agent Runner 失败:{force_error}")
})
}
}
}
fn force_terminate_external_agent_runner_and_cleanup(
endpoint_path: &Path,
endpoint: &ExternalAgentRunnerEndpoint,
allow_verified_legacy_identity: bool,
) -> Result<(), String> {
force_terminate_external_agent_runner_process(endpoint, allow_verified_legacy_identity)?;
let config_dir = endpoint_path
.parent()
.ok_or_else(|| "Agent Runner endpoint 缺少 AppData 父目录".to_string())?;
let lock_path = external_agent_runner_lock_path(config_dir);
let deadline = Instant::now() + EXTERNAL_AGENT_RUNNER_GUI_FORCE_TERMINATE_GRACE;
loop {
if let Some(lock) =
try_open_external_agent_runner_lock(&lock_path, "Agent Runner 单实例锁")?
{
if read_external_agent_runner_endpoint(endpoint_path)
.ok()
.is_some_and(|current| current.boot_id == endpoint.boot_id)
{
fs::remove_file(endpoint_path)
.map_err(|error| format!("清理已终止 Agent Runner endpoint 失败:{error}"))?;
}
drop(lock);
return Ok(());
}
if Instant::now() >= deadline {
return Err("Agent Runner 已终止但实例锁未在期限内释放".to_string());
}
thread::sleep(Duration::from_millis(25));
}
}
#[cfg(target_os = "linux")]
struct ExternalAgentRunnerPidFd(i32);
#[cfg(target_os = "linux")]
impl Drop for ExternalAgentRunnerPidFd {
fn drop(&mut self) {
// SAFETY: self.0 is an owned pidfd returned by pidfd_open.
unsafe { libc::close(self.0) };
}
}
#[cfg(target_os = "linux")]
fn open_external_agent_runner_pidfd(pid: i32) -> Result<Option<ExternalAgentRunnerPidFd>, String> {
// SAFETY: pidfd_open receives a range-checked pid and flags=0.
let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) };
if fd < 0 {
let error = io::Error::last_os_error();
if error.raw_os_error() == Some(libc::ESRCH) {
return Ok(None);
}
return Err(format!("打开 Agent Runner pidfd 失败:{error}"));
}
Ok(Some(ExternalAgentRunnerPidFd(fd as i32)))
}
#[cfg(target_os = "linux")]
fn signal_external_agent_runner_pidfd(
pidfd: &ExternalAgentRunnerPidFd,
signal: i32,
) -> Result<bool, String> {
// SAFETY: pidfd is owned and live; null siginfo with flags=0 matches pidfd_send_signal.
let result = unsafe {
libc::syscall(
libc::SYS_pidfd_send_signal,
pidfd.0,
signal,
std::ptr::null::<libc::siginfo_t>(),
0,
)
};
if result == 0 {
return Ok(true);
}
let error = io::Error::last_os_error();
if error.raw_os_error() == Some(libc::ESRCH) {
Ok(false)
} else {
Err(format!("通过 pidfd 发送信号失败:{error}"))
}
}
#[cfg(target_os = "linux")]
fn force_terminate_external_agent_runner_process(
endpoint: &ExternalAgentRunnerEndpoint,
allow_verified_legacy_identity: bool,
) -> Result<(), String> {
if endpoint.pid == std::process::id() {
return Err("拒绝终止当前 GUI 进程".to_string());
}
let pid = i32::try_from(endpoint.pid).map_err(|_| "Agent Runner pid 超出范围".to_string())?;
let Some(pidfd) = open_external_agent_runner_pidfd(pid)? else {
return Ok(());
};
if let Some(expected_start_identity) = endpoint.process_start_identity.as_deref() {
let actual_start_identity = external_agent_runner_process_start_identity(endpoint.pid)?
.ok_or_else(|| "当前平台未返回 Agent Runner 进程启动身份".to_string())?;
if actual_start_identity != expected_start_identity {
return Err("Agent Runner pid 已被其他进程复用,拒绝终止".to_string());
}
} else if allow_verified_legacy_identity {
verify_external_agent_runner_ping_identity(endpoint)?;
} else {
return Err("旧 Agent Runner endpoint 缺少进程启动身份,拒绝强制终止".to_string());
}
if !signal_external_agent_runner_pidfd(&pidfd, libc::SIGTERM)? {
return Ok(());
}
let deadline = Instant::now() + EXTERNAL_AGENT_RUNNER_GUI_FORCE_TERMINATE_GRACE;
while signal_external_agent_runner_pidfd(&pidfd, 0)? && Instant::now() < deadline {
thread::sleep(Duration::from_millis(25));
}
if !signal_external_agent_runner_pidfd(&pidfd, 0)? {
return Ok(());
}
signal_external_agent_runner_pidfd(&pidfd, libc::SIGKILL)?;
Ok(())
}
#[cfg(windows)]
fn force_terminate_external_agent_runner_process(
endpoint: &ExternalAgentRunnerEndpoint,
allow_verified_legacy_identity: bool,
) -> Result<(), String> {
use std::ffi::c_void;
#[repr(C)]
struct FileTime {
low_date_time: u32,
high_date_time: u32,
}
#[link(name = "kernel32")]
unsafe extern "system" {
fn OpenProcess(access: u32, inherit_handle: i32, process_id: u32) -> *mut c_void;
fn GetProcessTimes(
process: *mut c_void,
creation_time: *mut FileTime,
exit_time: *mut FileTime,
kernel_time: *mut FileTime,
user_time: *mut FileTime,
) -> i32;
fn TerminateProcess(process: *mut c_void, exit_code: u32) -> i32;
fn WaitForSingleObject(handle: *mut c_void, milliseconds: u32) -> u32;
fn CloseHandle(handle: *mut c_void) -> i32;
}
const PROCESS_TERMINATE: u32 = 0x0001;
const SYNCHRONIZE: u32 = 0x0010_0000;
const PROCESS_QUERY_LIMITED_INFORMATION: u32 = 0x1000;
if endpoint.pid == std::process::id() {
return Err("拒绝终止当前 GUI 进程".to_string());
}
// SAFETY: OpenProcess returns an owned kernel handle or null; it is closed below.
let process = unsafe {
OpenProcess(
PROCESS_TERMINATE | SYNCHRONIZE | PROCESS_QUERY_LIMITED_INFORMATION,
0,
endpoint.pid,
)
};
if process.is_null() {
let error = io::Error::last_os_error();
if error.raw_os_error() == Some(87) {
return Ok(());
}
return Err(format!("打开 Agent Runner 进程失败:{error}"));
}
let result = (|| {
if let Some(expected_start_identity) = endpoint.process_start_identity.as_deref() {
// SAFETY: FileTime is plain data filled by GetProcessTimes.
let mut creation = unsafe { std::mem::zeroed::<FileTime>() };
let mut exit = unsafe { std::mem::zeroed::<FileTime>() };
let mut kernel = unsafe { std::mem::zeroed::<FileTime>() };
let mut user = unsafe { std::mem::zeroed::<FileTime>() };
// SAFETY: process is live and all output pointers refer to writable FileTime values.
if unsafe { GetProcessTimes(process, &mut creation, &mut exit, &mut kernel, &mut user) }
== 0
{
return Err(format!(
"读取 Agent Runner 进程启动身份失败:{}",
io::Error::last_os_error()
));
}
let actual_start_identity = ((creation.high_date_time as u64) << 32
| creation.low_date_time as u64)
.to_string();
if actual_start_identity != expected_start_identity {
return Err("Agent Runner pid 已被其他进程复用,拒绝终止".to_string());
}
} else if allow_verified_legacy_identity {
verify_external_agent_runner_ping_identity(endpoint)?;
} else {
return Err("旧 Agent Runner endpoint 缺少进程启动身份,拒绝强制终止".to_string());
}
// SAFETY: the handle includes PROCESS_TERMINATE and its process start identity was verified.
if unsafe { TerminateProcess(process, 1) } == 0 {
return Err(format!(
"终止 Agent Runner 失败:{}",
io::Error::last_os_error()
));
}
// SAFETY: the handle includes SYNCHRONIZE and remains valid for this bounded wait.
let wait = unsafe { WaitForSingleObject(process, 1_000) };
if wait != 0 {
return Err(format!("等待 Agent Runner 退出失败:waitResult={wait}"));
}
Ok(())
})();
// SAFETY: process is an owned non-null handle returned by OpenProcess.
unsafe { CloseHandle(process) };
result
}
#[cfg(target_os = "macos")]
fn force_terminate_external_agent_runner_process(
_endpoint: &ExternalAgentRunnerEndpoint,
_allow_verified_legacy_identity: bool,
) -> Result<(), String> {
Err("macOS 不提供可绑定进程实例的安全强制终止句柄,拒绝按裸 pid 终止 Agent Runner".to_string())
}
#[cfg(not(any(target_os = "linux", windows, target_os = "macos")))]
fn force_terminate_external_agent_runner_process(
_endpoint: &ExternalAgentRunnerEndpoint,
_allow_verified_legacy_identity: bool,
) -> Result<(), String> {
Err("当前平台不支持核验并强制终止 Agent Runner".to_string())
}
pub(crate) fn shutdown_external_agent_runner() -> Result<(), String> {
let config_dir = external_agent_runner_config_dir()
.ok_or_else(|| "外部 Agent Runner 尚未配置 AppData;请显式传入 --config-dir".to_string())?;
shutdown_external_agent_runner_at(&config_dir)
}
pub(crate) fn attach_external_agent_runner_gui_owner(
event_sink: &GameCreatorManifestInvalidationEventSink,
) -> Result<(), String> {
EXTERNAL_AGENT_RUNNER_GUI_OWNER_REQUIRED_CLIENT
.store(true, std::sync::atomic::Ordering::Release);
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
let config_dir = external_agent_runner_config_dir()
.ok_or_else(|| "外部 Agent Runner 尚未配置 AppData;请显式传入 --config-dir".to_string())?;
let platform_session = crate::current_platform_session();
// 启动阶段先采纳现有 durable claim:第二个及后续窗口与第一个窗口共享同一
// epoch,因此不会被判定为抢走登录态权威;claim 缺失或不可读时才发布新 claim。
let session_revision = reserve_external_agent_runner_gui_owner_claim_revision(
external_agent_runner_gui_owner_attachment_state(),
);
let claim = resolve_external_agent_runner_gui_owner_claim(
&config_dir,
ExternalAgentRunnerGuiOwnerClaimMode::Adopt,
session_revision,
)?;
register_external_agent_runner_gui_owner_attachment(
external_agent_runner_gui_owner_attachment_state(),
&config_dir,
ExternalAgentRunnerGuiOwnerClaimMode::Adopt,
ExternalAgentRunnerRequestParams {
event_sink_port: Some(event_sink.port),
event_sink_token: Some(event_sink.token.clone()),
gui_owner_epoch: Some(claim.owner_epoch),
gui_owner_session_revision: Some(claim.session_revision),
platform_user_id: platform_session
.as_ref()
.map(|session| session.user_id.clone()),
platform_access_token: platform_session
.as_ref()
.map(|session| session.access_token.clone()),
platform_api_base_url: platform_session
.as_ref()
.map(|session| session.api_base_url.clone()),
platform_auth_generation: platform_session
.as_ref()
.map(|session| session.identity_generation),
platform_auth_revision: platform_session.map(|session| session.revision),
..ExternalAgentRunnerRequestParams::default()
},
)?;
ensure_external_agent_runner(&config_dir).map(|_| ())
}
pub(crate) fn install_external_agent_runner_platform_session(
user_id: &str,
access_token: &str,
api_base_url: &str,
identity_generation: u64,
revision: u64,
) -> Result<(), String> {
let config_dir = external_agent_runner_config_dir()
.ok_or_else(|| "外部 Agent Runner 尚未配置 AppData".to_string())?;
synchronize_external_agent_runner_platform_session_with(
|| {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
remember_external_agent_runner_platform_session(
external_agent_runner_gui_owner_attachment_state(),
Some((user_id, access_token, api_base_url)),
identity_generation,
revision,
)
.and_then(|_| ensure_external_agent_runner(&config_dir))
.and_then(|endpoint| {
validate_external_agent_runner_platform_session_attachment(
external_agent_runner_gui_owner_attachment_state(),
&config_dir,
&endpoint,
Some((user_id, access_token, api_base_url)),
identity_generation,
revision,
)
})
},
|| shutdown_external_agent_runner_at(&config_dir),
)
}
pub(crate) fn clear_external_agent_runner_platform_session(
identity_generation: u64,
revision: u64,
) -> Result<(), String> {
let Some(config_dir) = external_agent_runner_config_dir() else {
return Ok(());
};
synchronize_external_agent_runner_platform_session_with(
|| {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
remember_external_agent_runner_platform_session(
external_agent_runner_gui_owner_attachment_state(),
None,
identity_generation,
revision,
)
.and_then(|_| ensure_external_agent_runner(&config_dir))
.and_then(|endpoint| {
validate_external_agent_runner_platform_session_attachment(
external_agent_runner_gui_owner_attachment_state(),
&config_dir,
&endpoint,
None,
identity_generation,
revision,
)
})
},
|| shutdown_external_agent_runner_at(&config_dir),
)
}
fn validate_external_agent_runner_platform_session_attachment(
state: &Mutex<ExternalAgentRunnerGuiOwnerAttachmentState>,
config_dir: &Path,
endpoint: &ExternalAgentRunnerEndpoint,
session: Option<(&str, &str, &str)>,
identity_generation: u64,
revision: u64,
) -> Result<(), String> {
let state = lock_unpoisoned(state);
let registration = state.registration.as_ref().ok_or_else(|| {
"Agent Runner 尚未建立带 owner epoch 的 GUI owner 登记,平台登录态拒绝下发".to_string()
})?;
let expected_user_id = session.map(|(user_id, _, _)| user_id);
let expected_access_token = session.map(|(_, access_token, _)| access_token);
let expected_api_base_url = session.map(|(_, _, api_base_url)| api_base_url);
if registration.config_dir != config_dir
|| registration.attached_boot_id.as_deref() != Some(endpoint.boot_id.as_str())
|| registration.params.gui_owner_epoch.is_none()
|| registration.params.gui_owner_session_revision != Some(registration.generation)
|| registration.params.platform_auth_generation != Some(identity_generation)
|| registration.params.platform_auth_revision != Some(revision)
|| registration.params.platform_user_id.as_deref() != expected_user_id
|| registration.params.platform_access_token.as_deref() != expected_access_token
|| registration.params.platform_api_base_url.as_deref() != expected_api_base_url
{
return Err("Agent Runner 未确认当前 GUI owner epoch 的平台登录态".to_string());
}
Ok(())
}
pub(super) fn synchronize_external_agent_runner_platform_session_with(
sync_attempt: impl FnOnce() -> Result<(), String>,
fence_runner: impl FnOnce() -> Result<(), String>,
) -> Result<(), String> {
let sync_result = sync_attempt();
let Err(sync_error) = sync_result else {
return Ok(());
};
match fence_runner() {
Ok(()) => Err(format!(
"{sync_error};为避免旧账号继续执行,Agent Runner 已停止,后续请求将按当前账号重建"
)),
Err(fence_error) => Err(format!(
"{sync_error};阻断旧账号 Agent Runner 失败:{fence_error}"
)),
}
}
pub(super) fn remember_external_agent_runner_platform_session(
state: &Mutex<ExternalAgentRunnerGuiOwnerAttachmentState>,
session: Option<(&str, &str, &str)>,
identity_generation: u64,
revision: u64,
) -> Result<(), String> {
remember_external_agent_runner_platform_session_with(
state,
session,
identity_generation,
revision,
publish_external_agent_runner_gui_owner_claim,
)
}
pub(super) fn remember_external_agent_runner_platform_session_with(
state: &Mutex<ExternalAgentRunnerGuiOwnerAttachmentState>,
session: Option<(&str, &str, &str)>,
identity_generation: u64,
revision: u64,
publish_claim: impl FnOnce(&Path, u64) -> Result<ExternalAgentRunnerGuiOwnerClaim, String>,
) -> Result<(), String> {
let mut state = lock_unpoisoned(state);
let Some(registration) = state.registration.as_ref() else {
return Ok(());
};
// 写入顺序只认 revision;身份代次只表达主体归属,同一账号续期会推进 revision
// 但保持 identity generation 不变。
let current_revision = registration.params.platform_auth_revision.unwrap_or(0);
if revision < current_revision {
return Ok(());
}
if revision == current_revision {
match session {
Some((user_id, access_token, api_base_url))
if registration.params.platform_user_id.as_deref() == Some(user_id)
&& registration.params.platform_auth_generation
== Some(identity_generation)
&& registration.params.platform_access_token.as_deref()
== Some(access_token)
&& registration.params.platform_api_base_url.as_deref()
== Some(api_base_url) =>
{
return Ok(());
}
Some(_) => return Ok(()),
None if registration.params.platform_user_id.is_none()
&& registration.params.platform_access_token.is_none() =>
{
return Ok(());
}
None => {}
}
}
state.generation = state.generation.wrapping_add(1);
let registration_generation = state.generation;
// 本窗口改动了平台登录态:发布新 epoch 的 claim,成为新的登录态权威。
// 并发发布以最后一次成功写入为准,落败窗口在 attach 阶段按最新 claim 重试。
// 只有已经建立过 claim 的登记才需要发布:没有 epoch 的登记(纯 CLI / 单元测试替身)
// 不写任何 claim 文件。
let published_claim = state
.registration
.as_ref()
.filter(|registration| registration.params.gui_owner_epoch.is_some())
.map(|registration| registration.config_dir.clone())
.map(|config_dir| publish_claim(&config_dir, registration_generation))
.transpose()?;
let registration = state
.registration
.as_mut()
.expect("checked GUI owner registration must remain present while locked");
registration.generation = registration_generation;
registration.claim_mode = ExternalAgentRunnerGuiOwnerClaimMode::Publish;
registration.attached_boot_id = None;
registration.params.platform_user_id = session.map(|(user_id, _, _)| user_id.to_string());
registration.params.platform_access_token =
session.map(|(_, access_token, _)| access_token.to_string());
registration.params.platform_api_base_url =
session.map(|(_, _, api_base_url)| api_base_url.to_string());
registration.params.platform_auth_generation = Some(identity_generation);
registration.params.platform_auth_revision = Some(revision);
if let Some(claim) = published_claim {
registration.params.gui_owner_epoch = Some(claim.owner_epoch);
registration.params.gui_owner_session_revision = Some(claim.session_revision);
}
Ok(())
}
pub(super) fn validate_external_agent_runner_gui_owner_attachment_result(
result: &Value,
) -> Result<(), String> {
if result.get("attached").and_then(Value::as_bool) == Some(true)
&& result.get("eventSinkAttached").and_then(Value::as_bool) == Some(true)
{
Ok(())
} else {
Err("Agent Runner attach_gui_owner 响应未确认 owner 与事件接收端".to_string())
}
}
fn attach_external_agent_runner_gui_owner_at(
endpoint: &ExternalAgentRunnerEndpoint,
params: ExternalAgentRunnerRequestParams,
) -> Result<(), String> {
let result = send_external_agent_runner_request(endpoint, "runner.attach_gui_owner", params)?;
validate_external_agent_runner_gui_owner_attachment_result(&result)
}
fn attach_registered_external_agent_runner_gui_owner_if_needed(
config_dir: &Path,
endpoint: &ExternalAgentRunnerEndpoint,
) -> Result<(), String> {
attach_registered_external_agent_runner_gui_owner_if_needed_with(
external_agent_runner_gui_owner_attachment_state(),
config_dir,
endpoint,
attach_external_agent_runner_gui_owner_at,
)
}
pub(super) fn shutdown_external_agent_runner_for_client_exit_at(
config_dir: &Path,
) -> Result<bool, String> {
let Some((endpoint_path, endpoint)) =
read_external_agent_runner_endpoint_for_shutdown(config_dir)?
else {
return Ok(true);
};
let request_id = random_identifier(b"genarrative-agent-runner-client-exit-request-id")?;
let result = match send_external_agent_runner_request_with_protocol_and_id(
&endpoint,
endpoint.protocol_version,
request_id,
"runner.shutdown_for_client_exit",
ExternalAgentRunnerRequestParams::default(),
) {
Ok(result) => result,
Err(error) => {
return match read_external_agent_runner_endpoint(&endpoint_path) {
Ok(current) if current.boot_id == endpoint.boot_id => Err(error),
_ => Ok(true),
};
}
};
let accepted = result
.get("accepted")
.and_then(Value::as_bool)
.ok_or_else(|| "Agent Runner shutdown_for_client_exit 响应缺少 accepted".to_string())?;
let busy = result
.get("busy")
.and_then(Value::as_bool)
.ok_or_else(|| "Agent Runner shutdown_for_client_exit 响应缺少 busy".to_string())?;
let will_shutdown = result
.get("willShutdown")
.and_then(Value::as_bool)
.ok_or_else(|| "Agent Runner shutdown_for_client_exit 响应缺少 willShutdown".to_string())?;
match (accepted, busy, will_shutdown) {
(false, true, false) => return Ok(false),
(true, false, true) => {}
_ => return Err("Agent Runner shutdown_for_client_exit 响应状态不一致".to_string()),
}
wait_for_external_agent_runner_boot_exit(
&endpoint_path,
&endpoint,
AGENT_RUNNER_CLIENT_EXIT_TIMEOUT,
"Agent Runner 未在客户端退出期限内停止",
)?;
Ok(true)
}
pub(crate) fn shutdown_external_agent_runner_for_client_exit() -> Result<bool, String> {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
let Some(config_dir) = external_agent_runner_config_dir() else {
return Ok(true);
};
shutdown_external_agent_runner_for_client_exit_at(&config_dir)
}
/// 窗口退出的收尾:先释放本窗口参与锁,再决定 Runner 是否需要关闭。
///
/// 返回 `Ok(false)` 表示仍检测到其它窗口持有参与锁,Runner 必须保留给它们;
/// 返回 `Ok(true)` 表示本窗口是最后一个界面进程,Runner 已请求关闭。
pub(crate) fn shutdown_external_agent_runner_for_gui_exit() -> Result<bool, String> {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
let Some(config_dir) = external_agent_runner_config_dir() else {
return Ok(true);
};
release_external_agent_runner_gui_participant_lock();
if external_agent_runner_lock_is_held(&external_agent_runner_gui_participant_lock_path(
&config_dir,
))? {
return Ok(false);
}
shutdown_external_agent_runner_at(&config_dir)?;
Ok(true)
}
pub(super) fn wait_for_external_agent_runner(
config_dir: &Path,
child: &mut Child,
executable_fingerprint: &str,
) -> Result<ExternalAgentRunnerEndpoint, String> {
let endpoint_path = external_agent_runner_endpoint_path(config_dir);
let deadline = Instant::now() + EXTERNAL_AGENT_RUNNER_START_TIMEOUT;
loop {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Err("外部 Agent Runner 未在启动期限内就绪".to_string());
}
if let Some(endpoint) =
read_current_external_agent_runner_endpoint(&endpoint_path, executable_fingerprint)
{
let ping_timeout = EXTERNAL_AGENT_RUNNER_IO_TIMEOUT.min(remaining);
if ping_external_agent_runner_with_timeout(&endpoint, ping_timeout).is_ok() {
return Ok(endpoint);
}
}
if let Some(status) = child
.try_wait()
.map_err(|error| format!("检查外部 Agent Runner 子进程失败:{error}"))?
{
return Err(format!("外部 Agent Runner 在就绪前退出:{status}"));
}
if Instant::now() >= deadline {
return Err("外部 Agent Runner 未在启动期限内就绪".to_string());
}
thread::sleep(Duration::from_millis(50));
}
}
pub(super) fn ensure_external_agent_runner(
config_dir: &Path,
) -> Result<ExternalAgentRunnerEndpoint, String> {
let endpoint_path = external_agent_runner_endpoint_path(config_dir);
let executable_fingerprint = current_external_agent_runner_executable_fingerprint()?;
if let Some(endpoint) = reuse_or_retire_external_agent_runner_endpoint(
config_dir,
&endpoint_path,
&executable_fingerprint,
)? {
return Ok(endpoint);
}
let mut launched = launch_external_agent_runner(config_dir)?;
match wait_for_external_agent_runner(config_dir, &mut launched.child, &executable_fingerprint) {
Ok(endpoint) => {
thread::Builder::new()
.name("agent-runner-reaper".to_string())
.spawn(move || {
let _ = launched.child.wait();
})
.map_err(|error| format!("启动 Agent Runner 子进程回收线程失败:{error}"))?;
attach_registered_external_agent_runner_gui_owner_if_needed(config_dir, &endpoint)?;
Ok(endpoint)
}
Err(error) => {
let _ = launched.child.kill();
let _ = launched.child.wait();
// 同一 AppData 的另一个窗口可能在这段时间里已经启动了 Runner:
// 实例锁竞争失败不能立刻报成启动失败,先按最新 endpoint 复用一次。
if let Some(endpoint) = reuse_or_retire_external_agent_runner_endpoint(
config_dir,
&endpoint_path,
&executable_fingerprint,
)? {
return Ok(endpoint);
}
Err(error)
}
}
}
fn reuse_or_retire_external_agent_runner_endpoint(
config_dir: &Path,
endpoint_path: &Path,
executable_fingerprint: &str,
) -> Result<Option<ExternalAgentRunnerEndpoint>, String> {
if let Ok(endpoint) = read_external_agent_runner_endpoint(&endpoint_path) {
match external_agent_runner_endpoint_reuse_decision(&endpoint, executable_fingerprint) {
ExternalAgentRunnerReuseDecision::Reuse => {
if ping_external_agent_runner(&endpoint).is_ok() {
attach_registered_external_agent_runner_gui_owner_if_needed(
config_dir, &endpoint,
)?;
return Ok(Some(endpoint));
}
}
ExternalAgentRunnerReuseDecision::Retire => {
let incompatible_ping = send_external_agent_runner_request_with_protocol_and_id(
&endpoint,
endpoint.protocol_version,
random_identifier(b"genarrative-agent-runner-upgrade-ping-id")?,
"runner.ping",
ExternalAgentRunnerRequestParams::default(),
);
if incompatible_ping.is_ok() {
retire_incompatible_external_agent_runner(
endpoint_path,
&endpoint,
EXTERNAL_AGENT_RUNNER_GUI_OWNER_REQUIRED_CLIENT
.load(std::sync::atomic::Ordering::Acquire),
)?;
}
}
}
}
Ok(None)
}
pub(crate) fn configure_external_agent_runner(config_dir: impl AsRef<Path>) -> Result<(), String> {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
let config_dir = normalize_external_agent_runner_config_dir(config_dir.as_ref())?;
set_external_agent_runner_config_dir(config_dir);
Ok(())
}
pub(crate) fn configure_external_agent_runner_read_only(
config_dir: impl AsRef<Path>,
) -> Result<(), String> {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
let config_dir = inspect_external_agent_runner_config_dir(config_dir.as_ref())?;
set_external_agent_runner_config_dir(config_dir);
Ok(())
}
pub(crate) fn ensure_external_agent_runner_started() -> Result<(), String> {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
let config_dir = external_agent_runner_config_dir()
.ok_or_else(|| "外部 Agent Runner 尚未配置 AppData;请显式传入 --config-dir".to_string())?;
ensure_external_agent_runner(&config_dir).map(|_| ())
}
pub(crate) fn ensure_external_agent_runner_started_for_gui() -> Result<(), String> {
EXTERNAL_AGENT_RUNNER_GUI_OWNER_REQUIRED_CLIENT
.store(true, std::sync::atomic::Ordering::Release);
ensure_external_agent_runner_started()
}
pub(crate) fn require_external_agent_runner_for_cli_runtime_write(
root: &Path,
) -> Result<(), String> {
require_external_agent_runner_configured_for_cli_runtime_write(root)?;
ensure_external_agent_runner_started()
}
pub(crate) fn require_external_agent_runner_configured_for_cli_runtime_write(
root: &Path,
) -> Result<(), String> {
if external_agent_runner_is_server_process() {
return Err("Agent Runner 进程不能作为普通 CLI 执行 Runtime 写命令".to_string());
}
let config_dir = external_agent_runner_config_dir().ok_or_else(|| {
"Agent Runtime 写命令必须显式传入 --config-dir <项目外 AppData 绝对路径>".to_string()
})?;
if !root.is_absolute() {
return Err("Agent Runtime 写命令的项目路径必须是绝对路径".to_string());
}
crate::validate_game_creator_runtime_config_dir_outside_project(&config_dir, root)
}
pub(super) fn parse_external_agent_runner_notification_kind(
kind: &str,
) -> Result<(&'static str, Option<String>), String> {
match kind.trim() {
"wake_pending" | "runtime.wake_pending" => Ok(("runtime.wake_pending", None)),
"resume" | "runtime.resume" => Ok(("runtime.resume", None)),
"shutdown_if_idle" | "runner.shutdown_if_idle" => Ok(("runner.shutdown_if_idle", None)),
value => {
let agent = value
.strip_prefix("continue_action:")
.or_else(|| value.strip_prefix("runtime.continue_action:"))
.map(str::trim)
.filter(|value| !value.is_empty());
match agent {
Some(agent) => Ok(("runtime.continue_action", Some(agent.to_string()))),
None => Err("未知 Agent Runner 通知类型".to_string()),
}
}
}
}
pub(super) fn send_external_agent_runner_runtime_request(
root: &Path,
method: &str,
agent: Option<&str>,
run_id: Option<&str>,
action_id: Option<&str>,
steer_id: Option<&str>,
) -> Result<Value, String> {
send_external_agent_runner_runtime_request_with_stable_identity(
root, method, agent, run_id, action_id, steer_id, None,
)
}
pub(super) fn send_external_agent_runner_runtime_request_with_stable_identity(
root: &Path,
method: &str,
agent: Option<&str>,
run_id: Option<&str>,
action_id: Option<&str>,
steer_id: Option<&str>,
stable_identity: Option<&str>,
) -> Result<Value, String> {
let root = canonicalize_external_agent_runner_project_root(root)?;
let root = root
.to_str()
.ok_or_else(|| "通知 Agent Runner 的项目 root 必须是 UTF-8 路径".to_string())?;
let config_dir = external_agent_runner_config_dir()
.ok_or_else(|| "外部 Agent Runner 尚未配置".to_string())?;
crate::validate_game_creator_runtime_config_dir_outside_project(&config_dir, Path::new(root))?;
// Establish liveness before the write request. The write itself is sent exactly once.
let endpoint = {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
ensure_external_agent_runner(&config_dir)?
};
let params = ExternalAgentRunnerRequestParams {
root: Some(root.to_string()),
agent: agent.map(str::to_string),
session_id: None,
run_id: run_id.map(str::to_string),
action_id: action_id.map(str::to_string),
steer_id: steer_id.map(str::to_string),
event_sink_port: None,
event_sink_token: None,
gui_owner_epoch: None,
gui_owner_session_revision: None,
platform_user_id: None,
platform_access_token: None,
platform_api_base_url: None,
platform_auth_generation: None,
platform_auth_revision: None,
};
match stable_identity {
Some(stable_identity) => {
let identity = format!("{root}\n{method}\n{stable_identity}");
let fingerprint = format!("{:x}", Sha256::digest(identity.as_bytes()));
send_external_agent_runner_request_with_id(
&endpoint,
format!("runtime-parent-wake-{}", &fingerprint[..32]),
method,
params,
)
}
None => send_external_agent_runner_request(&endpoint, method, params),
}
}
pub(crate) fn compact_external_agent_runner_context(
root: &Path,
agent: &str,
session_id: Option<&str>,
) -> Result<AgentRuntimeContextCompactionResult, String> {
let agent = agent.trim();
if agent.is_empty() {
return Err("手动压缩 Agent 上下文必须提供 agent".to_string());
}
let root = canonicalize_external_agent_runner_project_root(root)?;
let root_text = root
.to_str()
.ok_or_else(|| "通知 Agent Runner 的项目 root 必须是 UTF-8 路径".to_string())?;
let config_dir = external_agent_runner_config_dir()
.ok_or_else(|| "外部 Agent Runner 尚未配置".to_string())?;
crate::validate_game_creator_runtime_config_dir_outside_project(&config_dir, &root)?;
let endpoint = {
let _configure = lock_unpoisoned(external_agent_runner_configure_lock());
ensure_external_agent_runner(&config_dir)?
};
let result = send_external_agent_runner_request(
&endpoint,
"runtime.compact",
ExternalAgentRunnerRequestParams {
root: Some(root_text.to_string()),
agent: Some(agent.to_string()),
session_id: session_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string),
..ExternalAgentRunnerRequestParams::default()
},
)?;
serde_json::from_value(result)
.map_err(|error| format!("解析 Agent Runner 上下文压缩结果失败:{error}"))
}
pub(crate) fn wake_external_agent_runner_pending(root: &Path) -> Result<(), String> {
send_external_agent_runner_runtime_request(root, "runtime.wake_pending", None, None, None, None)
.map(|_| ())
}
pub(crate) fn wake_external_agent_runner_pending_for_run(
root: &Path,
agent: &str,
run_id: &str,
loop_iteration: u32,
) -> Result<(), String> {
if agent.trim().is_empty() || run_id.trim().is_empty() {
return Err("父 run wake 必须同时提供 agent 和 runId".to_string());
}
let stable_identity = format!("{agent}\n{run_id}\n{loop_iteration}");
send_external_agent_runner_runtime_request_with_stable_identity(
root,
"runtime.wake_pending",
Some(agent),
Some(run_id),
None,
None,
Some(&stable_identity),
)
.map(|_| ())
}
pub(crate) fn resume_external_agent_runner(root: &Path) -> Result<(), String> {
send_external_agent_runner_runtime_request(root, "runtime.resume", None, None, None, None)
.map(|_| ())
}
pub(crate) fn continue_external_agent_runner_action(
root: &Path,
agent: &str,
run_id: &str,
action_id: &str,
) -> Result<(), String> {
if [agent, run_id, action_id]
.into_iter()
.any(|value| value.trim().is_empty())
{
return Err("继续 Agent Runtime 动作必须同时提供 agent/runId/actionId".to_string());
}
send_external_agent_runner_runtime_request(
root,
"runtime.continue_action",
Some(agent),
Some(run_id),
Some(action_id),
None,
)
.map(|_| ())
}
pub(crate) fn steer_external_agent_runner(
root: &Path,
agent: &str,
run_id: &str,
steer_id: &str,
) -> Result<bool, String> {
if [agent, run_id, steer_id]
.into_iter()
.any(|value| value.trim().is_empty())
{
return Err("追加 Agent 指令必须同时提供 agent/runId/steerId".to_string());
}
let agent = agent.trim();
let run_id = run_id.trim();
let steer_id = steer_id.trim();
let result = send_external_agent_runner_runtime_request(
root,
"runtime.steer",
Some(agent),
Some(run_id),
None,
Some(steer_id),
)?;
parse_external_agent_runner_steer_result(&result)
}
pub(crate) fn interrupt_external_agent_runner_provider_for_steer_decision(
root: &Path,
agent: &str,
run_id: &str,
steer_id: &str,
) -> Result<bool, String> {
if [agent, run_id, steer_id]
.into_iter()
.any(|value| value.trim().is_empty())
{
return Err("LLM steer 中断判定必须同时提供 agent/runId/steerId".to_string());
}
let result = send_external_agent_runner_runtime_request(
root,
"runtime.interrupt_for_steer_decision",
Some(agent.trim()),
Some(run_id.trim()),
None,
Some(steer_id.trim()),
)?;
parse_external_agent_runner_steer_result(&result)
}
pub(crate) fn pause_external_agent_runner(
root: &Path,
agent: &str,
run_id: &str,
) -> Result<bool, String> {
if [agent, run_id]
.into_iter()
.any(|value| value.trim().is_empty())
{
return Err("暂停 Agent Goal 必须同时提供 agent/runId".to_string());
}
let result = send_external_agent_runner_runtime_request(
root,
"runtime.pause",
Some(agent.trim()),
Some(run_id.trim()),
None,
None,
)?;
result
.get("providerInterrupted")
.and_then(Value::as_bool)
.ok_or_else(|| "Agent Runner runtime.pause 响应缺少 providerInterrupted".to_string())
}
pub(crate) fn cancel_external_agent_runner_goal(
root: &Path,
agent: &str,
run_id: &str,
) -> Result<bool, String> {
if [agent, run_id]
.into_iter()
.any(|value| value.trim().is_empty())
{
return Err("清理 Agent Goal 必须同时提供 agent/runId".to_string());
}
let result = send_external_agent_runner_runtime_request(
root,
"runtime.cancel",
Some(agent.trim()),
Some(run_id.trim()),
None,
None,
)?;
result
.get("providerInterrupted")
.and_then(Value::as_bool)
.ok_or_else(|| "Agent Runner runtime.cancel 响应缺少 providerInterrupted".to_string())
}
pub(super) fn parse_external_agent_runner_steer_result(result: &Value) -> Result<bool, String> {
result
.get("providerInterrupted")
.and_then(Value::as_bool)
.ok_or_else(|| "Agent Runner runtime.steer 响应缺少 providerInterrupted".to_string())
}
pub(crate) fn notify_external_agent_runner(root: &Path, kind: &str) -> Result<(), String> {
let (method, agent) = parse_external_agent_runner_notification_kind(kind)?;
if method == "runtime.continue_action" {
return Err(
"continue_action 通知必须改用 typed helper 并绑定 agent/runId/actionId".to_string(),
);
}
send_external_agent_runner_runtime_request(root, method, agent.as_deref(), None, None, None)
.map(|_| ())
}
pub(super) fn read_external_agent_runner_status_at(
config_dir: Option<&Path>,
) -> ExternalAgentRunnerStatus {
let Some(config_dir) = config_dir else {
return ExternalAgentRunnerStatus::disabled();
};
let endpoint_path = external_agent_runner_endpoint_path(config_dir);
let endpoint = match read_external_agent_runner_endpoint(&endpoint_path) {
Ok(endpoint) => endpoint,
Err(error) => {
return ExternalAgentRunnerStatus {
enabled: true,
running: false,
protocol_version: EXTERNAL_AGENT_RUNNER_PROTOCOL_VERSION,
pid: None,
boot_id: None,
port: None,
heartbeat_at: None,
error: Some(error),
};
}
};
let mut fallback = ExternalAgentRunnerStatus::from_endpoint(&endpoint, false);
if endpoint.protocol_version != EXTERNAL_AGENT_RUNNER_PROTOCOL_VERSION {
fallback.error = Some("Agent Runner endpoint 协议版本不兼容".to_string());
return fallback;
}
match send_external_agent_runner_request(
&endpoint,
"runner.status",
ExternalAgentRunnerRequestParams::default(),
) {
Ok(value) => match serde_json::from_value::<ExternalAgentRunnerStatus>(value) {
Ok(mut status) => {
status.enabled = true;
status.error = None;
status
}
Err(_) => {
fallback.error = Some("解析 Agent Runner 状态失败".to_string());
fallback
}
},
Err(error) => {
fallback.error = Some(redact_runner_secret(&error, &endpoint.token));
fallback
}
}
}
pub(crate) fn read_external_agent_runner_status() -> ExternalAgentRunnerStatus {
let config_dir = external_agent_runner_config_dir();
read_external_agent_runner_status_at(config_dir.as_deref())
}
#[cfg(test)]
mod diagnostic_log_tests {
use super::*;
#[test]
fn runner_log_output_redacts_config_paths_and_credentials() {
let config_dir = Path::new(r"C:\Users\example\AppData\Roaming\genarrative");
assert_eq!(
sanitize_agent_runner_output(
r"agent.runner.failed: failed to open C:\Users\example\AppData\Roaming\genarrative\state.json",
config_dir,
),
"agent.runner.failed: failed to open <appdata>\\state.json"
);
assert_eq!(
sanitize_agent_runner_output("Authorization: Bearer secret", config_dir),
"<sensitive runner output redacted>"
);
assert_eq!(
sanitize_agent_runner_output(
r"agent.runner.failed: project C:\private\game\index.html failed",
config_dir,
),
"agent.runner.failed: project <absolute-path> failed"
);
assert_eq!(
sanitize_agent_runner_output("normal model response body", config_dir),
"<non-diagnostic runner output omitted>"
);
assert_eq!(
sanitize_agent_runner_output(
"agent.runner.failed: request failed https://example.invalid/api?value=1",
config_dir,
),
"agent.runner.failed: request failed https://example.invalid/api?<query-redacted>"
);
assert_eq!(
sanitize_agent_runner_output("error password=hunter2", config_dir),
"<sensitive runner output redacted>"
);
}
#[test]
fn runner_log_line_reader_caps_long_lines_and_drains_to_next_line() {
let mut input = vec![b'x'; AGENT_RUNNER_LOG_INPUT_LINE_MAX_BYTES + 500];
input.extend_from_slice(b"\nerror: second line\n");
let mut reader = BufReader::new(std::io::Cursor::new(input));
let (first, first_truncated) = read_bounded_agent_runner_line(&mut reader)
.expect("read first line")
.expect("first line exists");
assert_eq!(first.len(), AGENT_RUNNER_LOG_INPUT_LINE_MAX_BYTES);
assert!(first_truncated);
let (second, second_truncated) = read_bounded_agent_runner_line(&mut reader)
.expect("read second line")
.expect("second line exists");
assert_eq!(second, "error: second line");
assert!(!second_truncated);
}
}