kdletters
737a2266b9
- owner 独占锁改为可多窗口同时持有的界面参与锁 agent-runner.gui-participant.lock - Runner 启动检查、attach 门禁与 watchdog 改用参与锁存活判定，最后一个窗口退出才关停 Runner - 窗口启动只采纳现有 owner claim，登录/refresh/退出/换号才发布新 epoch claim - 同 claim 重复 attach 改为幂等空操作，不再清空 Runner 平台登录态 - manifest 失效与 Runtime update relay 接收端改为按 token 去重的注册表并广播，失败只淘汰该接收端 - GUI 退出先释放参与锁，仍有其它窗口时保留 Runner 并记录 retained_for_other_windows - Runner 启动失败时按最新 endpoint 复用一次，避免两窗口同时冷启动被实例锁竞争误报 - 更新 runner 定向测试、GUI 生命周期集成测试与自检用例 - 同步 AGC 技术方案、决策记录、踩坑记录与多窗口里程碑/实施计划
369 lines
14 KiB
Rust
369 lines
14 KiB
Rust
use super::{dispatch::*, endpoint::*, protocol::*, state::*};
|
||
use sha2::{Digest as _, Sha256};
|
||
use std::fs;
|
||
use std::io;
|
||
use std::net::{Ipv4Addr, SocketAddrV4, TcpListener};
|
||
use std::path::{Path, PathBuf};
|
||
use std::sync::atomic::Ordering;
|
||
use std::sync::Arc;
|
||
use std::thread;
|
||
use std::time::{Duration, Instant};
|
||
|
||
pub(super) fn refresh_external_agent_runner_heartbeat(
|
||
state: &ExternalAgentRunnerServerState,
|
||
) -> Result<(), String> {
|
||
let endpoint = {
|
||
let mut endpoint = lock_unpoisoned(&state.endpoint);
|
||
endpoint.heartbeat_at = unix_millis();
|
||
endpoint.clone()
|
||
};
|
||
write_external_agent_runner_endpoint_atomic(&state.endpoint_path, &endpoint)
|
||
}
|
||
|
||
pub(super) fn bind_external_agent_runner_listener_with<T>(
|
||
mut fallback_ports: impl FnMut() -> Vec<u16>,
|
||
mut bind: impl FnMut(u16) -> io::Result<T>,
|
||
) -> io::Result<T> {
|
||
let primary_error = match bind(0) {
|
||
Ok(listener) => return Ok(listener),
|
||
Err(error) => error,
|
||
};
|
||
if primary_error.kind() != io::ErrorKind::AddrInUse {
|
||
return Err(primary_error);
|
||
}
|
||
for port in fallback_ports() {
|
||
match bind(port) {
|
||
Ok(listener) => return Ok(listener),
|
||
Err(error) if error.kind() == io::ErrorKind::AddrInUse => {}
|
||
Err(error) => return Err(error),
|
||
}
|
||
}
|
||
Err(primary_error)
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
pub(super) fn parse_external_agent_runner_linux_ephemeral_port_range(
|
||
content: &str,
|
||
) -> Option<(u16, u16)> {
|
||
let mut values = content.split_whitespace();
|
||
let start = values.next()?.parse::<u16>().ok()?;
|
||
let end = values.next()?.parse::<u16>().ok()?;
|
||
if values.next().is_some() || start > end {
|
||
return None;
|
||
}
|
||
Some((start, end))
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
pub(super) fn parse_external_agent_runner_linux_single_port(content: &str) -> Option<u16> {
|
||
let mut values = content.split_whitespace();
|
||
let value = values.next()?.parse::<u16>().ok()?;
|
||
values.next().is_none().then_some(value)
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
pub(super) fn parse_external_agent_runner_linux_reserved_ports(
|
||
content: &str,
|
||
) -> Option<Vec<(u16, u16)>> {
|
||
let content = content.trim();
|
||
if content.is_empty() {
|
||
return Some(Vec::new());
|
||
}
|
||
let mut ranges = Vec::new();
|
||
for part in content.split(',') {
|
||
let part = part.trim();
|
||
if part.is_empty() {
|
||
return None;
|
||
}
|
||
let mut bounds = part.split('-');
|
||
let start = bounds.next()?.parse::<u16>().ok()?;
|
||
let end = match bounds.next() {
|
||
Some(value) => value.parse::<u16>().ok()?,
|
||
None => start,
|
||
};
|
||
if bounds.next().is_some() || start > end {
|
||
return None;
|
||
}
|
||
ranges.push((start, end));
|
||
}
|
||
Some(ranges)
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
pub(super) fn external_agent_runner_linux_fallback_ports(
|
||
boot_id: &str,
|
||
(ephemeral_start, ephemeral_end): (u16, u16),
|
||
unprivileged_port_start: u16,
|
||
reserved_ports: &[(u16, u16)],
|
||
) -> Vec<u16> {
|
||
let start = EXTERNAL_AGENT_RUNNER_FALLBACK_PORT_START.max(unprivileged_port_start);
|
||
let mut ports = (start..=u16::MAX)
|
||
.filter(|port| {
|
||
!(ephemeral_start..=ephemeral_end).contains(port)
|
||
&& !reserved_ports.iter().any(|(reserved_start, reserved_end)| {
|
||
(*reserved_start..=*reserved_end).contains(port)
|
||
})
|
||
})
|
||
.collect::<Vec<_>>();
|
||
if !ports.is_empty() {
|
||
let digest = Sha256::digest(boot_id.as_bytes());
|
||
let seed = u64::from_be_bytes([
|
||
digest[0], digest[1], digest[2], digest[3], digest[4], digest[5], digest[6], digest[7],
|
||
]);
|
||
let offset = (seed % ports.len() as u64) as usize;
|
||
ports.rotate_left(offset);
|
||
}
|
||
ports
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
pub(super) fn read_external_agent_runner_linux_fallback_ports(boot_id: &str) -> Option<Vec<u16>> {
|
||
let ephemeral_range = fs::read_to_string(EXTERNAL_AGENT_RUNNER_LINUX_EPHEMERAL_PORT_RANGE_PATH)
|
||
.ok()
|
||
.and_then(|content| parse_external_agent_runner_linux_ephemeral_port_range(&content))?;
|
||
let unprivileged_port_start =
|
||
fs::read_to_string(EXTERNAL_AGENT_RUNNER_LINUX_UNPRIVILEGED_PORT_START_PATH)
|
||
.ok()
|
||
.and_then(|content| parse_external_agent_runner_linux_single_port(&content))?;
|
||
let reserved_ports = fs::read_to_string(EXTERNAL_AGENT_RUNNER_LINUX_RESERVED_PORTS_PATH)
|
||
.ok()
|
||
.and_then(|content| parse_external_agent_runner_linux_reserved_ports(&content))?;
|
||
Some(external_agent_runner_linux_fallback_ports(
|
||
boot_id,
|
||
ephemeral_range,
|
||
unprivileged_port_start,
|
||
&reserved_ports,
|
||
))
|
||
}
|
||
|
||
pub(crate) fn bind_loopback_listener_with_linux_fallback(seed: &str) -> io::Result<TcpListener> {
|
||
#[cfg(target_os = "linux")]
|
||
{
|
||
return bind_external_agent_runner_listener_with(
|
||
|| read_external_agent_runner_linux_fallback_ports(seed).unwrap_or_default(),
|
||
|port| TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, port)),
|
||
);
|
||
}
|
||
|
||
#[cfg(not(target_os = "linux"))]
|
||
{
|
||
let _ = seed;
|
||
TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0))
|
||
}
|
||
}
|
||
|
||
fn external_agent_runner_watchdog_tick(state: &ExternalAgentRunnerServerState) -> bool {
|
||
if !state.gui_owner_attached.load(Ordering::Acquire) {
|
||
return false;
|
||
}
|
||
match external_agent_runner_lock_is_held(&state.gui_participant_lock_path) {
|
||
Ok(true) => {
|
||
let _ = validate_external_agent_runner_gui_owner_claim_current(state);
|
||
false
|
||
}
|
||
Ok(false) | Err(_) => {
|
||
request_external_agent_runner_forced_shutdown(state);
|
||
true
|
||
}
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
pub(super) fn external_agent_runner_shutdown_if_gui_owner_lost(
|
||
state: &ExternalAgentRunnerServerState,
|
||
) -> Result<bool, String> {
|
||
Ok(external_agent_runner_watchdog_tick(state))
|
||
}
|
||
|
||
pub(super) fn spawn_external_agent_runner_gui_owner_watchdog(
|
||
state: Arc<ExternalAgentRunnerServerState>,
|
||
endpoint_path: PathBuf,
|
||
boot_id: String,
|
||
) -> Result<(), String> {
|
||
thread::Builder::new()
|
||
.name("agent-runner-gui-owner-watchdog".to_string())
|
||
.spawn(move || loop {
|
||
if !external_agent_runner_watchdog_tick(&state) {
|
||
thread::sleep(EXTERNAL_AGENT_RUNNER_GUI_OWNER_CHECK_INTERVAL);
|
||
continue;
|
||
}
|
||
thread::sleep(EXTERNAL_AGENT_RUNNER_GUI_OWNER_WATCHDOG_HARD_EXIT_TIMEOUT);
|
||
let _ = crate::agent::shutdown_game_creator_codex_app_servers();
|
||
remove_external_agent_runner_endpoint_if_boot_matches(&endpoint_path, &boot_id);
|
||
std::process::exit(1);
|
||
})
|
||
.map(|_| ())
|
||
.map_err(|error| format!("启动 Agent Runner GUI owner watchdog 失败：{error}"))
|
||
}
|
||
|
||
pub(super) fn resolve_external_agent_runner_initial_gui_owner(
|
||
gui_owner_required: bool,
|
||
gui_owner_present: bool,
|
||
) -> Result<bool, String> {
|
||
if gui_owner_required && !gui_owner_present {
|
||
return Err("GUI owner 在 Agent Runner 启动完成前已释放".to_string());
|
||
}
|
||
Ok(gui_owner_present)
|
||
}
|
||
|
||
pub(crate) fn run_external_agent_runner_server(
|
||
config_dir: impl AsRef<Path>,
|
||
gui_owner_required: bool,
|
||
) -> Result<(), String> {
|
||
let config_dir = normalize_external_agent_runner_config_dir(config_dir.as_ref())?;
|
||
EXTERNAL_AGENT_RUNNER_SERVER_PROCESS.store(true, Ordering::Release);
|
||
crate::set_game_creator_runtime_config_dir(config_dir.clone());
|
||
set_external_agent_runner_config_dir(config_dir.clone());
|
||
|
||
let boot_id = random_identifier(b"genarrative-agent-runner-boot-id")?;
|
||
crate::initialize_process_session_boot_id(&boot_id)?;
|
||
let token = random_identifier(b"genarrative-agent-runner-token")?;
|
||
let _instance_lock = acquire_external_agent_runner_instance_lock(
|
||
&external_agent_runner_lock_path(&config_dir),
|
||
&boot_id,
|
||
)?;
|
||
let gui_owner_present_at_start = resolve_external_agent_runner_initial_gui_owner(
|
||
gui_owner_required,
|
||
external_agent_runner_lock_is_held(&external_agent_runner_gui_participant_lock_path(
|
||
&config_dir,
|
||
))?,
|
||
)?;
|
||
let executable_fingerprint = current_external_agent_runner_executable_fingerprint()?;
|
||
let listener = bind_loopback_listener_with_linux_fallback(&boot_id)
|
||
.map_err(|error| format!("绑定 Agent Runner loopback 端口失败：{error}"))?;
|
||
listener
|
||
.set_nonblocking(true)
|
||
.map_err(|error| format!("配置 Agent Runner listener 失败：{error}"))?;
|
||
let port = listener
|
||
.local_addr()
|
||
.map_err(|error| format!("读取 Agent Runner loopback 地址失败：{error}"))?
|
||
.port();
|
||
let endpoint = ExternalAgentRunnerEndpoint {
|
||
protocol_version: EXTERNAL_AGENT_RUNNER_PROTOCOL_VERSION,
|
||
pid: std::process::id(),
|
||
boot_id: boot_id.clone(),
|
||
port,
|
||
token,
|
||
heartbeat_at: unix_millis(),
|
||
executable_fingerprint: Some(executable_fingerprint),
|
||
process_start_identity: external_agent_runner_process_start_identity(std::process::id())?,
|
||
};
|
||
let endpoint_path = external_agent_runner_endpoint_path(&config_dir);
|
||
write_external_agent_runner_endpoint_atomic(&endpoint_path, &endpoint)?;
|
||
let _endpoint_guard = ExternalAgentRunnerEndpointGuard {
|
||
path: endpoint_path.clone(),
|
||
boot_id,
|
||
};
|
||
let state = Arc::new(ExternalAgentRunnerServerState::new(endpoint_path, endpoint));
|
||
state
|
||
.gui_owner_attached
|
||
.store(gui_owner_present_at_start, Ordering::Release);
|
||
spawn_external_agent_runner_gui_owner_watchdog(
|
||
Arc::clone(&state),
|
||
state.endpoint_path.clone(),
|
||
state.endpoint_snapshot().boot_id,
|
||
)?;
|
||
let mut last_heartbeat = Instant::now();
|
||
let mut server_error = None;
|
||
|
||
loop {
|
||
if state.shutdown_requested.load(Ordering::Acquire) {
|
||
if state.force_shutdown_requested.load(Ordering::Acquire)
|
||
|| state.active_connections.load(Ordering::Acquire) == 0
|
||
{
|
||
break;
|
||
}
|
||
thread::sleep(EXTERNAL_AGENT_RUNNER_LOOP_INTERVAL);
|
||
continue;
|
||
}
|
||
|
||
match listener.accept() {
|
||
Ok((stream, _)) => {
|
||
let previous = state.active_connections.fetch_add(1, Ordering::AcqRel);
|
||
if previous >= EXTERNAL_AGENT_RUNNER_MAX_CONNECTIONS {
|
||
state.active_connections.fetch_sub(1, Ordering::AcqRel);
|
||
drop(stream);
|
||
continue;
|
||
}
|
||
let worker_state = Arc::clone(&state);
|
||
if thread::Builder::new()
|
||
.name("agent-runner-connection".to_string())
|
||
.spawn(move || {
|
||
let _ = handle_external_agent_runner_connection(stream, worker_state);
|
||
})
|
||
.is_err()
|
||
{
|
||
state.active_connections.fetch_sub(1, Ordering::AcqRel);
|
||
}
|
||
}
|
||
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
|
||
Err(error) => {
|
||
server_error = Some(format!("接受 Agent Runner 连接失败：{error}"));
|
||
break;
|
||
}
|
||
}
|
||
|
||
if last_heartbeat.elapsed() >= EXTERNAL_AGENT_RUNNER_HEARTBEAT_INTERVAL {
|
||
if let Err(error) = refresh_external_agent_runner_heartbeat(&state) {
|
||
server_error = Some(error);
|
||
break;
|
||
}
|
||
last_heartbeat = Instant::now();
|
||
}
|
||
thread::sleep(EXTERNAL_AGENT_RUNNER_LOOP_INTERVAL);
|
||
}
|
||
|
||
state.shutdown_requested.store(true, Ordering::Release);
|
||
let forced = state.force_shutdown_requested.load(Ordering::Acquire);
|
||
let forced_deadline =
|
||
forced.then(|| Instant::now() + EXTERNAL_AGENT_RUNNER_FORCED_TOTAL_DRAIN_TIMEOUT);
|
||
let worker_deadline = Instant::now()
|
||
+ if forced {
|
||
EXTERNAL_AGENT_RUNNER_FORCED_WORKER_DRAIN_TIMEOUT
|
||
} else {
|
||
EXTERNAL_AGENT_RUNNER_IO_TIMEOUT
|
||
};
|
||
while state.active_connections.load(Ordering::Acquire) > 0 && Instant::now() < worker_deadline {
|
||
thread::sleep(EXTERNAL_AGENT_RUNNER_LOOP_INTERVAL);
|
||
}
|
||
let forced_roots = forced.then(|| {
|
||
let roots = state.known_roots_snapshot();
|
||
crate::interrupt_game_creator_agent_runtime_provider_requests_for_roots(&roots);
|
||
roots
|
||
});
|
||
let process_timeout = forced_deadline
|
||
.map(|deadline| deadline.saturating_duration_since(Instant::now()))
|
||
.unwrap_or(Duration::from_secs(3));
|
||
let mut process_shutdown = crate::shutdown_all_process_sessions_and_wait(process_timeout);
|
||
if forced {
|
||
let roots = forced_roots.as_deref().unwrap_or_default();
|
||
let provider_deadline = forced_deadline.expect("forced shutdown has a deadline");
|
||
while crate::game_creator_agent_runtime_provider_request_count_for_roots(roots) > 0
|
||
&& Instant::now() < provider_deadline
|
||
{
|
||
thread::sleep(EXTERNAL_AGENT_RUNNER_LOOP_INTERVAL);
|
||
}
|
||
if crate::game_creator_agent_runtime_provider_request_count_for_roots(roots) > 0 {
|
||
let provider_error = "Runner 退出前未能中断全部 Provider 请求".to_string();
|
||
process_shutdown = Err(match process_shutdown {
|
||
Ok(()) => provider_error,
|
||
Err(process_error) => format!("{process_error}；{provider_error}"),
|
||
});
|
||
}
|
||
}
|
||
if let Err(app_server_error) = crate::agent::shutdown_game_creator_codex_app_servers() {
|
||
process_shutdown = Err(match process_shutdown {
|
||
Ok(()) => app_server_error,
|
||
Err(process_error) => format!("{process_error}；{app_server_error}"),
|
||
});
|
||
}
|
||
if let Some(error) = server_error {
|
||
Err(match process_shutdown {
|
||
Ok(()) => error,
|
||
Err(process_error) => format!("{error}；{process_error}"),
|
||
})
|
||
} else {
|
||
process_shutdown
|
||
}
|
||
}
|