kdletters
bdf7ed288a
- 将固定四切片与固定 drawImage 降为推荐路径 - 在导航前注入 Canvas/WebGL 平台素材运行时观测 - 对缺少核心素材渲染证据的 desktop/mobile 视口回灌同一 Codex thread 整改 - 补充直连验收、浏览器脚本与真实 Chrome fixture 测试 - 同步 AGC 直连 Runtime 实施计划
1652 lines
64 KiB
Rust
1652 lines
64 KiB
Rust
use super::*;
|
||
|
||
pub(crate) fn validate_process_session_command_spec(
|
||
spec: &ProjectCommandSpec,
|
||
) -> Result<(), String> {
|
||
const DETACH_ARGUMENTS: &[&str] = &[
|
||
"--background",
|
||
"--daemon",
|
||
"--daemonize",
|
||
"--detach",
|
||
"--fork",
|
||
];
|
||
if spec.arguments.iter().any(|argument| {
|
||
let argument = argument.trim().to_ascii_lowercase();
|
||
DETACH_ARGUMENTS.contains(&argument.as_str())
|
||
}) {
|
||
return Err("command.start 不允许 daemonize、detach、fork 或 background 参数".to_string());
|
||
}
|
||
if spec.program == "npm"
|
||
&& spec.arguments.first().map(String::as_str) == Some("run")
|
||
&& spec.arguments.len() >= 2
|
||
{
|
||
let package_path = spec.cwd.join("package.json");
|
||
let metadata = fs::metadata(&package_path)
|
||
.map_err(|error| format!("command.start 无法读取 package.json：{error}"))?;
|
||
if !metadata.is_file() || metadata.len() > 256 * 1024 {
|
||
return Err("command.start package.json 必须是 256 KiB 内的普通文件".to_string());
|
||
}
|
||
let package = fs::read_to_string(&package_path)
|
||
.map_err(|error| format!("command.start 无法读取 package.json：{error}"))?;
|
||
let package = serde_json::from_str::<serde_json::Value>(&package)
|
||
.map_err(|error| format!("command.start package.json JSON 无效：{error}"))?;
|
||
let script_name = &spec.arguments[1];
|
||
let script = package
|
||
.get("scripts")
|
||
.and_then(|value| value.get(script_name))
|
||
.and_then(serde_json::Value::as_str)
|
||
.ok_or_else(|| format!("command.start npm script 不存在：{script_name}"))?;
|
||
let normalized = script.to_ascii_lowercase();
|
||
if [
|
||
"nohup", "setsid", "disown", "start /b", "--detach", "--daemon",
|
||
]
|
||
.iter()
|
||
.any(|marker| normalized.contains(marker))
|
||
|| normalized.trim_end().ends_with('&')
|
||
{
|
||
return Err("command.start npm script 包含已知脱离 Runner 的启动方式".to_string());
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
pub(super) fn process_session_command_builder(
|
||
launch: &ProjectCommandLaunchSpec,
|
||
#[cfg(target_os = "linux")] bridge: &ProcessSessionBridgeServer,
|
||
) -> Result<CommandBuilder, String> {
|
||
#[cfg(windows)]
|
||
let is_npm_launch = launch
|
||
.executable
|
||
.file_name()
|
||
.and_then(|name| name.to_str())
|
||
.is_some_and(|name| name.eq_ignore_ascii_case("npm.cmd"))
|
||
|| launch.arguments.first().is_some_and(|argument| {
|
||
std::path::Path::new(argument)
|
||
.file_name()
|
||
.and_then(|name| name.to_str())
|
||
.is_some_and(|name| name.eq_ignore_ascii_case("npm-cli.js"))
|
||
});
|
||
#[cfg(target_os = "linux")]
|
||
let mut command = {
|
||
let current_executable = std::env::current_exe()
|
||
.map_err(|error| format!("定位 process session child wrapper 失败：{error}"))?;
|
||
let mut command = CommandBuilder::new(current_executable);
|
||
#[cfg(not(test))]
|
||
{
|
||
command.arg(PROCESS_SESSION_CHILD_MODE);
|
||
}
|
||
#[cfg(test)]
|
||
{
|
||
command.args([
|
||
"--exact",
|
||
"process_session::tests::process_session_child_wrapper_fixture",
|
||
"--nocapture",
|
||
"--test-threads=1",
|
||
]);
|
||
}
|
||
command
|
||
};
|
||
#[cfg(not(target_os = "linux"))]
|
||
let mut command = {
|
||
#[cfg(windows)]
|
||
{
|
||
if launch
|
||
.executable
|
||
.file_name()
|
||
.and_then(|name| name.to_str())
|
||
.is_some_and(|name| name.eq_ignore_ascii_case("npm.cmd"))
|
||
{
|
||
let npm_directory = launch
|
||
.executable
|
||
.parent()
|
||
.ok_or_else(|| "command.start 无法定位 Windows npm 安装目录".to_string())?;
|
||
let node_executable = npm_directory.join("node.exe");
|
||
let npm_cli = npm_directory.join("node_modules/npm/bin/npm-cli.js");
|
||
if !node_executable.is_file() || !npm_cli.is_file() {
|
||
return Err(
|
||
"command.start Windows npm 安装缺少 node.exe 或 npm-cli.js".to_string()
|
||
);
|
||
}
|
||
let node_executable = node_executable.to_string_lossy();
|
||
let npm_cli = npm_cli.to_string_lossy();
|
||
let mut command = CommandBuilder::new(
|
||
node_executable
|
||
.strip_prefix(r"\\?\")
|
||
.unwrap_or(&node_executable),
|
||
);
|
||
command.arg(npm_cli.strip_prefix(r"\\?\").unwrap_or(&npm_cli));
|
||
command.args(&launch.arguments);
|
||
command
|
||
} else {
|
||
let mut command = CommandBuilder::new(&launch.executable);
|
||
command.args(&launch.arguments);
|
||
command
|
||
}
|
||
}
|
||
#[cfg(not(windows))]
|
||
{
|
||
let mut command = CommandBuilder::new(&launch.executable);
|
||
command.args(&launch.arguments);
|
||
command
|
||
}
|
||
};
|
||
#[cfg(windows)]
|
||
{
|
||
let cwd = launch.cwd.to_string_lossy();
|
||
command.cwd(cwd.strip_prefix(r"\\?\").unwrap_or(&cwd));
|
||
}
|
||
#[cfg(not(windows))]
|
||
command.cwd(&launch.cwd);
|
||
command.env_clear();
|
||
#[cfg(not(target_os = "linux"))]
|
||
for (name, value) in &launch.environment {
|
||
command.env(name, value);
|
||
}
|
||
#[cfg(windows)]
|
||
if is_npm_launch {
|
||
let node_executable = launch.executable.to_string_lossy();
|
||
let node_executable = node_executable
|
||
.strip_prefix(r"\\?\")
|
||
.unwrap_or(&node_executable);
|
||
command.env("npm_node_execpath", node_executable);
|
||
command.env("NODE", node_executable);
|
||
command.env("npm_config_node_gyp", "");
|
||
// Windows 环境变量名不区分大小写。先移除继承的拼写，避免
|
||
// CommandBuilder 更新值后仍保留 `ComSpec` 而隐藏 npm 的小写键。
|
||
command.env_remove("ComSpec");
|
||
command.env("npm_config_script_shell", r"C:\Windows\System32\cmd.exe");
|
||
}
|
||
#[cfg(target_os = "linux")]
|
||
{
|
||
command.env(
|
||
PROCESS_SESSION_OWNER_PID_ENV,
|
||
std::process::id().to_string(),
|
||
);
|
||
command.env(PROCESS_SESSION_BRIDGE_ENDPOINT_ENV, bridge.endpoint());
|
||
command.env(PROCESS_SESSION_BRIDGE_NONCE_ENV, bridge.nonce_hex());
|
||
}
|
||
Ok(command)
|
||
}
|
||
|
||
pub(crate) fn validate_process_session_start_preflight_at(
|
||
root: &Path,
|
||
identity: &ProcessSessionIdentity,
|
||
spec: &ProjectCommandSpec,
|
||
) -> Result<(), String> {
|
||
validate_process_session_identity(identity)?;
|
||
if identity.project_id != game_creator_agent_runtime_context_project_id(root)? {
|
||
return Err("command.start projectId 与当前项目不匹配".to_string());
|
||
}
|
||
validate_process_session_command_spec(spec)?;
|
||
if find_existing_start_action_record(root, identity)?.is_some() {
|
||
return Ok(());
|
||
}
|
||
let records = active_process_session_records_at(root, None, None)?;
|
||
#[cfg(target_os = "linux")]
|
||
let pending = pending_process_launch_registry()
|
||
.lock()
|
||
.map_err(|_| "pending process launch registry 锁已损坏".to_string())?
|
||
.launches
|
||
.values()
|
||
.filter(|launch| launch.root == root)
|
||
.cloned()
|
||
.collect::<Vec<_>>();
|
||
if let Some(record) = records
|
||
.iter()
|
||
.find(|record| record.needs_reconciliation || record.status == "needs-reconciliation")
|
||
{
|
||
return Err(format!(
|
||
"项目存在待人工核对的进程会话 {}，禁止启动新会话",
|
||
record.process_id
|
||
));
|
||
}
|
||
#[cfg(target_os = "linux")]
|
||
let pending_project_count = pending.len();
|
||
#[cfg(not(target_os = "linux"))]
|
||
let pending_project_count = 0;
|
||
if records.len().saturating_add(pending_project_count) >= PROCESS_SESSION_MAX_PER_PROJECT {
|
||
return Err(format!(
|
||
"当前项目最多同时运行 {PROCESS_SESSION_MAX_PER_PROJECT} 个 process session"
|
||
));
|
||
}
|
||
let agent_count = records
|
||
.iter()
|
||
.filter(|record| record.agent_id == identity.agent_id)
|
||
.count();
|
||
#[cfg(target_os = "linux")]
|
||
let agent_count = agent_count.saturating_add(
|
||
pending
|
||
.iter()
|
||
.filter(|launch| launch.agent_id == identity.agent_id)
|
||
.count(),
|
||
);
|
||
if agent_count >= PROCESS_SESSION_MAX_PER_AGENT {
|
||
return Err(format!(
|
||
"当前 Agent 最多同时运行 {PROCESS_SESSION_MAX_PER_AGENT} 个 process session"
|
||
));
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
pub(crate) fn start_process_session_at(
|
||
root: &Path,
|
||
identity: ProcessSessionIdentity,
|
||
spec: &ProjectCommandSpec,
|
||
source_fingerprint_before: String,
|
||
) -> Result<ProcessSessionPollResult, String> {
|
||
let launch =
|
||
prepare_project_command_launch_spec(root, spec).map_err(|error| error.to_string())?;
|
||
start_prepared_process_session_at(
|
||
root,
|
||
identity,
|
||
spec,
|
||
&launch,
|
||
source_fingerprint_before,
|
||
|| Ok(()),
|
||
)
|
||
.map_err(|error| error.to_string())
|
||
}
|
||
|
||
pub(crate) fn start_prepared_process_session_at<F>(
|
||
root: &Path,
|
||
identity: ProcessSessionIdentity,
|
||
spec: &ProjectCommandSpec,
|
||
launch: &ProjectCommandLaunchSpec,
|
||
source_fingerprint_before: String,
|
||
durable_commit: F,
|
||
) -> Result<ProcessSessionPollResult, ProjectCommandError>
|
||
where
|
||
F: FnOnce() -> Result<(), String>,
|
||
{
|
||
#[cfg(target_os = "linux")]
|
||
{
|
||
start_linux_process_session_at(
|
||
root,
|
||
identity,
|
||
spec,
|
||
launch,
|
||
source_fingerprint_before,
|
||
durable_commit,
|
||
)
|
||
}
|
||
#[cfg(not(target_os = "linux"))]
|
||
{
|
||
start_legacy_process_session_at(
|
||
root,
|
||
identity,
|
||
spec,
|
||
launch,
|
||
source_fingerprint_before,
|
||
durable_commit,
|
||
)
|
||
}
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
fn start_linux_process_session_at<F>(
|
||
root: &Path,
|
||
identity: ProcessSessionIdentity,
|
||
spec: &ProjectCommandSpec,
|
||
launch: &ProjectCommandLaunchSpec,
|
||
source_fingerprint_before: String,
|
||
durable_commit: F,
|
||
) -> Result<ProcessSessionPollResult, ProjectCommandError>
|
||
where
|
||
F: FnOnce() -> Result<(), String>,
|
||
{
|
||
validate_process_session_start_preflight_at(root, &identity, spec)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Preflight, error))?;
|
||
if let Some(mut existing) = find_existing_start_action_record(root, &identity)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Preflight, error))?
|
||
{
|
||
if matches!(
|
||
existing.status.as_str(),
|
||
"prepared" | "launching" | "running" | "terminating"
|
||
) {
|
||
if existing.owner_boot_id == process_session_boot_id()
|
||
&& live_process_session(&existing.process_id)
|
||
.map_err(|error| {
|
||
ProjectCommandError::new(ProjectCommandErrorStage::LaunchUnknown, error)
|
||
})?
|
||
.is_some()
|
||
{
|
||
return poll_process_session_at(
|
||
root,
|
||
&identity,
|
||
&existing.process_id,
|
||
None,
|
||
Some(0),
|
||
Some(0),
|
||
)
|
||
.map_err(|error| {
|
||
ProjectCommandError::new(ProjectCommandErrorStage::LaunchUnknown, error)
|
||
});
|
||
}
|
||
reconcile_stale_active_process_session(&mut existing);
|
||
write_process_session_record(root, &existing).map_err(|error| {
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::AuditLog,
|
||
format!("command.start 旧启动状态无法写入 reconciliation：{error}"),
|
||
)
|
||
})?;
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::LaunchUnknown,
|
||
"command.start 已提交启动但缺少当前 Runner 句柄，禁止自动重放",
|
||
));
|
||
}
|
||
return poll_process_session_at(
|
||
root,
|
||
&identity,
|
||
&existing.process_id,
|
||
None,
|
||
Some(0),
|
||
Some(0),
|
||
)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Execution, error));
|
||
}
|
||
|
||
let process_id = process_session_id(&identity);
|
||
let command_id = format!(
|
||
"cmd-{}",
|
||
&format!(
|
||
"{:x}",
|
||
Sha256::digest(
|
||
serde_json::to_vec(&serde_json::json!({
|
||
"program": spec.program,
|
||
"args": spec.arguments,
|
||
"cwd": spec.cwd_relative,
|
||
}))
|
||
.unwrap_or_default()
|
||
)
|
||
)[..24]
|
||
);
|
||
let _pending_launch = reserve_pending_process_launch(root, &identity.agent_id, &process_id)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Preflight, error))?;
|
||
let bridge_server = ProcessSessionBridgeServer::bind()
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Preflight, error))?;
|
||
let pair = native_pty_system()
|
||
.openpty(PtySize {
|
||
rows: 30,
|
||
cols: 120,
|
||
pixel_width: 0,
|
||
pixel_height: 0,
|
||
})
|
||
.map_err(|error| {
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Preflight,
|
||
format!("创建 command.start PTY 失败：{error}"),
|
||
)
|
||
})?;
|
||
let reader = pair.master.try_clone_reader().map_err(|error| {
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Preflight,
|
||
format!("克隆 command.start PTY reader 失败：{error}"),
|
||
)
|
||
})?;
|
||
let writer = pair.master.take_writer().map_err(|error| {
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Preflight,
|
||
format!("取得 command.start PTY writer 失败：{error}"),
|
||
)
|
||
})?;
|
||
let command = process_session_command_builder(launch, &bridge_server)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Preflight, error))?;
|
||
let mut child = pair.slave.spawn_command(command).map_err(|error| {
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Spawn,
|
||
format!("启动 command.start wrapper 失败：{error}"),
|
||
)
|
||
})?;
|
||
drop(pair.slave);
|
||
let process_group_leader = pair.master.process_group_leader().or_else(|| {
|
||
child
|
||
.process_id()
|
||
.and_then(|value| i32::try_from(value).ok())
|
||
});
|
||
let peer_pid = child.process_id().ok_or_else(|| {
|
||
let termination = terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::Preflight,
|
||
"command.start wrapper 缺少 pid",
|
||
termination,
|
||
)
|
||
})?;
|
||
let process_group_leader = Some(
|
||
process_group_leader
|
||
.or_else(|| i32::try_from(peer_pid).ok())
|
||
.ok_or_else(|| {
|
||
let termination = terminate_pending_process_session_child(&mut child, None);
|
||
project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::Preflight,
|
||
"command.start wrapper 缺少进程组身份",
|
||
termination,
|
||
)
|
||
})?,
|
||
);
|
||
activate_pending_process_launch(
|
||
&process_id,
|
||
process_group_leader.expect("process group leader validated"),
|
||
)
|
||
.map_err(|error| {
|
||
let termination = terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::Preflight,
|
||
error,
|
||
termination,
|
||
)
|
||
})?;
|
||
let mut bridge = match bridge_server.accept(peer_pid, Duration::from_secs(3)) {
|
||
Ok(bridge) => bridge,
|
||
Err(error) => {
|
||
let termination =
|
||
terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
return Err(project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::Preflight,
|
||
error,
|
||
termination,
|
||
));
|
||
}
|
||
};
|
||
if let Err(error) = bridge.send_prepare(launch, spec) {
|
||
let termination = terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
return Err(project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::Preflight,
|
||
error,
|
||
termination,
|
||
));
|
||
}
|
||
match bridge.wait_sandbox_ready(Duration::from_secs(4)) {
|
||
Ok(ProcessSessionSandboxReadyVerdict::Ready) => {}
|
||
Ok(ProcessSessionSandboxReadyVerdict::Failed { failure_kind }) => {
|
||
let termination =
|
||
terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
return Err(project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::Preflight,
|
||
format!("command.start sandbox ready 前失败：{failure_kind}"),
|
||
termination,
|
||
));
|
||
}
|
||
Err(error) => {
|
||
let termination =
|
||
terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
return Err(project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::Preflight,
|
||
error,
|
||
termination,
|
||
));
|
||
}
|
||
}
|
||
|
||
let sandbox_ready_at = unix_timestamp();
|
||
let mut durable_record = initial_process_session_record(
|
||
&identity,
|
||
&process_id,
|
||
&command_id,
|
||
spec,
|
||
Some(launch),
|
||
&source_fingerprint_before,
|
||
"launching",
|
||
);
|
||
durable_record.sandbox_establishment = "established".to_string();
|
||
durable_record.target_exec = "not-attempted".to_string();
|
||
durable_record.started_at = sandbox_ready_at;
|
||
durable_record.sandbox_ready_at = Some(sandbox_ready_at);
|
||
if let Err(error) = durable_commit() {
|
||
let _ = bridge.abort_launch();
|
||
let termination = terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
return Err(project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::DurableCommit,
|
||
error,
|
||
termination,
|
||
));
|
||
}
|
||
if let Err(error) = write_process_session_record(root, &durable_record) {
|
||
let _ = bridge.abort_launch();
|
||
let termination = terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
return Err(project_command_error_after_pending_termination(
|
||
ProjectCommandErrorStage::DurableCommit,
|
||
format!("写入 command.start commit record 失败：{error}"),
|
||
termination,
|
||
));
|
||
}
|
||
|
||
if let Err(error) = bridge.commit_exec() {
|
||
let termination = terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
persist_process_session_launch_unknown(root, &mut durable_record);
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::LaunchUnknown,
|
||
format!("{error}；{termination}"),
|
||
));
|
||
}
|
||
let exec = bridge.wait_exec(Duration::from_secs(4));
|
||
match exec {
|
||
Ok(ProcessSessionExecVerdict::TargetExecFailed { errno }) => {
|
||
let termination =
|
||
terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
let needs_reconciliation = !termination.confirmed;
|
||
mark_process_session_launch_record(
|
||
&mut durable_record,
|
||
if needs_reconciliation {
|
||
"needs-reconciliation"
|
||
} else {
|
||
"failed"
|
||
},
|
||
"failed",
|
||
Some("target-exec-failed"),
|
||
needs_reconciliation,
|
||
);
|
||
write_process_session_record(root, &durable_record).map_err(|error| {
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::AuditLog,
|
||
format!(
|
||
"command.start target exec 失败后终态无法落盘：errno={errno}；{termination}；{error}"
|
||
),
|
||
)
|
||
})?;
|
||
if needs_reconciliation {
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Execution,
|
||
format!(
|
||
"command.start target exec 失败但 wrapper 回收无法确认：errno={errno}；{termination}"
|
||
),
|
||
));
|
||
}
|
||
return poll_process_session_at(root, &identity, &process_id, None, Some(0), Some(0))
|
||
.map_err(|error| {
|
||
ProjectCommandError::new(ProjectCommandErrorStage::AuditLog, error)
|
||
});
|
||
}
|
||
Ok(ProcessSessionExecVerdict::LaunchUnknown) => {
|
||
let termination =
|
||
terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
persist_process_session_launch_unknown(root, &mut durable_record);
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::LaunchUnknown,
|
||
format!("process session wrapper 报告 launch unknown；{termination}"),
|
||
));
|
||
}
|
||
Err(error) => {
|
||
let termination =
|
||
terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
persist_process_session_launch_unknown(root, &mut durable_record);
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::LaunchUnknown,
|
||
format!("{error}；{termination}"),
|
||
));
|
||
}
|
||
Ok(ProcessSessionExecVerdict::Established) => {}
|
||
}
|
||
|
||
let exec_established_at = unix_timestamp();
|
||
durable_record.status = "running".to_string();
|
||
durable_record.target_exec = "established".to_string();
|
||
durable_record.exec_established_at = Some(exec_established_at);
|
||
durable_record.stdin_open = true;
|
||
durable_record.updated_at = exec_established_at;
|
||
if let Err(error) = write_process_session_record(root, &durable_record) {
|
||
let termination = terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
persist_process_session_launch_unknown(root, &mut durable_record);
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Execution,
|
||
format!("command.start exec-established 状态无法落盘：{error}；{termination}"),
|
||
));
|
||
}
|
||
|
||
let (control_tx, control_rx) = std::sync::mpsc::channel();
|
||
let live = Arc::new(LiveProcessSession {
|
||
root: root.to_path_buf(),
|
||
identity,
|
||
process_id: process_id.clone(),
|
||
command_id,
|
||
program: spec.program.clone(),
|
||
cwd: spec.cwd_relative.clone(),
|
||
sandbox_backend: launch.sandbox_backend.clone(),
|
||
sandbox_mode: launch.sandbox_mode.clone(),
|
||
network_access: launch.network_access.clone(),
|
||
sandbox_profile_version: launch.sandbox_profile_version.clone(),
|
||
sandbox_establishment: "established".to_string(),
|
||
target_exec: "established".to_string(),
|
||
sandbox_ready_at: Some(sandbox_ready_at),
|
||
exec_established_at: Some(exec_established_at),
|
||
source_fingerprint_before,
|
||
started_at: durable_record.started_at,
|
||
output: Mutex::new(ProcessOutputState::running()),
|
||
output_changed: Condvar::new(),
|
||
writer: Mutex::new(Some(writer)),
|
||
master: Mutex::new(Some(pair.master)),
|
||
control: control_tx,
|
||
});
|
||
process_session_registry()
|
||
.lock()
|
||
.map_err(|_| {
|
||
let termination =
|
||
terminate_pending_process_session_child(&mut child, process_group_leader);
|
||
persist_process_session_launch_unknown(root, &mut durable_record);
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Execution,
|
||
format!("process session registry 锁已损坏；{termination}"),
|
||
)
|
||
})?
|
||
.sessions
|
||
.insert(process_id.clone(), Arc::clone(&live));
|
||
|
||
let reader_live = Arc::clone(&live);
|
||
thread::spawn(move || drain_process_session_output(reader_live, reader));
|
||
let supervisor_live = Arc::clone(&live);
|
||
let timeout_seconds = spec.timeout_seconds;
|
||
thread::spawn(move || {
|
||
supervise_process_session(
|
||
supervisor_live,
|
||
&mut child,
|
||
control_rx,
|
||
timeout_seconds,
|
||
process_group_leader,
|
||
bridge,
|
||
)
|
||
});
|
||
|
||
poll_process_session_at(root, &live.identity, &process_id, None, Some(0), Some(0))
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Execution, error))
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
struct PendingProcessTermination {
|
||
confirmed: bool,
|
||
summary: String,
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
impl std::fmt::Display for PendingProcessTermination {
|
||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||
formatter.write_str(&self.summary)
|
||
}
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
fn terminate_pending_process_session_child(
|
||
child: &mut Box<dyn Child + Send + Sync>,
|
||
process_group_leader: Option<i32>,
|
||
) -> PendingProcessTermination {
|
||
let group = process_group_leader.filter(|value| *value > 0);
|
||
let group_result = group.map(|group| {
|
||
let result = unsafe { libc::kill(-group, libc::SIGKILL) };
|
||
if result == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) {
|
||
Ok(())
|
||
} else {
|
||
Err(std::io::Error::last_os_error())
|
||
}
|
||
});
|
||
let _ = child.kill();
|
||
let wait = child.wait();
|
||
let (confirmed, summary) = match (group_result, wait) {
|
||
(Some(Ok(())), Ok(_)) => (true, "wrapper 进程组已终止并回收".to_string()),
|
||
(None, Ok(_)) => (false, "wrapper 主进程已回收但缺少进程组身份".to_string()),
|
||
(Some(Err(error)), Ok(_)) => (
|
||
false,
|
||
format!("wrapper 主进程已回收但进程组终止失败：{error}"),
|
||
),
|
||
(_, Err(error)) => (false, format!("wrapper 进程回收失败：{error}")),
|
||
};
|
||
PendingProcessTermination { confirmed, summary }
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
fn project_command_error_after_pending_termination(
|
||
confirmed_stage: ProjectCommandErrorStage,
|
||
message: impl Into<String>,
|
||
termination: PendingProcessTermination,
|
||
) -> ProjectCommandError {
|
||
let stage = if termination.confirmed {
|
||
confirmed_stage
|
||
} else {
|
||
ProjectCommandErrorStage::LaunchUnknown
|
||
};
|
||
ProjectCommandError::new(stage, format!("{}；{termination}", message.into()))
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
fn mark_process_session_launch_record(
|
||
record: &mut ProcessSessionRecord,
|
||
status: &str,
|
||
target_exec: &str,
|
||
launch_failure_kind: Option<&str>,
|
||
needs_reconciliation: bool,
|
||
) {
|
||
record.status = status.to_string();
|
||
record.target_exec = target_exec.to_string();
|
||
record.launch_failure_kind = launch_failure_kind.map(str::to_string);
|
||
record.stdin_open = false;
|
||
record.needs_reconciliation = needs_reconciliation;
|
||
record.terminal_at = Some(unix_timestamp());
|
||
record.updated_at = unix_timestamp();
|
||
}
|
||
|
||
#[cfg(target_os = "linux")]
|
||
fn persist_process_session_launch_unknown(root: &Path, record: &mut ProcessSessionRecord) {
|
||
mark_process_session_launch_record(
|
||
record,
|
||
"needs-reconciliation",
|
||
"unknown",
|
||
Some("launch-unknown"),
|
||
true,
|
||
);
|
||
let _ = write_process_session_record(root, record);
|
||
}
|
||
|
||
#[cfg(not(target_os = "linux"))]
|
||
fn start_legacy_process_session_at<F>(
|
||
root: &Path,
|
||
identity: ProcessSessionIdentity,
|
||
spec: &ProjectCommandSpec,
|
||
launch: &ProjectCommandLaunchSpec,
|
||
source_fingerprint_before: String,
|
||
durable_commit: F,
|
||
) -> Result<ProcessSessionPollResult, ProjectCommandError>
|
||
where
|
||
F: FnOnce() -> Result<(), String>,
|
||
{
|
||
validate_process_session_start_preflight_at(root, &identity, spec)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Preflight, error))?;
|
||
if let Some(existing) = find_existing_start_action_record(root, &identity)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Preflight, error))?
|
||
{
|
||
if matches!(
|
||
existing.status.as_str(),
|
||
"prepared" | "launching" | "running" | "terminating"
|
||
) {
|
||
if existing.owner_boot_id == process_session_boot_id()
|
||
&& live_process_session(&existing.process_id)
|
||
.map_err(|error| {
|
||
ProjectCommandError::new(ProjectCommandErrorStage::LaunchUnknown, error)
|
||
})?
|
||
.is_some()
|
||
{
|
||
return poll_process_session_at(
|
||
root,
|
||
&identity,
|
||
&existing.process_id,
|
||
None,
|
||
Some(0),
|
||
Some(0),
|
||
)
|
||
.map_err(|error| {
|
||
ProjectCommandError::new(ProjectCommandErrorStage::LaunchUnknown, error)
|
||
});
|
||
}
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::LaunchUnknown,
|
||
"command.start 已进入可能启动阶段但缺少当前 Runner 句柄，禁止自动重放",
|
||
));
|
||
}
|
||
return poll_process_session_at(
|
||
root,
|
||
&identity,
|
||
&existing.process_id,
|
||
None,
|
||
Some(0),
|
||
Some(0),
|
||
)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Execution, error));
|
||
}
|
||
|
||
durable_commit().map_err(|error| {
|
||
ProjectCommandError::new(ProjectCommandErrorStage::DurableCommit, error)
|
||
})?;
|
||
|
||
let process_id = process_session_id(&identity);
|
||
let command_id = format!(
|
||
"cmd-{}",
|
||
&format!(
|
||
"{:x}",
|
||
Sha256::digest(
|
||
serde_json::to_vec(&serde_json::json!({
|
||
"program": spec.program,
|
||
"args": spec.arguments,
|
||
"cwd": spec.cwd_relative,
|
||
}))
|
||
.unwrap_or_default()
|
||
)
|
||
)[..24]
|
||
);
|
||
|
||
let mut durable_record = initial_process_session_record(
|
||
&identity,
|
||
&process_id,
|
||
&command_id,
|
||
spec,
|
||
Some(launch),
|
||
&source_fingerprint_before,
|
||
"prepared",
|
||
);
|
||
write_process_session_record(root, &durable_record)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::AuditLog, error))?;
|
||
durable_record.status = "launching".to_string();
|
||
durable_record.updated_at = unix_timestamp();
|
||
write_process_session_record(root, &durable_record)
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::AuditLog, error))?;
|
||
|
||
let pair = native_pty_system()
|
||
.openpty(PtySize {
|
||
rows: 30,
|
||
cols: 120,
|
||
pixel_width: 0,
|
||
pixel_height: 0,
|
||
})
|
||
.map_err(|error| {
|
||
process_session_launch_failed(
|
||
root,
|
||
&mut durable_record,
|
||
format!("创建 command.start PTY 失败：{error}"),
|
||
)
|
||
})
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Execution, error))?;
|
||
let command = process_session_command_builder(launch)
|
||
.map_err(|error| process_session_launch_failed(root, &mut durable_record, error))
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Execution, error))?;
|
||
let mut child = pair
|
||
.slave
|
||
.spawn_command(command)
|
||
.map_err(|error| {
|
||
process_session_launch_failed(
|
||
root,
|
||
&mut durable_record,
|
||
format!("启动 command.start {} 失败：{error}", spec.program),
|
||
)
|
||
})
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Execution, error))?;
|
||
drop(pair.slave);
|
||
let reader = pair.master.try_clone_reader().map_err(|error| {
|
||
let _ = child.kill();
|
||
let _ = child.wait();
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Execution,
|
||
process_session_launch_failed(
|
||
root,
|
||
&mut durable_record,
|
||
format!("克隆 command.start PTY reader 失败：{error}"),
|
||
),
|
||
)
|
||
})?;
|
||
let writer = pair.master.take_writer().map_err(|error| {
|
||
let _ = child.kill();
|
||
let _ = child.wait();
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Execution,
|
||
process_session_launch_failed(
|
||
root,
|
||
&mut durable_record,
|
||
format!("取得 command.start PTY writer 失败：{error}"),
|
||
),
|
||
)
|
||
})?;
|
||
#[cfg(windows)]
|
||
let windows_job = match WindowsProcessJob::assign(child.as_ref()) {
|
||
Ok(job) => job,
|
||
Err(error) => {
|
||
let _ = child.kill();
|
||
let _ = child.wait();
|
||
durable_record.status = "needs-reconciliation".to_string();
|
||
durable_record.sandbox_establishment = "unknown".to_string();
|
||
durable_record.target_exec = "unknown".to_string();
|
||
durable_record.launch_failure_kind = Some("launch-unknown".to_string());
|
||
durable_record.needs_reconciliation = true;
|
||
durable_record.terminal_at = Some(unix_timestamp());
|
||
durable_record.updated_at = unix_timestamp();
|
||
let _ = write_process_session_record(root, &durable_record);
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::LaunchUnknown,
|
||
format!("command.start 已创建进程但无法纳入 Windows Job Object：{error}"),
|
||
));
|
||
}
|
||
};
|
||
#[cfg(unix)]
|
||
let process_group_leader = pair.master.process_group_leader().or_else(|| {
|
||
child
|
||
.process_id()
|
||
.and_then(|value| i32::try_from(value).ok())
|
||
});
|
||
#[cfg(not(unix))]
|
||
let process_group_leader: Option<i32> = None;
|
||
|
||
let (control_tx, control_rx) = std::sync::mpsc::channel();
|
||
let launch_established_at = unix_timestamp();
|
||
let live = Arc::new(LiveProcessSession {
|
||
root: root.to_path_buf(),
|
||
identity,
|
||
process_id: process_id.clone(),
|
||
command_id,
|
||
program: spec.program.clone(),
|
||
cwd: spec.cwd_relative.clone(),
|
||
sandbox_backend: launch.sandbox_backend.clone(),
|
||
sandbox_mode: launch.sandbox_mode.clone(),
|
||
network_access: launch.network_access.clone(),
|
||
sandbox_profile_version: launch.sandbox_profile_version.clone(),
|
||
sandbox_establishment: "established".to_string(),
|
||
target_exec: "established".to_string(),
|
||
sandbox_ready_at: Some(launch_established_at),
|
||
exec_established_at: Some(launch_established_at),
|
||
source_fingerprint_before,
|
||
started_at: launch_established_at,
|
||
output: Mutex::new(ProcessOutputState::running()),
|
||
output_changed: Condvar::new(),
|
||
writer: Mutex::new(Some(writer)),
|
||
master: Mutex::new(Some(pair.master)),
|
||
#[cfg(windows)]
|
||
job: Mutex::new(Some(windows_job)),
|
||
control: control_tx,
|
||
});
|
||
|
||
let record = {
|
||
let output = live.output.lock().map_err(|_| {
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::Execution,
|
||
"process session output 锁已损坏",
|
||
)
|
||
})?;
|
||
process_session_record_from_live(&live, &output)
|
||
};
|
||
if let Err(error) = write_process_session_record(root, &record) {
|
||
let _ = child.kill();
|
||
let _ = child.wait();
|
||
durable_record.status = "needs-reconciliation".to_string();
|
||
durable_record.sandbox_establishment = "unknown".to_string();
|
||
durable_record.target_exec = "unknown".to_string();
|
||
durable_record.launch_failure_kind = Some("launch-unknown".to_string());
|
||
durable_record.needs_reconciliation = true;
|
||
durable_record.terminal_at = Some(unix_timestamp());
|
||
durable_record.updated_at = unix_timestamp();
|
||
let _ = write_process_session_record(root, &durable_record);
|
||
return Err(ProjectCommandError::new(
|
||
ProjectCommandErrorStage::LaunchUnknown,
|
||
format!("command.start 已启动但 running 状态无法落盘，需要人工核对：{error}"),
|
||
));
|
||
}
|
||
process_session_registry()
|
||
.lock()
|
||
.map_err(|_| {
|
||
ProjectCommandError::new(
|
||
ProjectCommandErrorStage::LaunchUnknown,
|
||
"process session registry 锁已损坏",
|
||
)
|
||
})?
|
||
.sessions
|
||
.insert(process_id.clone(), Arc::clone(&live));
|
||
|
||
let reader_live = Arc::clone(&live);
|
||
thread::spawn(move || drain_process_session_output(reader_live, reader));
|
||
let supervisor_live = Arc::clone(&live);
|
||
let timeout_seconds = spec.timeout_seconds;
|
||
thread::spawn(move || {
|
||
supervise_process_session(
|
||
supervisor_live,
|
||
&mut child,
|
||
control_rx,
|
||
timeout_seconds,
|
||
process_group_leader,
|
||
)
|
||
});
|
||
|
||
poll_process_session_at(root, &live.identity, &process_id, None, Some(0), Some(0))
|
||
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Execution, error))
|
||
}
|
||
|
||
fn append_process_output_line(live: &LiveProcessSession, line: &[u8]) -> bool {
|
||
let text = String::from_utf8_lossy(line);
|
||
let mut sanitized = redact_agent_runtime_project_paths(
|
||
&live.root,
|
||
&sanitize_project_verification_output(&text),
|
||
PROCESS_SESSION_MAX_PENDING_LINE_BYTES,
|
||
);
|
||
if matches!(line.last(), Some(b'\n' | b'\r')) && !sanitized.ends_with('\n') {
|
||
sanitized.push('\n');
|
||
}
|
||
let mut output = match live.output.lock() {
|
||
Ok(output) => output,
|
||
Err(_) => return false,
|
||
};
|
||
if output.text.len().saturating_add(sanitized.len()) > PROCESS_SESSION_MAX_OUTPUT_BYTES {
|
||
output.output_limit_exceeded = true;
|
||
output.status = "output-limit-exceeded".to_string();
|
||
output.stdin_open = false;
|
||
live.output_changed.notify_all();
|
||
return false;
|
||
}
|
||
output.text.push_str(&sanitized);
|
||
live.output_changed.notify_all();
|
||
drop(output);
|
||
if persist_live_process_snapshot(live).is_err() {
|
||
if let Ok(mut output) = live.output.lock() {
|
||
output.status = "needs-reconciliation".to_string();
|
||
output.needs_reconciliation = true;
|
||
output.stdin_open = false;
|
||
live.output_changed.notify_all();
|
||
}
|
||
let _ = live.control.send(ProcessControl::Terminate);
|
||
}
|
||
true
|
||
}
|
||
|
||
fn persist_live_process_snapshot(live: &LiveProcessSession) -> Result<(), String> {
|
||
let output = live
|
||
.output
|
||
.lock()
|
||
.map_err(|_| "process session output 锁已损坏".to_string())?;
|
||
let output_sha256 = format!("{:x}", Sha256::digest(output.text.as_bytes()));
|
||
let transcript = ProcessSessionTranscript {
|
||
schema_version: PROCESS_SESSION_TRANSCRIPT_SCHEMA_VERSION.to_string(),
|
||
project_id: live.identity.project_id.clone(),
|
||
agent_id: live.identity.agent_id.clone(),
|
||
task_id: live.identity.task_id.clone(),
|
||
conversation_session_id: live.identity.conversation_session_id.clone(),
|
||
run_id: live.identity.run_id.clone(),
|
||
start_action_id: live.identity.start_action_id.clone(),
|
||
start_action_fingerprint: live.identity.start_action_fingerprint.clone(),
|
||
process_id: live.process_id.clone(),
|
||
output: output.text.clone(),
|
||
output_sha256,
|
||
output_bytes: output.text.len(),
|
||
updated_at: unix_timestamp(),
|
||
};
|
||
let record = process_session_record_from_live(live, &output);
|
||
write_agent_runtime_json_sidecar_with_max_bytes(
|
||
&live.root,
|
||
&process_session_transcript_relative_path(&live.process_id),
|
||
"Agent Runtime process transcript",
|
||
&transcript,
|
||
PROCESS_SESSION_TRANSCRIPT_MAX_BYTES,
|
||
)?;
|
||
write_process_session_record(&live.root, &record)
|
||
}
|
||
|
||
#[derive(Default)]
|
||
pub(super) struct AnsiStripper {
|
||
state: u8,
|
||
}
|
||
|
||
impl AnsiStripper {
|
||
pub(super) fn push(&mut self, byte: u8, visible: &mut Vec<u8>) {
|
||
match self.state {
|
||
0 if byte == 0x1b => self.state = 1,
|
||
0 if byte == b'\n' || byte == b'\r' || byte == b'\t' || byte >= 0x20 => {
|
||
visible.push(byte)
|
||
}
|
||
1 if byte == b'[' => self.state = 2,
|
||
1 if matches!(byte, b']' | b'P' | b'X' | b'^' | b'_') => self.state = 3,
|
||
1 => self.state = 0,
|
||
2 if (0x40..=0x7e).contains(&byte) => self.state = 0,
|
||
2 => {}
|
||
3 if byte == 0x07 => self.state = 0,
|
||
3 if byte == 0x1b => self.state = 4,
|
||
3 => {}
|
||
4 if byte == b'\\' => self.state = 0,
|
||
4 if byte == 0x1b => {}
|
||
4 => self.state = 3,
|
||
_ => self.state = 0,
|
||
}
|
||
}
|
||
}
|
||
|
||
#[cfg(windows)]
|
||
#[derive(Default)]
|
||
pub(super) struct AnsiTerminalRepositionDetector {
|
||
state: u8,
|
||
}
|
||
|
||
#[cfg(windows)]
|
||
impl AnsiTerminalRepositionDetector {
|
||
pub(super) fn push(&mut self, byte: u8) -> bool {
|
||
match self.state {
|
||
0 if byte == 0x1b => self.state = 1,
|
||
1 if byte == b'[' => self.state = 2,
|
||
1 => self.state = 0,
|
||
2 if (0x40..=0x7e).contains(&byte) => {
|
||
self.state = 0;
|
||
return matches!(byte, b'A'..=b'H' | b'f');
|
||
}
|
||
2 => {}
|
||
_ => self.state = 0,
|
||
}
|
||
false
|
||
}
|
||
}
|
||
|
||
fn drain_process_session_output(
|
||
live: Arc<LiveProcessSession>,
|
||
mut reader: Box<dyn std::io::Read + Send>,
|
||
) {
|
||
let mut buffer = [0u8; 4096];
|
||
let mut pending = Vec::new();
|
||
let mut pending_logical_line_bytes = 0usize;
|
||
let mut ansi = AnsiStripper::default();
|
||
let mut output_limit = false;
|
||
#[cfg(windows)]
|
||
let mut conpty_cursor_query_match = 0usize;
|
||
#[cfg(windows)]
|
||
let mut conpty_cursor_replied = false;
|
||
#[cfg(windows)]
|
||
let mut terminal_reposition = AnsiTerminalRepositionDetector::default();
|
||
#[cfg(windows)]
|
||
let mut conpty_soft_wrap = false;
|
||
loop {
|
||
match reader.read(&mut buffer) {
|
||
Ok(0) => break,
|
||
Ok(read) => {
|
||
for byte in &buffer[..read] {
|
||
#[cfg(windows)]
|
||
{
|
||
const CONPTY_CURSOR_QUERY: &[u8] = b"\x1b[6n";
|
||
if !conpty_cursor_replied
|
||
&& *byte == CONPTY_CURSOR_QUERY[conpty_cursor_query_match]
|
||
{
|
||
conpty_cursor_query_match += 1;
|
||
if conpty_cursor_query_match == CONPTY_CURSOR_QUERY.len() {
|
||
conpty_cursor_query_match = 0;
|
||
let reply_result = live
|
||
.writer
|
||
.lock()
|
||
.map_err(|_| "process session stdin 锁已损坏".to_string())
|
||
.and_then(|mut writer| {
|
||
let Some(writer) = writer.as_mut() else {
|
||
// 终止线程会先关闭 stdin；此时 ConPTY 可能仍把启动期
|
||
// 光标查询交给 reader。进程树已经进入收束阶段，无需再
|
||
// 把无法回复查询升级成 needs-reconciliation。
|
||
return Ok(());
|
||
};
|
||
writer
|
||
.write_all(b"\x1b[1;1R")
|
||
.and_then(|()| writer.flush())
|
||
.map_err(|error| {
|
||
format!("回复 Windows ConPTY 光标查询失败：{error}")
|
||
})
|
||
});
|
||
if let Err(error) = reply_result {
|
||
if let Ok(mut output) = live.output.lock() {
|
||
output.status = "failed".to_string();
|
||
output.needs_reconciliation = true;
|
||
output.stdin_open = false;
|
||
let detail = format!(
|
||
"\n<process output handshake failed: {error}>\n"
|
||
);
|
||
if output.text.len().saturating_add(detail.len())
|
||
<= PROCESS_SESSION_MAX_OUTPUT_BYTES
|
||
{
|
||
output.text.push_str(&detail);
|
||
}
|
||
live.output_changed.notify_all();
|
||
}
|
||
let _ = live.control.send(ProcessControl::Terminate);
|
||
return;
|
||
}
|
||
conpty_cursor_replied = true;
|
||
}
|
||
} else if !conpty_cursor_replied {
|
||
conpty_cursor_query_match =
|
||
usize::from(*byte == CONPTY_CURSOR_QUERY[0]);
|
||
}
|
||
}
|
||
#[cfg(windows)]
|
||
let ends_terminal_reposition = terminal_reposition.push(*byte);
|
||
let before = pending.len();
|
||
ansi.push(*byte, &mut pending);
|
||
let visible_bytes = pending.len().saturating_sub(before);
|
||
if visible_bytes > 0 && !matches!(pending.last(), Some(b'\n' | b'\r')) {
|
||
pending_logical_line_bytes =
|
||
pending_logical_line_bytes.saturating_add(visible_bytes);
|
||
if pending_logical_line_bytes > PROCESS_SESSION_MAX_PENDING_LINE_BYTES {
|
||
output_limit = true;
|
||
break;
|
||
}
|
||
}
|
||
#[cfg(windows)]
|
||
if ends_terminal_reposition && !pending.is_empty() {
|
||
pending.push(b'\n');
|
||
if !append_process_output_line(&live, &pending) {
|
||
output_limit = true;
|
||
break;
|
||
}
|
||
pending.clear();
|
||
continue;
|
||
}
|
||
if pending.len() == before {
|
||
continue;
|
||
}
|
||
if matches!(pending.last(), Some(b'\n' | b'\r')) {
|
||
if !append_process_output_line(&live, &pending) {
|
||
output_limit = true;
|
||
break;
|
||
}
|
||
#[cfg(windows)]
|
||
{
|
||
// ConPTY materializes an automatic terminal-width wrap as CR/LF.
|
||
// It is a display boundary, not an application line terminator, so
|
||
// it must not reset the logical-line safety limit. A real short line
|
||
// still resets at CR; the immediately following LF preserves that
|
||
// decision.
|
||
const PROCESS_SESSION_PTY_COLS: usize = 120;
|
||
match pending.last() {
|
||
Some(b'\r') => {
|
||
conpty_soft_wrap =
|
||
pending_logical_line_bytes >= PROCESS_SESSION_PTY_COLS;
|
||
if !conpty_soft_wrap {
|
||
pending_logical_line_bytes = 0;
|
||
}
|
||
}
|
||
Some(b'\n') => {
|
||
if !conpty_soft_wrap {
|
||
pending_logical_line_bytes = 0;
|
||
}
|
||
conpty_soft_wrap = false;
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
#[cfg(not(windows))]
|
||
{
|
||
pending_logical_line_bytes = 0;
|
||
}
|
||
pending.clear();
|
||
}
|
||
}
|
||
if output_limit {
|
||
if let Ok(mut output) = live.output.lock() {
|
||
output.output_limit_exceeded = true;
|
||
output.status = "output-limit-exceeded".to_string();
|
||
output.stdin_open = false;
|
||
live.output_changed.notify_all();
|
||
}
|
||
let _ = live.control.send(ProcessControl::OutputLimit);
|
||
break;
|
||
}
|
||
}
|
||
Err(error) => {
|
||
if let Ok(mut output) = live.output.lock() {
|
||
output.status = "failed".to_string();
|
||
output.needs_reconciliation = true;
|
||
output.stdin_open = false;
|
||
let detail = format!("\n<process output read failed: {error}>\n");
|
||
if output.text.len().saturating_add(detail.len())
|
||
<= PROCESS_SESSION_MAX_OUTPUT_BYTES
|
||
{
|
||
output.text.push_str(&detail);
|
||
}
|
||
live.output_changed.notify_all();
|
||
}
|
||
let _ = live.control.send(ProcessControl::Terminate);
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
if !pending.is_empty() && !output_limit {
|
||
let _ = append_process_output_line(&live, &pending);
|
||
}
|
||
if let Ok(mut output) = live.output.lock() {
|
||
output.reader_finished = true;
|
||
live.output_changed.notify_all();
|
||
}
|
||
}
|
||
|
||
fn supervise_process_session(
|
||
live: Arc<LiveProcessSession>,
|
||
child: &mut Box<dyn Child + Send + Sync>,
|
||
control_rx: std::sync::mpsc::Receiver<ProcessControl>,
|
||
timeout_seconds: u64,
|
||
#[cfg_attr(not(unix), allow(unused_variables))] process_group_leader: Option<i32>,
|
||
#[cfg(target_os = "linux")] mut launch_bridge: ProcessSessionBridge,
|
||
) {
|
||
let deadline = std::time::Instant::now() + Duration::from_secs(timeout_seconds);
|
||
let (terminal_status, exit_code, signal) = loop {
|
||
match child.try_wait() {
|
||
Ok(Some(status)) => {
|
||
#[cfg(target_os = "linux")]
|
||
let _ = &status;
|
||
#[cfg(target_os = "linux")]
|
||
let terminal = match launch_bridge.wait_terminal(Duration::from_secs(2)) {
|
||
Ok(ProcessSessionTerminalVerdict::Exited { code }) => {
|
||
("exited".to_string(), Some(code), None)
|
||
}
|
||
Ok(ProcessSessionTerminalVerdict::Signaled { signal }) => {
|
||
("exited".to_string(), None, Some(format!("signal-{signal}")))
|
||
}
|
||
Ok(ProcessSessionTerminalVerdict::Unknown) => {
|
||
let error = "process session target terminal 无法确认".to_string();
|
||
mark_process_session_reconciliation(&live, &error);
|
||
("needs-reconciliation".to_string(), None, Some(error))
|
||
}
|
||
Err(error) => {
|
||
mark_process_session_reconciliation(&live, &error);
|
||
("needs-reconciliation".to_string(), None, Some(error))
|
||
}
|
||
};
|
||
if let Err(error) = terminate_process_session_child(
|
||
&live,
|
||
child,
|
||
process_group_leader,
|
||
#[cfg(target_os = "linux")]
|
||
&mut launch_bridge,
|
||
true,
|
||
) {
|
||
mark_process_session_reconciliation(&live, &error);
|
||
break ("needs-reconciliation".to_string(), None, Some(error));
|
||
}
|
||
#[cfg(target_os = "linux")]
|
||
break terminal;
|
||
#[cfg(not(target_os = "linux"))]
|
||
break (
|
||
"exited".to_string(),
|
||
i32::try_from(status.exit_code()).ok(),
|
||
status.signal().map(str::to_string),
|
||
);
|
||
}
|
||
Ok(None) => {}
|
||
Err(error) => {
|
||
let wait_error = format!("wait failed: {error}");
|
||
if let Err(termination_error) = terminate_process_session_child(
|
||
&live,
|
||
child,
|
||
process_group_leader,
|
||
#[cfg(target_os = "linux")]
|
||
&mut launch_bridge,
|
||
true,
|
||
) {
|
||
let detail = format!("{wait_error}; {termination_error}");
|
||
mark_process_session_reconciliation(&live, &detail);
|
||
break ("needs-reconciliation".to_string(), None, Some(detail));
|
||
}
|
||
break ("failed".to_string(), None, Some(wait_error));
|
||
}
|
||
}
|
||
|
||
let remaining = deadline.saturating_duration_since(std::time::Instant::now());
|
||
let wait = remaining.min(Duration::from_millis(50));
|
||
match control_rx.recv_timeout(wait) {
|
||
Ok(ProcessControl::Terminate) => {
|
||
match terminate_process_session_child(
|
||
&live,
|
||
child,
|
||
process_group_leader,
|
||
#[cfg(target_os = "linux")]
|
||
&mut launch_bridge,
|
||
false,
|
||
) {
|
||
Ok(()) => break ("terminated".to_string(), None, None),
|
||
Err(error) => {
|
||
mark_process_session_reconciliation(&live, &error);
|
||
break ("needs-reconciliation".to_string(), None, Some(error));
|
||
}
|
||
}
|
||
}
|
||
Ok(ProcessControl::OutputLimit) => {
|
||
match terminate_process_session_child(
|
||
&live,
|
||
child,
|
||
process_group_leader,
|
||
#[cfg(target_os = "linux")]
|
||
&mut launch_bridge,
|
||
true,
|
||
) {
|
||
Ok(()) => break ("output-limit-exceeded".to_string(), None, None),
|
||
Err(error) => {
|
||
mark_process_session_reconciliation(&live, &error);
|
||
break ("needs-reconciliation".to_string(), None, Some(error));
|
||
}
|
||
}
|
||
}
|
||
Ok(ProcessControl::Shutdown) => {
|
||
match terminate_process_session_child(
|
||
&live,
|
||
child,
|
||
process_group_leader,
|
||
#[cfg(target_os = "linux")]
|
||
&mut launch_bridge,
|
||
true,
|
||
) {
|
||
Ok(()) => {
|
||
break (
|
||
"terminated".to_string(),
|
||
None,
|
||
Some("runner-shutdown".to_string()),
|
||
);
|
||
}
|
||
Err(error) => {
|
||
mark_process_session_reconciliation(&live, &error);
|
||
break ("needs-reconciliation".to_string(), None, Some(error));
|
||
}
|
||
}
|
||
}
|
||
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
|
||
match terminate_process_session_child(
|
||
&live,
|
||
child,
|
||
process_group_leader,
|
||
#[cfg(target_os = "linux")]
|
||
&mut launch_bridge,
|
||
true,
|
||
) {
|
||
Ok(()) => {
|
||
break (
|
||
"terminated".to_string(),
|
||
None,
|
||
Some("control-disconnected".to_string()),
|
||
);
|
||
}
|
||
Err(error) => {
|
||
mark_process_session_reconciliation(&live, &error);
|
||
break ("needs-reconciliation".to_string(), None, Some(error));
|
||
}
|
||
}
|
||
}
|
||
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
|
||
}
|
||
if std::time::Instant::now() >= deadline {
|
||
match terminate_process_session_child(
|
||
&live,
|
||
child,
|
||
process_group_leader,
|
||
#[cfg(target_os = "linux")]
|
||
&mut launch_bridge,
|
||
true,
|
||
) {
|
||
Ok(()) => break ("timed-out".to_string(), None, None),
|
||
Err(error) => {
|
||
mark_process_session_reconciliation(&live, &error);
|
||
break ("needs-reconciliation".to_string(), None, Some(error));
|
||
}
|
||
}
|
||
}
|
||
};
|
||
finalize_live_process_session(&live, &terminal_status, exit_code, signal);
|
||
}
|
||
|
||
pub(super) fn mark_process_session_reconciliation(live: &LiveProcessSession, error: &str) {
|
||
if let Ok(mut output) = live.output.lock() {
|
||
output.status = "needs-reconciliation".to_string();
|
||
output.needs_reconciliation = true;
|
||
output.stdin_open = false;
|
||
output.signal = Some(redact_agent_runtime_project_paths(&live.root, error, 300));
|
||
live.output_changed.notify_all();
|
||
}
|
||
}
|
||
|
||
fn terminate_process_session_child(
|
||
live: &LiveProcessSession,
|
||
child: &mut Box<dyn Child + Send + Sync>,
|
||
#[cfg_attr(not(unix), allow(unused_variables))] process_group_leader: Option<i32>,
|
||
#[cfg(target_os = "linux")] launch_bridge: &mut ProcessSessionBridge,
|
||
#[cfg_attr(windows, allow(unused_variables))] force: bool,
|
||
) -> Result<(), String> {
|
||
#[cfg(not(windows))]
|
||
let _ = live;
|
||
let mut child_reaped = child.try_wait().ok().flatten().is_some();
|
||
let tree_contained;
|
||
#[cfg(windows)]
|
||
{
|
||
tree_contained = live
|
||
.job
|
||
.lock()
|
||
.map_err(|_| "Windows process session Job Object 锁已损坏".to_string())?
|
||
.as_ref()
|
||
.ok_or_else(|| "Windows process session 缺少 Job Object".to_string())?
|
||
.terminate()
|
||
.is_ok();
|
||
}
|
||
#[cfg(unix)]
|
||
{
|
||
tree_contained = if let Some(group) = process_group_leader.filter(|value| *value > 0) {
|
||
#[cfg(target_os = "linux")]
|
||
if !force && !child_reaped {
|
||
if let Err(error) = launch_bridge.terminate_target() {
|
||
match child.try_wait() {
|
||
Ok(Some(_)) => child_reaped = true,
|
||
Ok(None) => return Err(error),
|
||
Err(wait_error) => {
|
||
return Err(format!("{error}；检查 wrapper 终态失败：{wait_error}"));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
#[cfg(all(unix, not(target_os = "linux")))]
|
||
if !force && !child_reaped {
|
||
unsafe {
|
||
libc::kill(-group, libc::SIGTERM);
|
||
}
|
||
}
|
||
if !force && !child_reaped {
|
||
let deadline = std::time::Instant::now()
|
||
+ Duration::from_millis(PROCESS_SESSION_TERMINATE_GRACE_MS);
|
||
while std::time::Instant::now() < deadline {
|
||
if !child_reaped {
|
||
if let Ok(Some(_)) = child.try_wait() {
|
||
child_reaped = true;
|
||
break;
|
||
}
|
||
}
|
||
thread::sleep(Duration::from_millis(25));
|
||
}
|
||
}
|
||
let killed = unsafe { libc::kill(-group, libc::SIGKILL) };
|
||
if killed == 0 {
|
||
true
|
||
} else {
|
||
std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) && child_reaped
|
||
}
|
||
} else {
|
||
false
|
||
};
|
||
}
|
||
#[cfg(not(any(unix, windows)))]
|
||
{
|
||
tree_contained = false;
|
||
}
|
||
if !child_reaped {
|
||
let _ = child.kill();
|
||
match child.wait() {
|
||
Ok(_) => child_reaped = true,
|
||
Err(error) => {
|
||
return Err(format!("process session child 回收失败：{error}"));
|
||
}
|
||
}
|
||
}
|
||
if !tree_contained {
|
||
return Err("process session 进程树终止结果无法确认".to_string());
|
||
}
|
||
if child_reaped {
|
||
Ok(())
|
||
} else {
|
||
Err("process session child 尚未回收".to_string())
|
||
}
|
||
}
|
||
|
||
fn finalize_live_process_session(
|
||
live: &Arc<LiveProcessSession>,
|
||
status: &str,
|
||
exit_code: Option<i32>,
|
||
signal: Option<String>,
|
||
) {
|
||
if let Ok(mut writer) = live.writer.lock() {
|
||
writer.take();
|
||
}
|
||
if let Ok(mut master) = live.master.lock() {
|
||
master.take();
|
||
}
|
||
#[cfg(windows)]
|
||
if let Ok(mut job) = live.job.lock() {
|
||
job.take();
|
||
}
|
||
let source_fingerprint_after = project_command_source_fingerprint(&live.root).ok();
|
||
let mut output = match live.output.lock() {
|
||
Ok(output) => output,
|
||
Err(_) => return,
|
||
};
|
||
let deadline = std::time::Instant::now() + Duration::from_secs(2);
|
||
while !output.reader_finished && std::time::Instant::now() < deadline {
|
||
let wait = live
|
||
.output_changed
|
||
.wait_timeout(output, Duration::from_millis(25));
|
||
let Ok((next, _)) = wait else {
|
||
return;
|
||
};
|
||
output = next;
|
||
}
|
||
output.status = if output.needs_reconciliation {
|
||
"needs-reconciliation".to_string()
|
||
} else if output.output_limit_exceeded {
|
||
"output-limit-exceeded".to_string()
|
||
} else if output.status == "failed" {
|
||
"failed".to_string()
|
||
} else {
|
||
status.to_string()
|
||
};
|
||
output.exit_code = exit_code;
|
||
output.signal = signal;
|
||
output.stdin_open = false;
|
||
output.source_changed = source_fingerprint_after
|
||
.as_ref()
|
||
.map(|after| after != &live.source_fingerprint_before);
|
||
output.source_fingerprint_after = source_fingerprint_after;
|
||
if output.source_fingerprint_after.is_none() || !output.reader_finished {
|
||
output.needs_reconciliation = true;
|
||
}
|
||
|
||
let output_sha256 = format!("{:x}", Sha256::digest(output.text.as_bytes()));
|
||
let transcript = ProcessSessionTranscript {
|
||
schema_version: PROCESS_SESSION_TRANSCRIPT_SCHEMA_VERSION.to_string(),
|
||
project_id: live.identity.project_id.clone(),
|
||
agent_id: live.identity.agent_id.clone(),
|
||
task_id: live.identity.task_id.clone(),
|
||
conversation_session_id: live.identity.conversation_session_id.clone(),
|
||
run_id: live.identity.run_id.clone(),
|
||
start_action_id: live.identity.start_action_id.clone(),
|
||
start_action_fingerprint: live.identity.start_action_fingerprint.clone(),
|
||
process_id: live.process_id.clone(),
|
||
output: output.text.clone(),
|
||
output_sha256,
|
||
output_bytes: output.text.len(),
|
||
updated_at: unix_timestamp(),
|
||
};
|
||
let transcript_result = write_agent_runtime_json_sidecar_with_max_bytes(
|
||
&live.root,
|
||
&process_session_transcript_relative_path(&live.process_id),
|
||
"Agent Runtime process transcript",
|
||
&transcript,
|
||
PROCESS_SESSION_TRANSCRIPT_MAX_BYTES,
|
||
);
|
||
if transcript_result.is_err() {
|
||
output.needs_reconciliation = true;
|
||
}
|
||
let record = process_session_record_from_live(live, &output);
|
||
let record_persisted = write_process_session_record(&live.root, &record).is_ok();
|
||
if !record_persisted {
|
||
output.needs_reconciliation = true;
|
||
}
|
||
let needs_reconciliation = output.needs_reconciliation;
|
||
// 只有 live registry 已收束后才能发布可信终态。等待 `output_changed` 的调用方
|
||
// 可能立即执行 run 级取消清理，而 registry 中的任何 live 条目都会被视为未完成会话。
|
||
if record_persisted && !needs_reconciliation {
|
||
if let Ok(mut registry) = process_session_registry().lock() {
|
||
registry.sessions.remove(&live.process_id);
|
||
}
|
||
}
|
||
live.output_changed.notify_all();
|
||
drop(output);
|
||
}
|