Files
Genarrative/apps/ai-game-creator-shell/src-tauri/src/process_session.rs
T
AIGameCreator App 689082901f 为持久进程增加可信启动握手
新增 PTY 外私有 bridge 与 sandbox ready/commit/exec 协议
升级 process record v3 并收紧恢复、幂等与 final/idle 门禁
完善 target 前台进程组、graceful 后代收束和审计失败处理
补齐跨平台回归测试及 Runtime 文档
2026-07-14 09:41:18 +08:00

4837 lines
184 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::*;
use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::sync::Condvar;
#[cfg(target_os = "linux")]
use crate::process_session_bridge::*;
const PROCESS_SESSION_SCHEMA_VERSION: &str = "3";
const PROCESS_SESSION_TRANSCRIPT_SCHEMA_VERSION: &str = "2";
const PROCESS_SESSION_CURSOR_VERSION: &str = "v1";
const PROCESS_SESSION_MAX_PER_PROJECT: usize = 4;
const PROCESS_SESSION_MAX_PER_AGENT: usize = 2;
const PROCESS_SESSION_MAX_OUTPUT_BYTES: usize = 256 * 1024;
const PROCESS_SESSION_MAX_PENDING_LINE_BYTES: usize = 16 * 1024;
const PROCESS_SESSION_MAX_STDIN_BYTES: usize = 8 * 1024;
const PROCESS_SESSION_DEFAULT_POLL_CHARS: usize = 8_000;
const PROCESS_SESSION_MAX_POLL_CHARS: usize = 16_000;
const PROCESS_SESSION_MAX_POLL_WAIT_MS: u64 = 30_000;
const PROCESS_SESSION_RECORD_MAX_BYTES: usize = 32 * 1024;
const PROCESS_SESSION_TRANSCRIPT_MAX_BYTES: usize = 320 * 1024;
const PROCESS_SESSION_TERMINATE_GRACE_MS: u64 = 800;
#[cfg(target_os = "linux")]
const PROCESS_SESSION_OWNER_PID_ENV: &str = "GENARRATIVE_PROCESS_SESSION_OWNER_PID";
#[cfg(target_os = "linux")]
const PROCESS_SESSION_CHILD_MODE: &str = "--process-session-child";
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct ProcessSessionIdentity {
pub(crate) project_id: String,
pub(crate) agent_id: String,
pub(crate) task_id: String,
pub(crate) conversation_session_id: String,
pub(crate) run_id: String,
pub(crate) start_action_id: String,
pub(crate) start_action_fingerprint: String,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub(crate) struct ProcessSessionRecord {
pub(crate) schema_version: String,
pub(crate) project_id: String,
pub(crate) agent_id: String,
pub(crate) task_id: String,
pub(crate) conversation_session_id: String,
pub(crate) run_id: String,
pub(crate) start_action_id: String,
pub(crate) start_action_fingerprint: String,
pub(crate) process_id: String,
pub(crate) owner_boot_id: String,
pub(crate) command_id: String,
pub(crate) program: String,
pub(crate) cwd: String,
#[serde(default)]
pub(crate) sandbox_backend: String,
#[serde(default)]
pub(crate) sandbox_mode: String,
#[serde(default)]
pub(crate) network_access: String,
#[serde(default)]
pub(crate) sandbox_profile_version: String,
#[serde(default)]
pub(crate) sandbox_establishment: String,
#[serde(default)]
pub(crate) target_exec: String,
#[serde(default)]
pub(crate) launch_failure_kind: Option<String>,
#[serde(default)]
pub(crate) sandbox_ready_at: Option<u64>,
#[serde(default)]
pub(crate) exec_established_at: Option<u64>,
pub(crate) status: String,
pub(crate) exit_code: Option<i32>,
pub(crate) signal: Option<String>,
pub(crate) stdin_open: bool,
pub(crate) output_bytes: usize,
pub(crate) output_sha256: String,
pub(crate) output_ref: Option<String>,
pub(crate) source_fingerprint_before: String,
pub(crate) source_fingerprint_after: Option<String>,
pub(crate) source_changed: Option<bool>,
pub(crate) needs_reconciliation: bool,
pub(crate) started_at: u64,
pub(crate) terminal_at: Option<u64>,
pub(crate) updated_at: u64,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct ProcessSessionTranscript {
schema_version: String,
project_id: String,
agent_id: String,
task_id: String,
conversation_session_id: String,
run_id: String,
start_action_id: String,
start_action_fingerprint: String,
process_id: String,
output: String,
output_sha256: String,
output_bytes: usize,
updated_at: u64,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct ProcessSessionPollResult {
pub(crate) process_id: String,
pub(crate) status: String,
pub(crate) output: String,
pub(crate) cursor: String,
pub(crate) next_cursor: String,
pub(crate) has_more: bool,
pub(crate) stdin_open: bool,
pub(crate) exit_code: Option<i32>,
pub(crate) signal: Option<String>,
pub(crate) output_bytes: usize,
pub(crate) output_sha256: String,
pub(crate) source_changed: Option<bool>,
pub(crate) needs_reconciliation: bool,
pub(crate) sandbox_backend: String,
pub(crate) sandbox_mode: String,
pub(crate) network_access: String,
pub(crate) sandbox_profile_version: String,
pub(crate) sandbox_establishment: String,
pub(crate) target_exec: String,
pub(crate) launch_failure_kind: Option<String>,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct ProcessSessionStdinResult {
pub(crate) process_id: String,
pub(crate) bytes_written: usize,
pub(crate) content_sha256: String,
pub(crate) stdin_open: bool,
pub(crate) eof: bool,
pub(crate) sandbox_backend: String,
pub(crate) sandbox_mode: String,
pub(crate) network_access: String,
pub(crate) sandbox_profile_version: String,
}
#[derive(Debug)]
struct ProcessOutputState {
text: String,
status: String,
exit_code: Option<i32>,
signal: Option<String>,
stdin_open: bool,
reader_finished: bool,
output_limit_exceeded: bool,
source_fingerprint_after: Option<String>,
source_changed: Option<bool>,
needs_reconciliation: bool,
launch_failure_kind: Option<String>,
}
impl ProcessOutputState {
fn running() -> Self {
Self {
text: String::new(),
status: "running".to_string(),
exit_code: None,
signal: None,
stdin_open: true,
reader_finished: false,
output_limit_exceeded: false,
source_fingerprint_after: None,
source_changed: None,
needs_reconciliation: false,
launch_failure_kind: None,
}
}
}
#[derive(Debug)]
enum ProcessControl {
Terminate,
OutputLimit,
Shutdown,
}
struct LiveProcessSession {
root: PathBuf,
identity: ProcessSessionIdentity,
process_id: String,
command_id: String,
program: String,
cwd: String,
sandbox_backend: String,
sandbox_mode: String,
network_access: String,
sandbox_profile_version: String,
sandbox_establishment: String,
target_exec: String,
sandbox_ready_at: Option<u64>,
exec_established_at: Option<u64>,
source_fingerprint_before: String,
started_at: u64,
output: Mutex<ProcessOutputState>,
output_changed: Condvar,
writer: Mutex<Option<Box<dyn std::io::Write + Send>>>,
master: Mutex<Option<Box<dyn MasterPty + Send>>>,
#[cfg(windows)]
job: Mutex<Option<WindowsProcessJob>>,
control: std::sync::mpsc::Sender<ProcessControl>,
}
#[cfg(windows)]
struct WindowsProcessJob(windows_sys::Win32::Foundation::HANDLE);
#[cfg(windows)]
unsafe impl Send for WindowsProcessJob {}
#[cfg(windows)]
unsafe impl Sync for WindowsProcessJob {}
#[cfg(windows)]
impl WindowsProcessJob {
fn assign(child: &dyn Child) -> Result<Self, String> {
use std::mem::size_of;
use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE};
use windows_sys::Win32::System::JobObjects::{
AssignProcessToJobObject, CreateJobObjectW, JobObjectExtendedLimitInformation,
SetInformationJobObject, JOBOBJECT_EXTENDED_LIMIT_INFORMATION,
JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
};
let process = child
.as_raw_handle()
.ok_or_else(|| "command.start Windows child 缺少 process handle".to_string())?
as windows_sys::Win32::Foundation::HANDLE;
let handle = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) };
if handle.is_null() || handle == INVALID_HANDLE_VALUE {
return Err(format!(
"创建 command.start Windows Job Object 失败:{}",
std::io::Error::last_os_error()
));
}
let mut information = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default();
information.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
let configured = unsafe {
SetInformationJobObject(
handle,
JobObjectExtendedLimitInformation,
&information as *const _ as *const _,
size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
)
};
let assigned = configured != 0 && unsafe { AssignProcessToJobObject(handle, process) } != 0;
if !assigned {
let error = std::io::Error::last_os_error();
unsafe {
CloseHandle(handle);
}
return Err(format!(
"配置 command.start Windows Job Object 失败:{error}"
));
}
Ok(Self(handle))
}
fn terminate(&self) -> Result<(), String> {
use windows_sys::Win32::System::JobObjects::TerminateJobObject;
if unsafe { TerminateJobObject(self.0, 1) } == 0 {
return Err(format!(
"终止 command.start Windows Job Object 失败:{}",
std::io::Error::last_os_error()
));
}
Ok(())
}
}
#[cfg(windows)]
impl Drop for WindowsProcessJob {
fn drop(&mut self) {
unsafe {
windows_sys::Win32::Foundation::CloseHandle(self.0);
}
}
}
impl std::fmt::Debug for LiveProcessSession {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("LiveProcessSession")
.field("process_id", &self.process_id)
.field("agent_id", &self.identity.agent_id)
.field("run_id", &self.identity.run_id)
.finish_non_exhaustive()
}
}
#[derive(Default)]
struct ProcessSessionRegistry {
sessions: HashMap<String, Arc<LiveProcessSession>>,
}
#[cfg(target_os = "linux")]
#[derive(Clone, Debug)]
struct PendingProcessLaunch {
root: PathBuf,
agent_id: String,
process_group_leader: Option<i32>,
shutdown_requested: bool,
}
#[cfg(target_os = "linux")]
#[derive(Default)]
struct PendingProcessLaunchRegistry {
launches: HashMap<String, PendingProcessLaunch>,
}
#[cfg(target_os = "linux")]
struct PendingProcessLaunchGuard {
process_id: String,
}
#[cfg(target_os = "linux")]
impl Drop for PendingProcessLaunchGuard {
fn drop(&mut self) {
if let Ok(mut registry) = pending_process_launch_registry().lock() {
registry.launches.remove(&self.process_id);
}
}
}
static PROCESS_SESSION_REGISTRY: OnceLock<Mutex<ProcessSessionRegistry>> = OnceLock::new();
static PROCESS_SESSION_BOOT_ID: OnceLock<String> = OnceLock::new();
#[cfg(target_os = "linux")]
static PENDING_PROCESS_LAUNCH_REGISTRY: OnceLock<Mutex<PendingProcessLaunchRegistry>> =
OnceLock::new();
fn process_session_registry() -> &'static Mutex<ProcessSessionRegistry> {
PROCESS_SESSION_REGISTRY.get_or_init(|| Mutex::new(ProcessSessionRegistry::default()))
}
#[cfg(target_os = "linux")]
fn pending_process_launch_registry() -> &'static Mutex<PendingProcessLaunchRegistry> {
PENDING_PROCESS_LAUNCH_REGISTRY
.get_or_init(|| Mutex::new(PendingProcessLaunchRegistry::default()))
}
#[cfg(target_os = "linux")]
fn reserve_pending_process_launch(
root: &Path,
agent_id: &str,
process_id: &str,
) -> Result<PendingProcessLaunchGuard, String> {
let mut registry = pending_process_launch_registry()
.lock()
.map_err(|_| "pending process launch registry 锁已损坏".to_string())?;
if registry.launches.contains_key(process_id) {
return Err("command.start pending launch 身份冲突".to_string());
}
registry.launches.insert(
process_id.to_string(),
PendingProcessLaunch {
root: root.to_path_buf(),
agent_id: agent_id.to_string(),
process_group_leader: None,
shutdown_requested: false,
},
);
Ok(PendingProcessLaunchGuard {
process_id: process_id.to_string(),
})
}
#[cfg(target_os = "linux")]
fn activate_pending_process_launch(
process_id: &str,
process_group_leader: i32,
) -> Result<(), String> {
if process_group_leader <= 1 {
return Err("command.start wrapper 进程组身份无效".to_string());
}
let mut registry = pending_process_launch_registry()
.lock()
.map_err(|_| "pending process launch registry 锁已损坏".to_string())?;
let launch = registry
.launches
.get_mut(process_id)
.ok_or_else(|| "command.start pending launch reservation 缺失".to_string())?;
launch.process_group_leader = Some(process_group_leader);
if launch.shutdown_requested {
unsafe {
libc::kill(-process_group_leader, libc::SIGKILL);
}
return Err("Runner shutdown 已取消 pending process launch".to_string());
}
Ok(())
}
pub(crate) fn process_session_boot_id() -> &'static str {
PROCESS_SESSION_BOOT_ID
.get_or_init(|| {
let mut digest = Sha256::new();
digest.update(std::process::id().to_le_bytes());
digest.update(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos()
.to_le_bytes(),
);
digest.update(unix_timestamp().to_le_bytes());
let value = format!("{:x}", digest.finalize());
format!("boot-{}", &value[..32])
})
.as_str()
}
pub(crate) fn initialize_process_session_boot_id(boot_id: &str) -> Result<(), String> {
let boot_id = boot_id.trim();
if boot_id.is_empty()
|| boot_id.chars().count() > 160
|| boot_id.chars().any(|character| character.is_control())
{
return Err("Agent Runner bootId 无效,无法初始化 process session owner".to_string());
}
match PROCESS_SESSION_BOOT_ID.set(boot_id.to_string()) {
Ok(()) => Ok(()),
Err(_) if PROCESS_SESSION_BOOT_ID.get().map(String::as_str) == Some(boot_id) => Ok(()),
Err(_) => Err("process session owner bootId 已被其他 Runner 初始化".to_string()),
}
}
#[cfg(target_os = "linux")]
pub(crate) fn is_process_session_child_mode(args: &[String]) -> bool {
args.first().map(String::as_str) == Some(PROCESS_SESSION_CHILD_MODE)
}
#[cfg(target_os = "linux")]
pub(crate) fn run_process_session_child(args: &[String]) -> Result<i32, String> {
if args != [PROCESS_SESSION_CHILD_MODE] {
return Err("process session child 参数无效".to_string());
}
let expected_parent = std::env::var(PROCESS_SESSION_OWNER_PID_ENV)
.map_err(|_| "process session child 缺少 owner pid".to_string())?
.parse::<libc::pid_t>()
.map_err(|_| "process session child owner pid 无效".to_string())?;
if expected_parent <= 1 {
return Err("process session child owner pid 无效".to_string());
}
if unsafe { libc::getppid() } != expected_parent {
return Err("process session owner 在 child containment 生效前已退出".to_string());
}
unsafe {
libc::signal(libc::SIGHUP, libc::SIG_IGN);
}
std::env::remove_var(PROCESS_SESSION_OWNER_PID_ENV);
run_process_session_bridge_child(expected_parent)
}
fn validate_process_session_identity(identity: &ProcessSessionIdentity) -> Result<(), String> {
for (label, value, max_chars) in [
("projectId", identity.project_id.as_str(), 160),
("agentId", identity.agent_id.as_str(), 96),
("taskId", identity.task_id.as_str(), 96),
(
"conversationSessionId",
identity.conversation_session_id.as_str(),
160,
),
("runId", identity.run_id.as_str(), 160),
("startActionId", identity.start_action_id.as_str(), 160),
] {
let trimmed = value.trim();
if trimmed.is_empty()
|| trimmed.chars().count() > max_chars
|| trimmed.chars().any(|character| character.is_control())
{
return Err(format!("command.start {label} 无效"));
}
}
if identity.start_action_fingerprint.len() != 64
|| !identity
.start_action_fingerprint
.bytes()
.all(|byte| byte.is_ascii_hexdigit())
{
return Err("command.start action fingerprint 无效".to_string());
}
Ok(())
}
fn validate_process_id(process_id: &str) -> Result<(), String> {
if process_id.len() != 37
|| !process_id.starts_with("proc-")
|| !process_id[5..].bytes().all(|byte| byte.is_ascii_hexdigit())
{
return Err("processId 格式无效".to_string());
}
Ok(())
}
fn process_session_id(identity: &ProcessSessionIdentity) -> String {
let payload = serde_json::to_vec(&serde_json::json!({
"projectId": identity.project_id,
"agentId": identity.agent_id,
"taskId": identity.task_id,
"conversationSessionId": identity.conversation_session_id,
"runId": identity.run_id,
"startActionId": identity.start_action_id,
"startActionFingerprint": identity.start_action_fingerprint,
"ownerBootId": process_session_boot_id(),
}))
.unwrap_or_default();
let value = format!("{:x}", Sha256::digest(payload));
format!("proc-{}", &value[..32])
}
fn process_session_record_relative_path(process_id: &str) -> String {
format!(".agent/runtime/process-sessions/{process_id}.json")
}
fn process_session_transcript_relative_path(process_id: &str) -> String {
format!(".agent/runtime/process-sessions/{process_id}.output.json")
}
fn process_session_cursor(process_id: &str, offset: usize) -> String {
format!("{PROCESS_SESSION_CURSOR_VERSION}:{process_id}:{offset}")
}
fn parse_process_session_cursor(
process_id: &str,
cursor: Option<&str>,
output: &str,
) -> Result<usize, String> {
let Some(cursor) = cursor.filter(|value| !value.trim().is_empty()) else {
return Ok(0);
};
let mut parts = cursor.split(':');
let version = parts.next().unwrap_or_default();
let cursor_process_id = parts.next().unwrap_or_default();
let offset = parts
.next()
.ok_or_else(|| "command.poll cursor 无效".to_string())?
.parse::<usize>()
.map_err(|_| "command.poll cursor offset 无效".to_string())?;
if parts.next().is_some()
|| version != PROCESS_SESSION_CURSOR_VERSION
|| cursor_process_id != process_id
|| offset > output.len()
|| !output.is_char_boundary(offset)
{
return Err("command.poll cursor 与当前进程输出不匹配".to_string());
}
Ok(offset)
}
fn write_process_session_record(root: &Path, record: &ProcessSessionRecord) -> Result<(), String> {
write_agent_runtime_json_sidecar_with_max_bytes(
root,
&process_session_record_relative_path(&record.process_id),
"Agent Runtime process session",
record,
PROCESS_SESSION_RECORD_MAX_BYTES,
)
}
fn read_process_session_record(
root: &Path,
process_id: &str,
) -> Result<Option<ProcessSessionRecord>, String> {
validate_process_id(process_id)?;
let mut record = read_agent_runtime_json_sidecar_with_max_bytes::<ProcessSessionRecord>(
root,
&process_session_record_relative_path(process_id),
"Agent Runtime process session",
PROCESS_SESSION_RECORD_MAX_BYTES,
)?;
if let Some(record) = record.as_mut() {
let normalized = normalize_process_session_record(record);
validate_process_session_record(root, record, process_id)?;
if normalized {
write_process_session_record(root, record)?;
}
}
Ok(record)
}
fn normalize_process_session_record(record: &mut ProcessSessionRecord) -> bool {
let legacy_schema = matches!(record.schema_version.as_str(), "1" | "2");
if !legacy_schema {
return false;
}
if record.schema_version == "1" {
record.sandbox_backend = "legacy-unknown".to_string();
record.sandbox_mode = "unknown".to_string();
record.network_access = "unknown".to_string();
record.sandbox_profile_version = "legacy-v1".to_string();
}
let legacy_active = matches!(
record.status.as_str(),
"prepared" | "launching" | "running" | "terminating"
);
record.schema_version = PROCESS_SESSION_SCHEMA_VERSION.to_string();
record.sandbox_establishment = "unknown".to_string();
record.target_exec = "unknown".to_string();
record.launch_failure_kind = Some(
if legacy_active {
"legacy-active-record"
} else {
"legacy-record"
}
.to_string(),
);
record.sandbox_ready_at = None;
record.exec_established_at = None;
if legacy_active {
record.status = "needs-reconciliation".to_string();
record.stdin_open = false;
record.needs_reconciliation = true;
record.terminal_at = Some(unix_timestamp());
record.updated_at = unix_timestamp();
}
legacy_active
}
fn validate_process_session_record(
root: &Path,
record: &ProcessSessionRecord,
process_id: &str,
) -> Result<(), String> {
if record.schema_version != PROCESS_SESSION_SCHEMA_VERSION
|| record.process_id != process_id
|| record.project_id != game_creator_agent_runtime_context_project_id(root)?
|| record.sandbox_backend.is_empty()
|| record.sandbox_mode.is_empty()
|| record.network_access.is_empty()
|| record.sandbox_profile_version.is_empty()
|| !matches!(
record.sandbox_establishment.as_str(),
"not-established" | "established" | "unknown"
)
|| !matches!(
record.target_exec.as_str(),
"not-attempted" | "established" | "failed" | "unknown"
)
|| record.launch_failure_kind.as_deref().is_some_and(|value| {
!matches!(
value,
"pre-exec-failed"
| "durable-commit-failed"
| "target-exec-failed"
| "launch-unknown"
| "legacy-active-record"
| "legacy-record"
| "start-audit-failed"
)
})
{
return Err("Agent Runtime process session 身份不匹配".to_string());
}
validate_process_id(&record.process_id)?;
if !matches!(
record.status.as_str(),
"prepared"
| "launching"
| "running"
| "terminating"
| "exited"
| "terminated"
| "timed-out"
| "output-limit-exceeded"
| "needs-reconciliation"
| "failed"
) {
return Err("Agent Runtime process session 状态无效".to_string());
}
if matches!(
record.status.as_str(),
"prepared" | "launching" | "running" | "terminating"
) && record.terminal_at.is_some()
{
return Err("运行中的 process session 不应有 terminalAt".to_string());
}
if !matches!(
record.status.as_str(),
"prepared" | "launching" | "running" | "terminating"
) && record.terminal_at.is_none()
{
return Err("终态 process session 缺少 terminalAt".to_string());
}
let launch_state_valid = match record.launch_failure_kind.as_deref() {
None => match record.status.as_str() {
"prepared" => {
record.sandbox_establishment == "not-established"
&& record.target_exec == "not-attempted"
&& !record.needs_reconciliation
}
"launching" => {
matches!(
record.sandbox_establishment.as_str(),
"not-established" | "established"
) && record.target_exec == "not-attempted"
&& !record.needs_reconciliation
}
"running" | "terminating" => {
record.sandbox_establishment == "established"
&& record.target_exec == "established"
&& !record.needs_reconciliation
}
"needs-reconciliation" => {
record.sandbox_establishment == "established"
&& record.target_exec == "established"
&& record.needs_reconciliation
}
"exited" | "terminated" | "timed-out" | "output-limit-exceeded" | "failed" => {
record.sandbox_establishment == "established" && record.target_exec == "established"
}
_ => false,
},
Some("pre-exec-failed") => {
record.status == "failed"
&& record.sandbox_establishment == "not-established"
&& record.target_exec == "not-attempted"
&& !record.needs_reconciliation
}
Some("durable-commit-failed") => {
record.status == "failed"
&& matches!(
record.sandbox_establishment.as_str(),
"not-established" | "established"
)
&& record.target_exec == "not-attempted"
&& !record.needs_reconciliation
}
Some("target-exec-failed") => {
matches!(record.status.as_str(), "failed" | "needs-reconciliation")
&& record.sandbox_establishment == "established"
&& record.target_exec == "failed"
&& (record.status == "needs-reconciliation") == record.needs_reconciliation
}
Some("launch-unknown") => {
record.status == "needs-reconciliation"
&& record.needs_reconciliation
&& record.target_exec == "unknown"
&& matches!(
record.sandbox_establishment.as_str(),
"established" | "unknown"
)
}
Some("legacy-active-record") => {
record.status == "needs-reconciliation"
&& record.needs_reconciliation
&& record.sandbox_establishment == "unknown"
&& record.target_exec == "unknown"
}
Some("legacy-record") => {
!matches!(
record.status.as_str(),
"prepared" | "launching" | "running" | "terminating"
) && record.sandbox_establishment == "unknown"
&& record.target_exec == "unknown"
&& (record.status != "needs-reconciliation" || record.needs_reconciliation)
}
Some("start-audit-failed") => {
record.status == "needs-reconciliation"
&& record.needs_reconciliation
&& record.sandbox_establishment == "established"
&& record.target_exec == "established"
}
Some(_) => false,
};
if !launch_state_valid {
return Err("process session 可信 launch 状态组合无效".to_string());
}
if record.started_at > record.updated_at
|| record.terminal_at.is_some_and(|terminal_at| {
record.started_at > terminal_at || terminal_at > record.updated_at
})
{
return Err("process session 生命周期时间顺序无效".to_string());
}
match record.sandbox_establishment.as_str() {
"established" if record.sandbox_ready_at.is_none() => {
return Err("已建立的 process session sandbox 缺少 ready 时间".to_string());
}
"not-established" | "unknown" if record.sandbox_ready_at.is_some() => {
return Err("未建立或未知的 process session sandbox 不应有 ready 时间".to_string());
}
_ => {}
}
match record.target_exec.as_str() {
"established" if record.exec_established_at.is_none() => {
return Err("已建立的 process session target 缺少 exec 时间".to_string());
}
"not-attempted" | "failed" | "unknown" if record.exec_established_at.is_some() => {
return Err("未建立的 process session target 不应有 exec 时间".to_string());
}
_ => {}
}
if let Some(sandbox_ready_at) = record.sandbox_ready_at {
if record.started_at > sandbox_ready_at
|| sandbox_ready_at > record.updated_at
|| record
.terminal_at
.is_some_and(|terminal_at| sandbox_ready_at > terminal_at)
{
return Err("process session sandbox-ready 时间顺序无效".to_string());
}
}
if let Some(exec_established_at) = record.exec_established_at {
if record
.sandbox_ready_at
.is_none_or(|sandbox_ready_at| sandbox_ready_at > exec_established_at)
|| exec_established_at > record.updated_at
|| record
.terminal_at
.is_some_and(|terminal_at| exec_established_at > terminal_at)
{
return Err("process session exec-established 时间顺序无效".to_string());
}
}
Ok(())
}
fn reconcile_stale_active_process_session(record: &mut ProcessSessionRecord) {
let previous_status = record.status.clone();
record.status = "needs-reconciliation".to_string();
record.stdin_open = false;
record.needs_reconciliation = true;
if previous_status == "prepared" {
record.sandbox_establishment = "unknown".to_string();
record.sandbox_ready_at = None;
record.target_exec = "unknown".to_string();
record.exec_established_at = None;
record.launch_failure_kind = Some("launch-unknown".to_string());
} else if previous_status == "launching" {
if record.sandbox_establishment != "established" {
record.sandbox_establishment = "unknown".to_string();
record.sandbox_ready_at = None;
}
record.target_exec = "unknown".to_string();
record.exec_established_at = None;
record.launch_failure_kind = Some("launch-unknown".to_string());
}
record.terminal_at = Some(unix_timestamp());
record.updated_at = unix_timestamp();
}
fn process_session_record_from_live(
live: &LiveProcessSession,
output: &ProcessOutputState,
) -> ProcessSessionRecord {
let output_sha256 = format!("{:x}", Sha256::digest(output.text.as_bytes()));
let terminal = output.status != "running";
ProcessSessionRecord {
schema_version: PROCESS_SESSION_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(),
owner_boot_id: process_session_boot_id().to_string(),
command_id: live.command_id.clone(),
program: live.program.clone(),
cwd: live.cwd.clone(),
sandbox_backend: live.sandbox_backend.clone(),
sandbox_mode: live.sandbox_mode.clone(),
network_access: live.network_access.clone(),
sandbox_profile_version: live.sandbox_profile_version.clone(),
sandbox_establishment: live.sandbox_establishment.clone(),
target_exec: live.target_exec.clone(),
launch_failure_kind: output.launch_failure_kind.clone(),
sandbox_ready_at: live.sandbox_ready_at,
exec_established_at: live.exec_established_at,
status: output.status.clone(),
exit_code: output.exit_code,
signal: output.signal.clone(),
stdin_open: output.stdin_open,
output_bytes: output.text.len(),
output_sha256,
output_ref: Some(process_session_transcript_relative_path(&live.process_id)),
source_fingerprint_before: live.source_fingerprint_before.clone(),
source_fingerprint_after: output.source_fingerprint_after.clone(),
source_changed: output.source_changed,
needs_reconciliation: output.needs_reconciliation,
started_at: live.started_at,
terminal_at: terminal.then(unix_timestamp),
updated_at: unix_timestamp(),
}
}
fn initial_process_session_record(
identity: &ProcessSessionIdentity,
process_id: &str,
command_id: &str,
spec: &ProjectCommandSpec,
launch: Option<&ProjectCommandLaunchSpec>,
source_fingerprint_before: &str,
status: &str,
) -> ProcessSessionRecord {
let now = unix_timestamp();
let sandbox_backend = launch
.map(|value| value.sandbox_backend.clone())
.unwrap_or_else(|| "test-unknown".to_string());
let sandbox_mode = launch
.map(|value| value.sandbox_mode.clone())
.unwrap_or_else(|| "unknown".to_string());
let network_access = launch
.map(|value| value.network_access.clone())
.unwrap_or_else(|| "unknown".to_string());
let sandbox_profile_version = launch
.map(|value| value.sandbox_profile_version.clone())
.unwrap_or_else(|| "test-v1".to_string());
ProcessSessionRecord {
schema_version: PROCESS_SESSION_SCHEMA_VERSION.to_string(),
project_id: identity.project_id.clone(),
agent_id: identity.agent_id.clone(),
task_id: identity.task_id.clone(),
conversation_session_id: identity.conversation_session_id.clone(),
run_id: identity.run_id.clone(),
start_action_id: identity.start_action_id.clone(),
start_action_fingerprint: identity.start_action_fingerprint.clone(),
process_id: process_id.to_string(),
owner_boot_id: process_session_boot_id().to_string(),
command_id: command_id.to_string(),
program: spec.program.clone(),
cwd: spec.cwd_relative.clone(),
sandbox_backend,
sandbox_mode,
network_access,
sandbox_profile_version,
sandbox_establishment: "not-established".to_string(),
target_exec: "not-attempted".to_string(),
launch_failure_kind: None,
sandbox_ready_at: None,
exec_established_at: None,
status: status.to_string(),
exit_code: None,
signal: None,
stdin_open: false,
output_bytes: 0,
output_sha256: format!("{:x}", Sha256::digest([])),
output_ref: None,
source_fingerprint_before: source_fingerprint_before.to_string(),
source_fingerprint_after: None,
source_changed: None,
needs_reconciliation: false,
started_at: now,
terminal_at: None,
updated_at: now,
}
}
#[cfg(not(target_os = "linux"))]
fn process_session_launch_failed(
root: &Path,
record: &mut ProcessSessionRecord,
error: String,
) -> String {
record.status = "failed".to_string();
record.target_exec = "not-attempted".to_string();
record.launch_failure_kind = Some("pre-exec-failed".to_string());
record.stdin_open = false;
record.terminal_at = Some(unix_timestamp());
record.updated_at = unix_timestamp();
match write_process_session_record(root, record) {
Ok(()) => error,
Err(record_error) => format!("{error}process session 失败终态无法落盘:{record_error}"),
}
}
fn validate_process_session_access(
record: &ProcessSessionRecord,
identity: &ProcessSessionIdentity,
) -> Result<(), String> {
if record.project_id != identity.project_id
|| record.agent_id != identity.agent_id
|| record.task_id != identity.task_id
|| record.conversation_session_id != identity.conversation_session_id
|| record.run_id != identity.run_id
{
return Err("process session 不属于当前 Agent run".to_string());
}
Ok(())
}
pub(crate) fn process_session_identity_for_run_at(
root: &Path,
agent_id: &str,
task_id: &str,
conversation_session_id: &str,
run_id: &str,
process_id: &str,
) -> Result<ProcessSessionIdentity, String> {
let record = read_process_session_record(root, process_id)?
.ok_or_else(|| "process session 不存在".to_string())?;
if record.agent_id != agent_id
|| record.task_id != task_id
|| record.conversation_session_id != conversation_session_id
|| record.run_id != run_id
{
return Err("process session 不属于当前 Agent run".to_string());
}
Ok(ProcessSessionIdentity {
project_id: record.project_id,
agent_id: record.agent_id,
task_id: record.task_id,
conversation_session_id: record.conversation_session_id,
run_id: record.run_id,
start_action_id: record.start_action_id,
start_action_fingerprint: record.start_action_fingerprint,
})
}
fn validate_process_session_transcript(
transcript: &ProcessSessionTranscript,
record: &ProcessSessionRecord,
) -> Result<(), String> {
let output_sha256 = format!("{:x}", Sha256::digest(transcript.output.as_bytes()));
if !matches!(transcript.schema_version.as_str(), "1" | "2")
|| transcript.project_id != record.project_id
|| transcript.agent_id != record.agent_id
|| transcript.task_id != record.task_id
|| transcript.conversation_session_id != record.conversation_session_id
|| transcript.run_id != record.run_id
|| transcript.start_action_id != record.start_action_id
|| transcript.start_action_fingerprint != record.start_action_fingerprint
|| transcript.process_id != record.process_id
|| transcript.output_bytes != transcript.output.len()
|| transcript.output_sha256 != output_sha256
|| record.output_bytes != transcript.output_bytes
|| record.output_sha256 != transcript.output_sha256
{
return Err("Agent Runtime process transcript 身份或摘要不匹配".to_string());
}
Ok(())
}
fn find_existing_start_action_record(
root: &Path,
identity: &ProcessSessionIdentity,
) -> Result<Option<ProcessSessionRecord>, String> {
let directory = root.join(".agent/runtime/process-sessions");
let entries = match fs::read_dir(&directory) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(format!("读取 process session 目录失败:{error}")),
};
for entry in entries {
let entry = entry.map_err(|error| format!("读取 process session 目录项失败:{error}"))?;
let name = entry.file_name();
let Some(name) = name.to_str() else {
continue;
};
let Some(process_id) = name
.strip_suffix(".json")
.filter(|value| !value.ends_with(".output"))
else {
continue;
};
if validate_process_id(process_id).is_err() {
continue;
}
let Some(record) = read_process_session_record(root, process_id)? else {
continue;
};
if record.agent_id == identity.agent_id
&& record.run_id == identity.run_id
&& record.start_action_id == identity.start_action_id
{
if record.start_action_fingerprint != identity.start_action_fingerprint {
return Err("command.start action identity 冲突".to_string());
}
return Ok(Some(record));
}
}
Ok(None)
}
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(())
}
fn process_session_command_builder(
launch: &ProjectCommandLaunchSpec,
#[cfg(target_os = "linux")] bridge: &ProcessSessionBridgeServer,
) -> Result<CommandBuilder, String> {
#[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 = {
let mut command = CommandBuilder::new(&launch.executable);
command.args(&launch.arguments);
command
};
command.cwd(&launch.cwd);
command.env_clear();
#[cfg(not(target_os = "linux"))]
for (name, value) in &launch.environment {
command.env(name, value);
}
#[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 reader = pair
.master
.try_clone_reader()
.map_err(|error| {
process_session_launch_failed(
root,
&mut durable_record,
format!("克隆 command.start PTY reader 失败:{error}"),
)
})
.map_err(|error| ProjectCommandError::new(ProjectCommandErrorStage::Execution, error))?;
let writer = pair
.master
.take_writer()
.map_err(|error| {
process_session_launch_failed(
root,
&mut durable_record,
format!("取得 command.start PTY writer 失败:{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);
#[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 (transcript, record) = {
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(),
};
(transcript, 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)]
struct AnsiStripper {
state: u8,
}
impl AnsiStripper {
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,
}
}
}
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 ansi = AnsiStripper::default();
let mut output_limit = false;
loop {
match reader.read(&mut buffer) {
Ok(0) => break,
Ok(read) => {
for byte in &buffer[..read] {
let before = pending.len();
ansi.push(*byte, &mut pending);
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;
}
pending.clear();
} else if pending.len() > PROCESS_SESSION_MAX_PENDING_LINE_BYTES {
output_limit = true;
break;
}
}
if output_limit {
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);
}
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.output_changed.notify_all();
drop(output);
if record_persisted && !needs_reconciliation {
if let Ok(mut registry) = process_session_registry().lock() {
registry.sessions.remove(&live.process_id);
}
}
}
fn live_process_session(process_id: &str) -> Result<Option<Arc<LiveProcessSession>>, String> {
validate_process_id(process_id)?;
Ok(process_session_registry()
.lock()
.map_err(|_| "process session registry 锁已损坏".to_string())?
.sessions
.get(process_id)
.cloned())
}
fn poll_result_from_output(
process_id: &str,
output: &str,
state: &ProcessOutputState,
sandbox_backend: &str,
sandbox_mode: &str,
network_access: &str,
sandbox_profile_version: &str,
sandbox_establishment: &str,
target_exec: &str,
launch_failure_kind: Option<&str>,
cursor: Option<&str>,
max_chars: usize,
) -> Result<ProcessSessionPollResult, String> {
let offset = parse_process_session_cursor(process_id, cursor, output)?;
let end = output[offset..]
.char_indices()
.nth(max_chars)
.map(|(index, _)| offset + index)
.unwrap_or(output.len());
let next_cursor = process_session_cursor(process_id, end);
Ok(ProcessSessionPollResult {
process_id: process_id.to_string(),
status: state.status.clone(),
output: output[offset..end].to_string(),
cursor: process_session_cursor(process_id, offset),
next_cursor,
has_more: end < output.len(),
stdin_open: state.stdin_open,
exit_code: state.exit_code,
signal: state.signal.clone(),
output_bytes: output.len(),
output_sha256: format!("{:x}", Sha256::digest(output.as_bytes())),
source_changed: state.source_changed,
needs_reconciliation: state.needs_reconciliation,
sandbox_backend: sandbox_backend.to_string(),
sandbox_mode: sandbox_mode.to_string(),
network_access: network_access.to_string(),
sandbox_profile_version: sandbox_profile_version.to_string(),
sandbox_establishment: sandbox_establishment.to_string(),
target_exec: target_exec.to_string(),
launch_failure_kind: launch_failure_kind.map(str::to_string),
})
}
pub(crate) fn poll_process_session_at(
root: &Path,
identity: &ProcessSessionIdentity,
process_id: &str,
cursor: Option<&str>,
max_chars: Option<usize>,
wait_ms: Option<u64>,
) -> Result<ProcessSessionPollResult, String> {
validate_process_session_identity(identity)?;
let max_chars = max_chars
.unwrap_or(PROCESS_SESSION_DEFAULT_POLL_CHARS)
.min(PROCESS_SESSION_MAX_POLL_CHARS);
let wait_ms = wait_ms.unwrap_or(0).min(PROCESS_SESSION_MAX_POLL_WAIT_MS);
if let Some(live) = live_process_session(process_id)? {
if live.root != root || live.identity != *identity {
return Err("process session 不属于当前 Agent run".to_string());
}
let mut output = live
.output
.lock()
.map_err(|_| "process session output 锁已损坏".to_string())?;
let initial_offset = parse_process_session_cursor(process_id, cursor, &output.text)?;
if wait_ms > 0 && initial_offset == output.text.len() && output.status == "running" {
let waited = live
.output_changed
.wait_timeout(output, Duration::from_millis(wait_ms))
.map_err(|_| "process session output 锁已损坏".to_string())?;
output = waited.0;
}
return poll_result_from_output(
process_id,
&output.text,
&output,
&live.sandbox_backend,
&live.sandbox_mode,
&live.network_access,
&live.sandbox_profile_version,
&live.sandbox_establishment,
&live.target_exec,
output.launch_failure_kind.as_deref(),
cursor,
max_chars,
);
}
let mut record = read_process_session_record(root, process_id)?
.ok_or_else(|| "process session 不存在".to_string())?;
validate_process_session_access(&record, identity)?;
if matches!(
record.status.as_str(),
"prepared" | "launching" | "running" | "terminating"
) && record.owner_boot_id != process_session_boot_id()
{
reconcile_stale_active_process_session(&mut record);
write_process_session_record(root, &record)?;
}
let transcript = if let Some(output_ref) = record.output_ref.as_deref() {
read_agent_runtime_json_sidecar_with_max_bytes::<ProcessSessionTranscript>(
root,
output_ref,
"Agent Runtime process transcript",
PROCESS_SESSION_TRANSCRIPT_MAX_BYTES,
)?
} else {
None
};
if let Some(transcript) = &transcript {
if let Err(error) = validate_process_session_transcript(transcript, &record) {
record.status = "needs-reconciliation".to_string();
record.stdin_open = false;
record.needs_reconciliation = true;
record.terminal_at = Some(unix_timestamp());
record.updated_at = unix_timestamp();
let _ = write_process_session_record(root, &record);
return Err(error);
}
}
let output = transcript
.as_ref()
.map(|value| value.output.as_str())
.unwrap_or_default();
let state = ProcessOutputState {
text: output.to_string(),
status: record.status,
exit_code: record.exit_code,
signal: record.signal,
stdin_open: record.stdin_open,
reader_finished: true,
output_limit_exceeded: false,
source_fingerprint_after: record.source_fingerprint_after,
source_changed: record.source_changed,
needs_reconciliation: record.needs_reconciliation,
launch_failure_kind: record.launch_failure_kind.clone(),
};
poll_result_from_output(
process_id,
output,
&state,
&record.sandbox_backend,
&record.sandbox_mode,
&record.network_access,
&record.sandbox_profile_version,
&record.sandbox_establishment,
&record.target_exec,
record.launch_failure_kind.as_deref(),
cursor,
max_chars,
)
}
pub(crate) fn write_process_session_stdin_at(
root: &Path,
identity: &ProcessSessionIdentity,
process_id: &str,
data: &str,
append_newline: bool,
eof: bool,
) -> Result<ProcessSessionStdinResult, String> {
write_process_session_stdin_at_with_after_write(
root,
identity,
process_id,
data,
append_newline,
eof,
|_| {},
)
}
fn write_process_session_stdin_at_with_after_write<F>(
root: &Path,
identity: &ProcessSessionIdentity,
process_id: &str,
data: &str,
append_newline: bool,
eof: bool,
after_write: F,
) -> Result<ProcessSessionStdinResult, String>
where
F: FnOnce(&LiveProcessSession),
{
validate_process_session_identity(identity)?;
let live = live_process_session(process_id)?
.ok_or_else(|| "process session 不在当前 Runner 中运行".to_string())?;
if live.root != root || live.identity != *identity {
return Err("process session 不属于当前 Agent run".to_string());
}
let mut bytes = data.as_bytes().to_vec();
if append_newline {
bytes.push(b'\n');
}
if bytes.len() > PROCESS_SESSION_MAX_STDIN_BYTES {
return Err(format!(
"command.stdin 单次最多写入 {PROCESS_SESSION_MAX_STDIN_BYTES} 字节"
));
}
if bytes.iter().any(|byte| *byte == 0) {
return Err("command.stdin 不接受 NUL 或二进制正文".to_string());
}
let content_sha256 = format!("{:x}", Sha256::digest(&bytes));
let mut writer = live
.writer
.lock()
.map_err(|_| "process session stdin 锁已损坏".to_string())?;
if live
.output
.lock()
.map_err(|_| "process session output 锁已损坏".to_string())?
.status
!= "running"
{
return Err("process session 已进入终态".to_string());
}
if !bytes.is_empty() {
let stream = writer
.as_mut()
.ok_or_else(|| "process session stdin 已关闭".to_string())?;
stream
.write_all(&bytes)
.and_then(|()| stream.flush())
.map_err(|error| format!("写入 process session stdin 失败:{error}"))?;
}
if eof {
writer.take();
}
drop(writer);
after_write(&live);
let mut output = live
.output
.lock()
.map_err(|_| "process session output 锁已损坏".to_string())?;
if eof || output.status != "running" {
output.stdin_open = false;
}
let record = process_session_record_from_live(&live, &output);
if let Err(error) = write_process_session_record(root, &record) {
output.status = "needs-reconciliation".to_string();
output.needs_reconciliation = true;
output.stdin_open = false;
let reconciliation = process_session_record_from_live(&live, &output);
let _ = write_process_session_record(root, &reconciliation);
let _ = live.control.send(ProcessControl::Terminate);
return Err(format!(
"command.stdin 已写入但状态无法落盘,需要人工核对:{error}"
));
}
Ok(ProcessSessionStdinResult {
process_id: process_id.to_string(),
bytes_written: bytes.len(),
content_sha256,
stdin_open: output.stdin_open,
eof,
sandbox_backend: live.sandbox_backend.clone(),
sandbox_mode: live.sandbox_mode.clone(),
network_access: live.network_access.clone(),
sandbox_profile_version: live.sandbox_profile_version.clone(),
})
}
pub(crate) fn terminate_process_session_at(
root: &Path,
identity: &ProcessSessionIdentity,
process_id: &str,
cursor: Option<&str>,
) -> Result<ProcessSessionPollResult, String> {
validate_process_session_identity(identity)?;
if let Some(live) = live_process_session(process_id)? {
if live.root != root || live.identity != *identity {
return Err("process session 不属于当前 Agent run".to_string());
}
let running = live
.output
.lock()
.map_err(|_| "process session output 锁已损坏".to_string())?
.status
== "running";
if running {
live.control
.send(ProcessControl::Terminate)
.map_err(|_| "process session 监督线程已结束".to_string())?;
let mut output = live
.output
.lock()
.map_err(|_| "process session output 锁已损坏".to_string())?;
let deadline = std::time::Instant::now()
+ Duration::from_millis(PROCESS_SESSION_TERMINATE_GRACE_MS + 1_500);
while output.status == "running" && std::time::Instant::now() < deadline {
let remaining = deadline.saturating_duration_since(std::time::Instant::now());
let waited = live
.output_changed
.wait_timeout(output, remaining.min(Duration::from_millis(100)))
.map_err(|_| "process session output 锁已损坏".to_string())?;
output = waited.0;
}
}
let mut result =
poll_process_session_at(root, identity, process_id, cursor, Some(1), Some(0))?;
let cursor_offset = result
.cursor
.rsplit_once(':')
.and_then(|(_, offset)| offset.parse::<usize>().ok())
.ok_or_else(|| "command.terminate 返回了无效 cursor".to_string())?;
result.output.clear();
result.next_cursor = result.cursor.clone();
result.has_more = cursor_offset < result.output_bytes;
return Ok(result);
}
let mut result = poll_process_session_at(root, identity, process_id, cursor, Some(1), Some(0))?;
let cursor_offset = result
.cursor
.rsplit_once(':')
.and_then(|(_, offset)| offset.parse::<usize>().ok())
.ok_or_else(|| "command.terminate 返回了无效 cursor".to_string())?;
result.output.clear();
result.next_cursor = result.cursor.clone();
result.has_more = cursor_offset < result.output_bytes;
Ok(result)
}
pub(crate) fn mark_process_session_start_audit_failure_at(
root: &Path,
process_id: &str,
error: &str,
) -> Result<(), String> {
if let Some(live) = live_process_session(process_id)? {
if live.root != root {
return Err("process session 不属于当前项目".to_string());
}
mark_process_session_reconciliation(
&live,
&format!("command.start audit persistence failed: {error}"),
);
if let Ok(mut output) = live.output.lock() {
output.launch_failure_kind = Some("start-audit-failed".to_string());
}
let _ = live.control.send(ProcessControl::Terminate);
}
let mut record = read_process_session_record(root, process_id)?
.ok_or_else(|| "command.start audit 失败后 process record 缺失".to_string())?;
record.status = "needs-reconciliation".to_string();
record.stdin_open = false;
record.needs_reconciliation = true;
record.launch_failure_kind = Some("start-audit-failed".to_string());
record.signal = Some(redact_agent_runtime_project_paths(root, error, 240));
record.terminal_at = Some(unix_timestamp());
record.updated_at = unix_timestamp();
write_process_session_record(root, &record)
}
pub(crate) fn active_process_session_records_at(
root: &Path,
agent_id: Option<&str>,
run_id: Option<&str>,
) -> Result<Vec<ProcessSessionRecord>, String> {
let live_sessions = process_session_registry()
.lock()
.map_err(|_| "process session registry 锁已损坏".to_string())?
.sessions
.values()
.filter(|live| live.root == root)
.cloned()
.collect::<Vec<_>>();
let mut records = Vec::with_capacity(live_sessions.len());
for live in live_sessions {
if agent_id.is_some_and(|value| value != live.identity.agent_id.as_str())
|| run_id.is_some_and(|value| value != live.identity.run_id.as_str())
{
continue;
}
let output = live
.output
.lock()
.map_err(|_| "process session output 锁已损坏".to_string())?;
records.push(process_session_record_from_live(&live, &output));
}
let directory = root.join(".agent/runtime/process-sessions");
let entries = match fs::read_dir(&directory) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
records.sort_by(|left, right| left.started_at.cmp(&right.started_at));
return Ok(records);
}
Err(error) => return Err(format!("读取 process session 目录失败:{error}")),
};
for entry in entries {
let entry = entry.map_err(|error| format!("读取 process session 目录项失败:{error}"))?;
let name = entry.file_name();
let Some(name) = name.to_str() else {
continue;
};
let Some(process_id) = name
.strip_suffix(".json")
.filter(|value| !value.ends_with(".output"))
else {
continue;
};
if validate_process_id(process_id).is_err() {
continue;
}
if records.iter().any(|record| record.process_id == process_id) {
continue;
}
let Some(mut record) = read_process_session_record(root, process_id)? else {
continue;
};
if matches!(
record.status.as_str(),
"prepared" | "launching" | "running" | "terminating"
) && record.owner_boot_id != process_session_boot_id()
{
reconcile_stale_active_process_session(&mut record);
write_process_session_record(root, &record)?;
}
if (record.needs_reconciliation
|| matches!(
record.status.as_str(),
"prepared" | "launching" | "running" | "terminating" | "needs-reconciliation"
))
&& agent_id.is_none_or(|value| value == record.agent_id)
&& run_id.is_none_or(|value| value == record.run_id)
{
records.push(record);
}
}
records.sort_by(|left, right| left.started_at.cmp(&right.started_at));
Ok(records)
}
pub(crate) fn has_active_process_sessions_at(root: &Path) -> Result<bool, String> {
if !active_process_session_records_at(root, None, None)?.is_empty() {
return Ok(true);
}
#[cfg(target_os = "linux")]
{
return Ok(pending_process_launch_registry()
.lock()
.map_err(|_| "pending process launch registry 锁已损坏".to_string())?
.launches
.values()
.any(|launch| launch.root == root));
}
#[cfg(not(target_os = "linux"))]
Ok(false)
}
pub(crate) fn terminate_process_sessions_for_run_at(
root: &Path,
agent_id: &str,
run_id: &str,
) -> Result<(), String> {
let records = active_process_session_records_at(root, Some(agent_id), Some(run_id))?;
for record in &records {
if record.status == "needs-reconciliation" || record.needs_reconciliation {
return Err(format!(
"进程会话 {} 需要人工核对,不能把 run 标记为已取消",
record.process_id
));
}
let identity = ProcessSessionIdentity {
project_id: record.project_id.clone(),
agent_id: record.agent_id.clone(),
task_id: record.task_id.clone(),
conversation_session_id: record.conversation_session_id.clone(),
run_id: record.run_id.clone(),
start_action_id: record.start_action_id.clone(),
start_action_fingerprint: record.start_action_fingerprint.clone(),
};
let terminal = terminate_process_session_at(root, &identity, &record.process_id, None)?;
if terminal.status == "running" || terminal.needs_reconciliation {
return Err(format!(
"进程会话 {} 尚未形成可信终态,不能把 run 标记为已取消",
record.process_id
));
}
}
if active_process_session_records_at(root, Some(agent_id), Some(run_id))?.is_empty() {
Ok(())
} else {
Err("仍有未收束的 process session,不能把 run 标记为已取消".to_string())
}
}
pub(crate) fn shutdown_all_process_sessions() {
#[cfg(target_os = "linux")]
{
let pending = pending_process_launch_registry()
.lock()
.ok()
.map(|mut registry| {
registry
.launches
.values_mut()
.filter_map(|launch| {
launch.shutdown_requested = true;
launch.process_group_leader
})
.collect::<Vec<_>>()
})
.unwrap_or_default();
for process_group_leader in pending {
unsafe {
libc::kill(-process_group_leader, libc::SIGKILL);
}
}
}
let sessions = process_session_registry()
.lock()
.ok()
.map(|registry| registry.sessions.values().cloned().collect::<Vec<_>>())
.unwrap_or_default();
for live in sessions {
let _ = live.control.send(ProcessControl::Shutdown);
}
}
pub(crate) fn shutdown_all_process_sessions_and_wait(timeout: Duration) -> Result<(), String> {
shutdown_all_process_sessions();
let deadline = std::time::Instant::now() + timeout;
loop {
let sessions = process_session_registry()
.lock()
.map_err(|_| "process session registry 锁已损坏".to_string())?
.sessions
.values()
.cloned()
.collect::<Vec<_>>();
let running = sessions.iter().filter(|live| {
live.output
.lock()
.map(|output| {
matches!(
output.status.as_str(),
"prepared" | "launching" | "running" | "terminating"
)
})
.unwrap_or(true)
});
if running.count() == 0 {
#[cfg(target_os = "linux")]
let pending_empty = pending_process_launch_registry()
.lock()
.map_err(|_| "pending process launch registry 锁已损坏".to_string())?
.launches
.is_empty();
#[cfg(not(target_os = "linux"))]
let pending_empty = true;
if pending_empty {
return Ok(());
}
}
if std::time::Instant::now() >= deadline {
return Err("Runner 退出前未能回收全部 process session".to_string());
}
thread::sleep(Duration::from_millis(25));
}
}
#[cfg(test)]
pub(crate) fn clear_process_session_registry_for_tests() {
shutdown_all_process_sessions();
if let Ok(mut registry) = process_session_registry().lock() {
registry.sessions.clear();
}
#[cfg(target_os = "linux")]
if let Ok(mut registry) = pending_process_launch_registry().lock() {
registry.launches.clear();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(target_os = "linux")]
use std::process::Stdio;
static PROCESS_SESSION_TEST_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
fn process_session_test_guard() -> std::sync::MutexGuard<'static, ()> {
PROCESS_SESSION_TEST_LOCK
.get_or_init(|| Mutex::new(()))
.lock()
.expect("process session test lock")
}
fn process_identity(project_id: &str) -> ProcessSessionIdentity {
ProcessSessionIdentity {
project_id: project_id.to_string(),
agent_id: "code-prototype".to_string(),
task_id: "code-prototype".to_string(),
conversation_session_id: "session-process-test".to_string(),
run_id: "run-process-test".to_string(),
start_action_id: "action-process-start-test".to_string(),
start_action_fingerprint: "a".repeat(64),
}
}
#[test]
fn process_session_cursor_preserves_unicode_boundaries() {
let process_id = "proc-0123456789abcdef0123456789abcdef";
let state = ProcessOutputState {
text: "甲乙abc".to_string(),
status: "running".to_string(),
exit_code: None,
signal: None,
stdin_open: true,
reader_finished: false,
output_limit_exceeded: false,
source_fingerprint_after: None,
source_changed: None,
needs_reconciliation: false,
launch_failure_kind: None,
};
let first = poll_result_from_output(
process_id,
&state.text,
&state,
"test",
"test",
"test",
"test-v1",
"established",
"established",
None,
None,
2,
)
.expect("first unicode page");
assert_eq!(first.output, "甲乙");
assert!(first.has_more);
let second = poll_result_from_output(
process_id,
&state.text,
&state,
"test",
"test",
"test",
"test-v1",
"established",
"established",
None,
Some(&first.next_cursor),
3,
)
.expect("second unicode page");
assert_eq!(second.output, "abc");
assert!(!second.has_more);
}
fn write_legacy_process_session_record(
root: &Path,
record: &ProcessSessionRecord,
schema_version: &str,
) {
let mut value = serde_json::to_value(record).expect("serialize legacy record");
let object = value.as_object_mut().expect("legacy record object");
object.insert(
"schemaVersion".to_string(),
serde_json::Value::String(schema_version.to_string()),
);
for field in [
"sandboxEstablishment",
"targetExec",
"launchFailureKind",
"sandboxReadyAt",
"execEstablishedAt",
] {
object.remove(field);
}
if schema_version == "1" {
for field in [
"sandboxBackend",
"sandboxMode",
"networkAccess",
"sandboxProfileVersion",
] {
object.remove(field);
}
}
write_agent_runtime_json_sidecar_with_max_bytes(
root,
&process_session_record_relative_path(&record.process_id),
"legacy process session",
&value,
PROCESS_SESSION_RECORD_MAX_BYTES,
)
.expect("write legacy process record");
}
#[test]
fn process_session_v1_v2_active_records_migrate_to_v3_reconciliation() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "legacy-active-project", "Legacy Active Project")
.expect("initialize project");
for (index, schema_version) in ["1", "2"].into_iter().enumerate() {
let mut identity = process_identity("legacy-active-project");
identity.start_action_id = format!("legacy-active-action-{index}");
identity.start_action_fingerprint = format!("{}", index + 1).repeat(64);
let process_id = format!("proc-{:032x}", index + 1);
let record = initial_process_session_record(
&identity,
&process_id,
&format!("cmd-legacy-active-{index}"),
&resolve_project_command_spec_at(root, "bash", &["--version".to_string()], ".", 30)
.expect("resolve command"),
None,
&"a".repeat(64),
"running",
);
write_legacy_process_session_record(root, &record, schema_version);
let migrated = read_process_session_record(root, &process_id)
.expect("read migrated active record")
.expect("active record exists");
assert_eq!(migrated.schema_version, "3");
assert_eq!(migrated.status, "needs-reconciliation");
assert_eq!(migrated.sandbox_establishment, "unknown");
assert_eq!(migrated.target_exec, "unknown");
assert_eq!(
migrated.launch_failure_kind.as_deref(),
Some("legacy-active-record")
);
assert!(migrated.needs_reconciliation);
assert!(migrated.terminal_at.is_some());
let repeated = read_process_session_record(root, &process_id)
.expect("read migrated active record again")
.expect("active record remains");
assert_eq!(repeated, migrated);
}
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_v1_v2_terminal_records_remain_readable_without_reconciliation() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "legacy-terminal-project", "Legacy Terminal Project")
.expect("initialize project");
for (index, schema_version) in ["1", "2"].into_iter().enumerate() {
let mut identity = process_identity("legacy-terminal-project");
identity.start_action_id = format!("legacy-terminal-action-{index}");
identity.start_action_fingerprint = format!("{}", index + 3).repeat(64);
let process_id = format!("proc-{:032x}", index + 16);
let mut record = initial_process_session_record(
&identity,
&process_id,
&format!("cmd-legacy-terminal-{index}"),
&resolve_project_command_spec_at(root, "bash", &["--version".to_string()], ".", 30)
.expect("resolve command"),
None,
&"b".repeat(64),
"exited",
);
record.exit_code = Some(index as i32);
record.terminal_at = Some(record.started_at);
let output = format!("LEGACY-TERMINAL-{schema_version}");
record.output_bytes = output.len();
record.output_sha256 = format!("{:x}", Sha256::digest(output.as_bytes()));
record.output_ref = Some(process_session_transcript_relative_path(&process_id));
let transcript = ProcessSessionTranscript {
schema_version: schema_version.to_string(),
project_id: identity.project_id.clone(),
agent_id: identity.agent_id.clone(),
task_id: identity.task_id.clone(),
conversation_session_id: identity.conversation_session_id.clone(),
run_id: identity.run_id.clone(),
start_action_id: identity.start_action_id.clone(),
start_action_fingerprint: identity.start_action_fingerprint.clone(),
process_id: process_id.clone(),
output: output.clone(),
output_sha256: record.output_sha256.clone(),
output_bytes: output.len(),
updated_at: record.updated_at,
};
write_agent_runtime_json_sidecar_with_max_bytes(
root,
record.output_ref.as_deref().expect("legacy output ref"),
"legacy process transcript",
&transcript,
PROCESS_SESSION_TRANSCRIPT_MAX_BYTES,
)
.expect("write legacy process transcript");
write_legacy_process_session_record(root, &record, schema_version);
let migrated = read_process_session_record(root, &process_id)
.expect("read migrated terminal record")
.expect("terminal record exists");
assert_eq!(migrated.schema_version, "3");
assert_eq!(migrated.status, "exited");
assert_eq!(migrated.sandbox_establishment, "unknown");
assert_eq!(migrated.target_exec, "unknown");
assert_eq!(
migrated.launch_failure_kind.as_deref(),
Some("legacy-record")
);
assert!(!migrated.needs_reconciliation);
let poll =
poll_process_session_at(root, &identity, &process_id, None, Some(8_000), Some(0))
.expect("poll migrated terminal record");
assert_eq!(poll.status, "exited");
assert_eq!(poll.exit_code, Some(index as i32));
assert_eq!(poll.output, output);
}
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_v3_rejects_untrusted_state_combinations_and_timestamps() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "v3-validation-project", "V3 Validation Project")
.expect("initialize project");
let identity = process_identity("v3-validation-project");
let process_id = "proc-bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
let spec =
resolve_project_command_spec_at(root, "bash", &["--version".to_string()], ".", 30)
.expect("resolve command");
let mut record = initial_process_session_record(
&identity,
process_id,
"cmd-v3-validation",
&spec,
None,
&"c".repeat(64),
"running",
);
assert!(validate_process_session_record(root, &record, process_id).is_err());
record.sandbox_establishment = "established".to_string();
record.target_exec = "established".to_string();
record.sandbox_ready_at = Some(record.started_at);
record.exec_established_at = Some(record.started_at);
record.launch_failure_kind = Some("arbitrary".to_string());
assert!(validate_process_session_record(root, &record, process_id).is_err());
record.launch_failure_kind = None;
record.sandbox_ready_at = Some(record.started_at.saturating_add(2));
record.exec_established_at = Some(record.started_at.saturating_add(1));
assert!(validate_process_session_record(root, &record, process_id).is_err());
record.status = "failed".to_string();
record.sandbox_establishment = "unknown".to_string();
record.target_exec = "unknown".to_string();
record.launch_failure_kind = Some("launch-unknown".to_string());
record.sandbox_ready_at = None;
record.exec_established_at = None;
record.terminal_at = Some(record.started_at);
record.needs_reconciliation = false;
assert!(validate_process_session_record(root, &record, process_id).is_err());
record.status = "needs-reconciliation".to_string();
record.needs_reconciliation = true;
assert!(validate_process_session_record(root, &record, process_id).is_ok());
record.needs_reconciliation = false;
assert!(validate_process_session_record(root, &record, process_id).is_err());
record.needs_reconciliation = true;
record.sandbox_establishment = "established".to_string();
record.target_exec = "established".to_string();
record.launch_failure_kind = Some("start-audit-failed".to_string());
record.sandbox_ready_at = Some(record.started_at);
record.exec_established_at = Some(record.started_at);
assert!(validate_process_session_record(root, &record, process_id).is_ok());
record.status = "failed".to_string();
assert!(validate_process_session_record(root, &record, process_id).is_err());
record.status = "launching".to_string();
record.needs_reconciliation = false;
record.target_exec = "not-attempted".to_string();
record.launch_failure_kind = None;
record.exec_established_at = None;
record.terminal_at = None;
assert!(validate_process_session_record(root, &record, process_id).is_ok());
record.sandbox_ready_at = None;
assert!(validate_process_session_record(root, &record, process_id).is_err());
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn pending_process_launch_blocks_idle_until_guard_is_dropped() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "pending-launch-project", "Pending Launch Project")
.expect("initialize project");
let pending = reserve_pending_process_launch(
root,
"code-prototype",
"proc-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
)
.expect("register pending launch");
assert!(has_active_process_sessions_at(root).expect("pending launch is active"));
shutdown_all_process_sessions();
assert!(
activate_pending_process_launch("proc-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", i32::MAX,)
.expect_err("shutdown cancels a launch before pid activation")
.contains("shutdown")
);
drop(pending);
assert!(!has_active_process_sessions_at(root).expect("pending launch removed"));
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_ansi_stripper_handles_split_csi_and_osc() {
let mut stripper = AnsiStripper::default();
let mut visible = Vec::new();
for chunk in [
b"A\x1b[3".as_slice(),
b"1mB\x1b]52;c;secret".as_slice(),
b"\x07C\x1b[0m\n".as_slice(),
] {
for byte in chunk {
stripper.push(*byte, &mut visible);
}
}
assert_eq!(String::from_utf8(visible).expect("utf8"), "ABC\n");
}
#[test]
fn process_session_real_pty_streams_stdin_and_terminates() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "process-project", "Process Project")
.expect("initialize project");
fs::write(
root.join("package.json"),
r#"{"scripts":{"dev":"node fixture.js"}}"#,
)
.expect("write package.json");
fs::write(
root.join("fixture.js"),
r#"
process.stdin.setEncoding('utf8');
console.log('\u001b[31mREADY\u001b[0m');
console.log(`BRIDGE_ENV:${Object.keys(globalThis['process']['env']).filter((name) => name.includes('PROCESS_SESSION_BRIDGE')).join(',')}`);
console.log(`TARGET_ARGV:${process.argv.join('|')}`);
process.stdin.on('data', (chunk) => console.log(`ECHO:${chunk.trim()}`));
process.on('SIGTERM', () => { console.log('STOPPED'); process.exit(0); });
setInterval(() => {}, 1000);
"#,
)
.expect("write fixture");
let spec = resolve_project_command_spec_at(
root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
30,
)
.expect("resolve npm command");
let identity = process_identity("process-project");
let source_fingerprint =
project_command_source_fingerprint(root).expect("source fingerprint");
let mut poll = start_process_session_at(root, identity.clone(), &spec, source_fingerprint)
.expect("start process session");
assert!(poll.output.is_empty());
assert_eq!(poll.cursor, poll.next_cursor);
#[cfg(target_os = "linux")]
{
assert_eq!(poll.sandbox_backend, "bubblewrap");
assert_eq!(poll.sandbox_mode, "workspace-write");
assert_eq!(poll.network_access, "disabled");
assert_eq!(poll.sandbox_profile_version, "workspace-v1");
assert_eq!(poll.sandbox_establishment, "established");
assert_eq!(poll.target_exec, "established");
assert_eq!(poll.launch_failure_kind, None);
}
assert!(has_active_process_sessions_at(root).expect("active process probe"));
let mut combined = poll.output.clone();
for _ in 0..20 {
if combined.contains("READY") {
break;
}
poll = poll_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(500),
)
.expect("poll ready");
combined.push_str(&poll.output);
}
assert!(combined.contains("READY"), "output: {combined}");
assert!(!combined.contains("[31m"), "output: {combined}");
let mut foreign_identity = identity.clone();
foreign_identity.agent_id = "art-director".to_string();
assert!(poll_process_session_at(
root,
&foreign_identity,
&poll.process_id,
None,
Some(10),
Some(0),
)
.is_err());
assert!(write_process_session_stdin_at(
root,
&foreign_identity,
&poll.process_id,
"blocked",
true,
false,
)
.is_err());
let stdin =
write_process_session_stdin_at(root, &identity, &poll.process_id, "你好", true, false)
.expect("write stdin");
assert_eq!(stdin.bytes_written, "你好\n".len());
let mut echo = String::new();
for _ in 0..20 {
poll = poll_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(500),
)
.expect("poll echo");
echo.push_str(&poll.output);
if echo.contains("ECHO:你好") {
break;
}
}
assert!(echo.contains("ECHO:你好"), "output: {echo}");
let terminal = terminate_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
)
.expect("terminate process session");
assert_ne!(terminal.status, "running");
let record = read_process_session_record(root, &poll.process_id)
.expect("read record")
.expect("record exists");
assert_eq!(record.status, "terminated");
assert!(record.output_ref.is_some());
#[cfg(target_os = "linux")]
{
assert_eq!(record.sandbox_backend, "bubblewrap");
assert_eq!(record.sandbox_mode, "workspace-write");
assert_eq!(record.network_access, "disabled");
assert_eq!(record.sandbox_profile_version, "workspace-v1");
}
let transcript =
read_agent_runtime_json_sidecar_with_max_bytes::<ProcessSessionTranscript>(
root,
record.output_ref.as_deref().expect("transcript ref"),
"Agent Runtime process transcript",
PROCESS_SESSION_TRANSCRIPT_MAX_BYTES,
)
.expect("read transcript")
.expect("transcript exists");
let transcript_lines = transcript.output.lines().collect::<Vec<_>>();
assert!(
transcript_lines.contains(&"READY"),
"{:?}",
transcript.output
);
assert!(
transcript_lines.contains(&"ECHO:你好"),
"{:?}",
transcript.output
);
assert!(
transcript_lines.contains(&"STOPPED"),
"{:?}",
transcript.output
);
assert!(transcript_lines.contains(&"BRIDGE_ENV:"));
let record_json =
fs::read_to_string(root.join(process_session_record_relative_path(&record.process_id)))
.expect("read process record json");
for private_marker in [
"GENARRATIVE_PROCESS_SESSION_BRIDGE_ENDPOINT",
"GENARRATIVE_PROCESS_SESSION_BRIDGE_NONCE",
"genarrative-ps-",
"sandbox_ready",
"commit_exec",
"exec_established",
] {
assert!(
!transcript.output.contains(private_marker),
"private marker leaked: {private_marker}: {:?}",
transcript.output
);
assert!(
!record_json.contains(private_marker),
"private marker leaked to record: {private_marker}: {record_json}"
);
}
assert!(!has_active_process_sessions_at(root).expect("terminal process probe"));
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_graceful_terminate_keeps_wrapper_alive_for_target_cleanup() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "graceful-process-project", "Graceful Process Project")
.expect("initialize project");
let spec = resolve_project_command_spec_at(
root,
"bash",
&[
"-lc".to_string(),
"(trap 'sleep 0.4; printf done > graceful-marker.txt; exit 0' TERM; while :; do sleep 1; done) & printf 'READY\\n'; exit 0".to_string(),
],
".",
30,
)
.expect("resolve graceful command");
let identity = process_identity("graceful-process-project");
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let mut poll =
start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start");
for _ in 0..20 {
if poll.output.contains("READY") {
break;
}
poll = poll_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(250),
)
.expect("poll graceful ready");
}
assert!(poll.output.contains("READY"));
let terminated = terminate_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
)
.expect("graceful terminate");
assert_eq!(terminated.status, "terminated");
assert!(!terminated.needs_reconciliation);
assert_eq!(
fs::read_to_string(root.join("graceful-marker.txt"))
.expect("target completed delayed SIGTERM cleanup"),
"done"
);
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_live_registry_blocks_when_durable_record_is_missing() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "live-record-project", "Live Record Project")
.expect("initialize project");
let spec = resolve_project_command_spec_at(
root,
"bash",
&[
"-lc".to_string(),
"printf 'READY\\n'; while :; do sleep 1; done".to_string(),
],
".",
30,
)
.expect("resolve live record command");
let identity = process_identity("live-record-project");
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let started =
start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start");
fs::remove_file(root.join(process_session_record_relative_path(&started.process_id)))
.expect("remove durable process record");
let active = active_process_session_records_at(
root,
Some(identity.agent_id.as_str()),
Some(identity.run_id.as_str()),
)
.expect("live registry remains authoritative blocker");
assert!(active
.iter()
.any(|record| record.process_id == started.process_id));
assert!(has_active_process_sessions_at(root).expect("live registry blocks idle"));
terminate_process_session_at(root, &identity, &started.process_id, None)
.expect("terminate live record fixture");
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_fast_exit_zero_and_seven_keep_same_process_id() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
for exit_code in [0, 7] {
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
let project_id = format!("fast-exit-{exit_code}-project");
init_local_game_project_at(root, &project_id, "Fast Exit Project")
.expect("initialize project");
let spec = resolve_project_command_spec_at(
root,
"bash",
&[
"-lc".to_string(),
format!("printf 'FAST-{exit_code}\\n'; exit {exit_code}"),
],
".",
30,
)
.expect("resolve fast exit command");
let mut identity = process_identity(&project_id);
identity.start_action_id = format!("fast-exit-action-{exit_code}");
identity.start_action_fingerprint = format!("{}", exit_code + 1).repeat(64);
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let started = start_process_session_at(root, identity.clone(), &spec, fingerprint)
.expect("start");
let process_id = started.process_id.clone();
assert!(started.output.is_empty());
assert_eq!(started.cursor, started.next_cursor);
assert_eq!(started.sandbox_establishment, "established");
assert_eq!(started.target_exec, "established");
let mut poll = started;
let mut output = String::new();
for _ in 0..30 {
poll = poll_process_session_at(
root,
&identity,
&process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(250),
)
.expect("poll fast exit");
assert_eq!(poll.process_id, process_id);
output.push_str(&poll.output);
if poll.status != "running" && !poll.has_more {
break;
}
}
assert_eq!(poll.status, "exited", "{output}");
assert_eq!(poll.exit_code, Some(exit_code), "{output}");
assert!(output.contains(&format!("FAST-{exit_code}")), "{output}");
let record = read_process_session_record(root, &process_id)
.expect("read fast exit record")
.expect("fast exit record exists");
assert_eq!(record.process_id, process_id);
assert_eq!(record.target_exec, "established");
assert!(record.exec_established_at.is_some());
clear_process_session_registry_for_tests();
}
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_durable_commit_failure_runs_no_target_and_writes_no_record() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "durable-failure-project", "Durable Failure Project")
.expect("initialize project");
let spec = resolve_project_command_spec_at(
root,
"bash",
&[
"-lc".to_string(),
"printf ran > durable-target-ran.txt".to_string(),
],
".",
30,
)
.expect("resolve durable failure command");
let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch");
let identity = process_identity("durable-failure-project");
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let error = start_prepared_process_session_at(
root,
identity.clone(),
&spec,
&launch,
fingerprint,
|| {
assert!(!root.join("durable-target-ran.txt").exists());
assert!(find_existing_start_action_record(root, &identity)
.expect("record absent inside durable callback")
.is_none());
Err("forced durable commit failure".to_string())
},
)
.expect_err("durable commit must fail");
assert_eq!(error.stage(), ProjectCommandErrorStage::DurableCommit);
assert!(!root.join("durable-target-ran.txt").exists());
assert!(find_existing_start_action_record(root, &identity)
.expect("search process record")
.is_none());
assert!(!has_active_process_sessions_at(root).expect("no pending launch"));
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_slow_durable_commit_keeps_target_blocked_until_commit() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "slow-commit-project", "Slow Commit Project")
.expect("initialize project");
let spec = resolve_project_command_spec_at(
root,
"bash",
&[
"-lc".to_string(),
"printf committed > slow-commit-target.txt".to_string(),
],
".",
30,
)
.expect("resolve slow commit command");
let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch");
let identity = process_identity("slow-commit-project");
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let started_at = std::time::Instant::now();
let result = start_prepared_process_session_at(
root,
identity.clone(),
&spec,
&launch,
fingerprint,
|| {
thread::sleep(Duration::from_millis(3_200));
assert!(!root.join("slow-commit-target.txt").exists());
Ok(())
},
)
.expect("slow durable commit must not time out in child");
assert!(started_at.elapsed() >= Duration::from_millis(3_200));
assert_eq!(result.sandbox_establishment, "established");
assert_eq!(result.target_exec, "established");
let mut poll = result;
for _ in 0..20 {
poll = poll_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(250),
)
.expect("poll slow commit target");
if poll.status != "running" {
break;
}
}
assert_eq!(poll.status, "exited");
assert_eq!(
fs::read_to_string(root.join("slow-commit-target.txt"))
.expect("read slow commit marker"),
"committed"
);
clear_process_session_registry_for_tests();
}
#[cfg(windows)]
#[test]
fn process_session_windows_replay_runs_durable_commit_only_once() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "windows-replay-project", "Windows Replay Project")
.expect("initialize project");
fs::write(
root.join("package.json"),
r#"{"scripts":{"dev":"node fixture.js"}}"#,
)
.expect("write package.json");
fs::write(root.join("fixture.js"), "setInterval(() => {}, 1000);\n")
.expect("write fixture");
let spec = resolve_project_command_spec_at(
root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
30,
)
.expect("resolve npm command");
let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch");
let identity = process_identity("windows-replay-project");
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let commit_count = std::sync::atomic::AtomicUsize::new(0);
let first = start_prepared_process_session_at(
root,
identity.clone(),
&spec,
&launch,
fingerprint.clone(),
|| {
commit_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
},
)
.expect("first start");
let second = start_prepared_process_session_at(
root,
identity.clone(),
&spec,
&launch,
fingerprint,
|| {
commit_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
},
)
.expect("idempotent replay");
assert_eq!(first.process_id, second.process_id);
assert_eq!(commit_count.load(std::sync::atomic::Ordering::SeqCst), 1);
let record = read_process_session_record(root, &first.process_id)
.expect("read Windows replay record")
.expect("Windows replay record exists");
assert_eq!(record.sandbox_ready_at, Some(record.started_at));
assert_eq!(record.exec_established_at, Some(record.started_at));
terminate_process_session_at(root, &identity, &first.process_id, None)
.expect("terminate replay fixture");
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_target_exec_failure_is_known_terminal_record() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "target-exec-failure-project", "Target Exec Failure")
.expect("initialize project");
let mut spec = resolve_project_command_spec_at(
root,
"bash",
&["-lc".to_string(), "exit 0".to_string()],
".",
30,
)
.expect("resolve command");
let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch");
spec.executable = PathBuf::from("/definitely-missing-genarrative-target");
let identity = process_identity("target-exec-failure-project");
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let committed = std::sync::atomic::AtomicBool::new(false);
let result =
start_prepared_process_session_at(root, identity, &spec, &launch, fingerprint, || {
committed.store(true, std::sync::atomic::Ordering::SeqCst);
Ok(())
})
.expect("target exec failure is a known result");
assert!(committed.load(std::sync::atomic::Ordering::SeqCst));
assert_eq!(result.status, "failed");
assert_eq!(result.sandbox_establishment, "established");
assert_eq!(result.target_exec, "failed");
assert_eq!(
result.launch_failure_kind.as_deref(),
Some("target-exec-failed")
);
assert!(!result.needs_reconciliation);
assert_eq!(result.cursor, result.next_cursor);
assert!(!has_active_process_sessions_at(root).expect("known target failure is terminal"));
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_start_audit_failure_terminates_and_persists_reconciliation() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "start-audit-project", "Start Audit Project")
.expect("initialize project");
let spec = resolve_project_command_spec_at(
root,
"bash",
&[
"-lc".to_string(),
"printf 'READY\\n'; while :; do sleep 1; done".to_string(),
],
".",
30,
)
.expect("resolve command");
let identity = process_identity("start-audit-project");
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let started = start_process_session_at(root, identity, &spec, fingerprint).expect("start");
mark_process_session_start_audit_failure_at(
root,
&started.process_id,
"forced agent db failure",
)
.expect("mark start audit reconciliation");
let mut record = None;
for _ in 0..30 {
let current = read_process_session_record(root, &started.process_id)
.expect("read audit failure record")
.expect("audit failure record exists");
if current.status == "needs-reconciliation"
&& current.launch_failure_kind.as_deref() == Some("start-audit-failed")
{
record = Some(current);
break;
}
thread::sleep(Duration::from_millis(50));
}
let record = record.expect("audit failure reconciliation persisted");
assert!(record.needs_reconciliation);
assert!(!record.stdin_open);
assert_eq!(record.target_exec, "established");
assert!(active_process_session_records_at(root, None, None)
.expect("audit failure blocks completion")
.iter()
.any(|value| value.process_id == started.process_id));
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_descendants_inherit_workspace_sandbox() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path().join("workspace");
let outside = directory.path().join("outside-secret.txt");
init_local_game_project_at(&root, "process-sandbox-project", "Process Sandbox Project")
.expect("initialize project");
fs::write(&outside, "OUTSIDE_SECRET").expect("write outside secret");
fs::create_dir_all(root.join(".git")).expect("create git control directory");
fs::write(root.join(".git/marker"), "git").expect("write git marker");
let outside_literal = outside.to_string_lossy().replace('"', "\\\"");
let script = format!(
r#"set -u
if cat "{outside_literal}" >/dev/null 2>&1; then echo OUTSIDE_VISIBLE; else echo OUTSIDE_BLOCKED; fi
if (printf no > .git/blocked-write) 2>/dev/null; then echo GIT_WRITABLE; else echo GIT_BLOCKED; fi
if cat .agent/manifest.json >/dev/null 2>&1; then echo AGENT_VISIBLE; else echo AGENT_HIDDEN; fi
/usr/bin/setsid /bin/bash -lc 'cd /tmp; if test -e "{outside_literal}"; then echo DESCENDANT_VISIBLE; else echo DESCENDANT_BLOCKED; fi'
/usr/bin/python3 - <<'PY'
import socket
s = socket.socket()
s.settimeout(0.2)
try:
s.connect(("1.1.1.1", 53))
print("NETWORK_VISIBLE")
except OSError:
print("NETWORK_BLOCKED")
finally:
s.close()
PY
"#
);
fs::write(root.join("sandbox-probe.sh"), script).expect("write sandbox probe");
let spec = resolve_project_command_spec_at(
&root,
"bash",
&["sandbox-probe.sh".to_string()],
".",
30,
)
.expect("resolve sandbox probe");
let identity = process_identity("process-sandbox-project");
let fingerprint = project_command_source_fingerprint(&root).expect("source fingerprint");
let mut poll = start_process_session_at(&root, identity.clone(), &spec, fingerprint)
.expect("start sandbox probe");
let mut output = poll.output.clone();
for _ in 0..30 {
if poll.status != "running" && !poll.has_more {
break;
}
poll = poll_process_session_at(
&root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(250),
)
.expect("poll sandbox probe");
output.push_str(&poll.output);
}
assert_eq!(poll.status, "exited", "{output}");
for marker in [
"OUTSIDE_BLOCKED",
"GIT_BLOCKED",
"AGENT_HIDDEN",
"DESCENDANT_BLOCKED",
"NETWORK_BLOCKED",
] {
assert!(output.contains(marker), "missing {marker}: {output}");
}
for marker in [
"OUTSIDE_VISIBLE",
"GIT_WRITABLE",
"AGENT_VISIBLE",
"DESCENDANT_VISIBLE",
"NETWORK_VISIBLE",
] {
assert!(!output.contains(marker), "unexpected {marker}: {output}");
}
assert_eq!(poll.sandbox_backend, "bubblewrap");
assert_eq!(poll.sandbox_mode, "workspace-write");
assert_eq!(poll.network_access, "disabled");
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_runner_shutdown_reaps_active_session() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "process-shutdown-project", "Process Shutdown Project")
.expect("initialize project");
fs::write(
root.join("package.json"),
r#"{"scripts":{"dev":"node fixture.js"}}"#,
)
.expect("write package.json");
fs::write(
root.join("fixture.js"),
r#"
console.log('READY');
setInterval(() => {}, 1000);
"#,
)
.expect("write fixture");
let spec = resolve_project_command_spec_at(
root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
30,
)
.expect("resolve npm command");
let identity = process_identity("process-shutdown-project");
let source_fingerprint =
project_command_source_fingerprint(root).expect("source fingerprint");
let started = start_process_session_at(root, identity.clone(), &spec, source_fingerprint)
.expect("start process session");
assert!(has_active_process_sessions_at(root).expect("active process probe"));
shutdown_all_process_sessions_and_wait(Duration::from_secs(3))
.expect("shutdown active process sessions");
let terminal = poll_process_session_at(
root,
&identity,
&started.process_id,
None,
Some(8_000),
Some(0),
)
.expect("poll shutdown terminal state");
assert_eq!(terminal.status, "terminated");
assert_eq!(terminal.signal.as_deref(), Some("runner-shutdown"));
assert!(live_process_session(&started.process_id)
.expect("inspect terminal registry")
.is_none());
assert!(!has_active_process_sessions_at(root).expect("terminal process probe"));
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_terminal_reconciliation_still_blocks_completion() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "process-reconciliation-project", "Process Project")
.expect("initialize project");
fs::write(
root.join("package.json"),
r#"{"scripts":{"dev":"node fixture.js"}}"#,
)
.expect("write package.json");
fs::write(root.join("fixture.js"), "setInterval(() => {}, 1000);\n")
.expect("write fixture");
let process_id = "proc-0123456789abcdef0123456789abcdef";
let now = unix_timestamp();
let mut record = ProcessSessionRecord {
schema_version: PROCESS_SESSION_SCHEMA_VERSION.to_string(),
project_id: "process-reconciliation-project".to_string(),
agent_id: "code-prototype".to_string(),
task_id: "code-prototype".to_string(),
conversation_session_id: "session-process-test".to_string(),
run_id: "run-process-test".to_string(),
start_action_id: "action-process-start-test".to_string(),
start_action_fingerprint: "a".repeat(64),
process_id: process_id.to_string(),
owner_boot_id: process_session_boot_id().to_string(),
command_id: "cmd-process-test".to_string(),
program: "npm".to_string(),
cwd: ".".to_string(),
sandbox_backend: "test-unknown".to_string(),
sandbox_mode: "unknown".to_string(),
network_access: "unknown".to_string(),
sandbox_profile_version: "test-v1".to_string(),
sandbox_establishment: "unknown".to_string(),
target_exec: "unknown".to_string(),
launch_failure_kind: Some("launch-unknown".to_string()),
sandbox_ready_at: None,
exec_established_at: None,
status: "needs-reconciliation".to_string(),
exit_code: None,
signal: Some("output-read-failed".to_string()),
stdin_open: false,
output_bytes: 0,
output_sha256: format!("{:x}", Sha256::digest([])),
output_ref: None,
source_fingerprint_before: "b".repeat(64),
source_fingerprint_after: None,
source_changed: None,
needs_reconciliation: true,
started_at: now,
terminal_at: Some(now),
updated_at: now,
};
write_process_session_record(root, &record).expect("write reconciliation record");
let active = active_process_session_records_at(
root,
Some("code-prototype"),
Some("run-process-test"),
)
.expect("read reconciliation blockers");
assert_eq!(active.len(), 1);
assert_eq!(active[0].process_id, process_id);
let spec = resolve_project_command_spec_at(
root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
30,
)
.expect("resolve blocked command");
let mut blocked_identity = process_identity("process-reconciliation-project");
blocked_identity.run_id = "run-process-blocked-test".to_string();
blocked_identity.start_action_id = "action-process-blocked-start".to_string();
let blocked = validate_process_session_start_preflight_at(root, &blocked_identity, &spec)
.expect_err("reconciliation must block a new process session");
assert!(blocked.contains(process_id));
record.needs_reconciliation = false;
write_process_session_record(root, &record).expect("write invalid reconciliation record");
assert!(active_process_session_records_at(
root,
Some("code-prototype"),
Some("run-process-test")
)
.expect_err("invalid reconciliation record must fail closed")
.contains("可信 launch 状态组合无效"));
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_capacity_preflight_counts_durable_records() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "process-capacity-project", "Process Capacity Project")
.expect("initialize project");
fs::write(
root.join("package.json"),
r#"{"scripts":{"dev":"node fixture.js"}}"#,
)
.expect("write package.json");
fs::write(root.join("fixture.js"), "setInterval(() => {}, 1000);\n")
.expect("write fixture");
let spec = resolve_project_command_spec_at(
root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
30,
)
.expect("resolve command");
for (index, process_id) in [
"proc-11111111111111111111111111111111",
"proc-22222222222222222222222222222222",
]
.into_iter()
.enumerate()
{
let mut identity = process_identity("process-capacity-project");
identity.run_id = format!("run-process-capacity-{index}");
identity.start_action_id = format!("action-process-capacity-{index}");
identity.start_action_fingerprint = format!("{}", index + 1).repeat(64);
let mut record = initial_process_session_record(
&identity,
process_id,
&format!("cmd-process-capacity-{index}"),
&spec,
None,
&"f".repeat(64),
"running",
);
record.sandbox_establishment = "established".to_string();
record.target_exec = "established".to_string();
record.sandbox_ready_at = Some(record.started_at);
record.exec_established_at = Some(record.started_at);
write_process_session_record(root, &record).expect("write durable running record");
}
let mut blocked_identity = process_identity("process-capacity-project");
blocked_identity.run_id = "run-process-capacity-blocked".to_string();
blocked_identity.start_action_id = "action-process-capacity-blocked".to_string();
let error = validate_process_session_start_preflight_at(root, &blocked_identity, &spec)
.expect_err("durable records must count toward Agent capacity");
assert!(error.contains("最多同时运行 2 个"), "{error}");
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_real_pty_eof_reaches_terminal() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "process-eof-project", "Process EOF Project")
.expect("initialize project");
fs::write(
root.join("package.json"),
r#"{"scripts":{"dev":"node fixture.js"}}"#,
)
.expect("write package.json");
fs::write(
root.join("fixture.js"),
r#"
process.stdin.setEncoding('utf8');
console.log('READY');
process.stdin.on('end', () => { console.log('EOF'); process.exit(0); });
process.stdin.resume();
"#,
)
.expect("write fixture");
let spec = resolve_project_command_spec_at(
root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
30,
)
.expect("resolve npm command");
let mut identity = process_identity("process-eof-project");
identity.run_id = "run-process-eof-test".to_string();
identity.start_action_id = "action-process-eof-start".to_string();
identity.start_action_fingerprint = "c".repeat(64);
let fingerprint = project_command_source_fingerprint(root).expect("source fingerprint");
let mut poll =
start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start");
for _ in 0..20 {
if poll.output.contains("READY") {
break;
}
poll = poll_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(500),
)
.expect("poll ready");
}
let eof =
write_process_session_stdin_at(root, &identity, &poll.process_id, "", false, true)
.expect("close stdin");
assert!(eof.eof);
assert!(!eof.stdin_open);
let mut tail = String::new();
for _ in 0..20 {
poll = poll_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(500),
)
.expect("poll eof");
tail.push_str(&poll.output);
if poll.status != "running" {
break;
}
}
assert_eq!(poll.status, "exited", "tail: {tail}");
assert!(tail.contains("EOF"), "tail: {tail}");
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_stdin_accepts_trusted_terminal_race_after_successful_eof() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "stdin-race-project", "Stdin Race Project")
.expect("initialize project");
let spec = resolve_project_command_spec_at(
root,
"bash",
&[
"-lc".to_string(),
"printf 'READY\\n'; while :; do sleep 1; done".to_string(),
],
".",
30,
)
.expect("resolve stdin race command");
let identity = process_identity("stdin-race-project");
let fingerprint = project_command_source_fingerprint(root).expect("fingerprint");
let started =
start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start");
let result = write_process_session_stdin_at_with_after_write(
root,
&identity,
&started.process_id,
"",
false,
true,
|live| {
let mut output = live.output.lock().expect("lock terminal race output");
output.status = "exited".to_string();
output.exit_code = Some(0);
output.stdin_open = false;
},
)
.expect("successful EOF remains successful after trusted terminal wins race");
assert!(result.eof);
assert!(!result.stdin_open);
let record = read_process_session_record(root, &started.process_id)
.expect("read terminal race record")
.expect("terminal race record exists");
assert_eq!(record.status, "exited");
assert!(!record.needs_reconciliation);
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_overlong_unterminated_line_is_stopped() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "process-output-project", "Process Output Project")
.expect("initialize project");
fs::write(
root.join("package.json"),
r#"{"scripts":{"dev":"node fixture.js"}}"#,
)
.expect("write package.json");
fs::write(
root.join("fixture.js"),
"process.stdout.write('x'.repeat(20000)); setInterval(() => {}, 1000);\n",
)
.expect("write fixture");
let spec = resolve_project_command_spec_at(
root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
30,
)
.expect("resolve npm command");
let mut identity = process_identity("process-output-project");
identity.run_id = "run-process-output-test".to_string();
identity.start_action_id = "action-process-output-start".to_string();
identity.start_action_fingerprint = "d".repeat(64);
let fingerprint = project_command_source_fingerprint(root).expect("source fingerprint");
let mut poll =
start_process_session_at(root, identity.clone(), &spec, fingerprint).expect("start");
for _ in 0..30 {
if poll.status != "running" {
break;
}
poll = poll_process_session_at(
root,
&identity,
&poll.process_id,
Some(&poll.next_cursor),
Some(8_000),
Some(250),
)
.expect("poll output limit");
}
assert_eq!(poll.status, "output-limit-exceeded");
clear_process_session_registry_for_tests();
}
#[test]
fn process_session_old_boot_becomes_reconciliation_without_relaunch() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "stale-process-project", "Stale Process Project")
.expect("initialize project");
let identity = process_identity("stale-process-project");
let process_id = "proc-fedcba9876543210fedcba9876543210";
let mut record = ProcessSessionRecord {
schema_version: PROCESS_SESSION_SCHEMA_VERSION.to_string(),
project_id: identity.project_id.clone(),
agent_id: identity.agent_id.clone(),
task_id: identity.task_id.clone(),
conversation_session_id: identity.conversation_session_id.clone(),
run_id: identity.run_id.clone(),
start_action_id: identity.start_action_id.clone(),
start_action_fingerprint: identity.start_action_fingerprint.clone(),
process_id: process_id.to_string(),
owner_boot_id: "old-runner-boot".to_string(),
command_id: "cmd-stale".to_string(),
program: "npm".to_string(),
cwd: ".".to_string(),
sandbox_backend: "bubblewrap".to_string(),
sandbox_mode: "workspace-write".to_string(),
network_access: "disabled".to_string(),
sandbox_profile_version: "workspace-v1".to_string(),
sandbox_establishment: "established".to_string(),
target_exec: "established".to_string(),
launch_failure_kind: None,
sandbox_ready_at: Some(unix_timestamp()),
exec_established_at: Some(unix_timestamp()),
status: "running".to_string(),
exit_code: None,
signal: None,
stdin_open: true,
output_bytes: 0,
output_sha256: format!("{:x}", Sha256::digest([])),
output_ref: None,
source_fingerprint_before: "b".repeat(64),
source_fingerprint_after: None,
source_changed: None,
needs_reconciliation: false,
started_at: unix_timestamp(),
terminal_at: None,
updated_at: unix_timestamp(),
};
write_process_session_record(root, &record).expect("write stale record");
let poll = poll_process_session_at(root, &identity, process_id, None, Some(10), Some(0))
.expect("reconcile stale record");
assert_eq!(poll.status, "needs-reconciliation");
assert!(poll.needs_reconciliation);
record = read_process_session_record(root, process_id)
.expect("read reconciled record")
.expect("record exists");
assert_eq!(record.status, "needs-reconciliation");
assert!(record.needs_reconciliation);
let launching_process_id = "proc-abcdefabcdefabcdefabcdefabcdefab";
let spec = resolve_project_command_spec_at(
root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
30,
)
.expect("resolve stale launching command");
let mut launching = initial_process_session_record(
&identity,
launching_process_id,
"cmd-stale-launching",
&spec,
None,
&"d".repeat(64),
"launching",
);
launching.owner_boot_id = "old-launching-boot".to_string();
launching.sandbox_establishment = "established".to_string();
launching.sandbox_ready_at = Some(launching.started_at);
write_process_session_record(root, &launching).expect("write stale launching record");
let launching_poll = poll_process_session_at(
root,
&identity,
launching_process_id,
None,
Some(10),
Some(0),
)
.expect("reconcile stale launching record");
assert_eq!(launching_poll.status, "needs-reconciliation");
assert_eq!(launching_poll.target_exec, "unknown");
assert_eq!(
launching_poll.launch_failure_kind.as_deref(),
Some("launch-unknown")
);
assert!(launching_poll.needs_reconciliation);
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_start_replay_reconciles_old_launching_record_once() {
let _guard = process_session_test_guard();
clear_process_session_registry_for_tests();
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "stale-replay-project", "Stale Replay Project")
.expect("initialize project");
let identity = process_identity("stale-replay-project");
let spec = resolve_project_command_spec_at(
root,
"bash",
&["-lc".to_string(), "exit 0".to_string()],
".",
30,
)
.expect("resolve stale replay command");
let launch = prepare_project_command_launch_spec(root, &spec).expect("prepare launch");
let process_id = process_session_id(&identity);
let mut record = initial_process_session_record(
&identity,
&process_id,
"cmd-stale-replay",
&spec,
Some(&launch),
&"e".repeat(64),
"launching",
);
record.owner_boot_id = "old-replay-boot".to_string();
record.sandbox_establishment = "established".to_string();
record.sandbox_ready_at = Some(record.started_at);
write_process_session_record(root, &record).expect("write stale replay record");
let callback_called = std::sync::atomic::AtomicBool::new(false);
let error = start_prepared_process_session_at(
root,
identity,
&spec,
&launch,
"e".repeat(64),
|| {
callback_called.store(true, std::sync::atomic::Ordering::SeqCst);
Ok(())
},
)
.expect_err("old launching replay must reconcile without relaunch");
assert_eq!(error.stage(), ProjectCommandErrorStage::LaunchUnknown);
assert!(!callback_called.load(std::sync::atomic::Ordering::SeqCst));
let reconciled = read_process_session_record(root, &process_id)
.expect("read replay reconciliation")
.expect("replay record exists");
assert_eq!(reconciled.status, "needs-reconciliation");
assert_eq!(reconciled.target_exec, "unknown");
assert!(reconciled.exec_established_at.is_none());
assert_eq!(
reconciled.launch_failure_kind.as_deref(),
Some("launch-unknown")
);
assert!(reconciled.needs_reconciliation);
clear_process_session_registry_for_tests();
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_child_wrapper_fixture() {
if std::env::var_os(PROCESS_SESSION_BRIDGE_ENDPOINT_ENV).is_none() {
return;
}
let args = vec![PROCESS_SESSION_CHILD_MODE.to_string()];
match run_process_session_child(&args) {
Ok(exit_code) => std::process::exit(exit_code),
Err(error) => panic!("process session child wrapper failed: {error}"),
}
}
#[test]
fn process_session_runner_owner_fixture() {
let Some(root) = std::env::var_os("GENARRATIVE_PROCESS_SESSION_OWNER_FIXTURE_ROOT") else {
return;
};
let root = PathBuf::from(root);
let spec = resolve_project_command_spec_at(
&root,
"npm",
&["run".to_string(), "dev".to_string()],
".",
300,
)
.expect("resolve owner fixture command");
let identity = process_identity("owner-process-project");
let source_fingerprint =
project_command_source_fingerprint(&root).expect("owner fixture fingerprint");
let poll = start_process_session_at(&root, identity, &spec, source_fingerprint)
.expect("start owner fixture process");
fs::write(root.join("owner-ready"), poll.process_id).expect("write owner ready");
loop {
thread::sleep(Duration::from_secs(1));
}
}
#[cfg(target_os = "linux")]
#[test]
fn process_session_owner_sigkill_leaves_no_child_process() {
fn project_processes(root: &Path) -> Vec<i32> {
let canonical_root = fs::canonicalize(root).expect("canonical test project");
fs::read_dir("/proc")
.into_iter()
.flatten()
.flatten()
.filter_map(|entry| {
let process_id = entry.file_name().to_string_lossy().parse::<i32>().ok()?;
let cwd = fs::read_link(entry.path().join("cwd")).ok()?;
(cwd == canonical_root).then_some(process_id)
})
.collect()
}
let directory = tempfile::tempdir().expect("temp project");
let root = directory.path();
init_local_game_project_at(root, "owner-process-project", "Owner Process Project")
.expect("initialize project");
fs::write(
root.join("package.json"),
r#"{"scripts":{"dev":"node owner-fixture.js"}}"#,
)
.expect("write package.json");
fs::write(
root.join("owner-fixture.js"),
r#"
process.on('SIGHUP', () => {});
require('fs').writeFileSync('child.pid', String(process.pid));
setInterval(() => {}, 1000);
"#,
)
.expect("write fixture");
let current_exe = std::env::current_exe().expect("current test binary");
let mut owner = std::process::Command::new(current_exe)
.arg("--exact")
.arg("process_session::tests::process_session_runner_owner_fixture")
.arg("--nocapture")
.env("GENARRATIVE_PROCESS_SESSION_OWNER_FIXTURE_ROOT", root)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.expect("spawn owner fixture test process");
let deadline = std::time::Instant::now() + Duration::from_secs(10);
while (!root.join("owner-ready").is_file() || project_processes(root).is_empty())
&& std::time::Instant::now() < deadline
{
thread::sleep(Duration::from_millis(25));
}
let sandbox_processes = project_processes(root);
assert!(
!sandbox_processes.is_empty(),
"sandbox child should be visible from host /proc"
);
let owner_pid = i32::try_from(owner.id()).expect("owner pid");
assert_eq!(unsafe { libc::kill(owner_pid, libc::SIGKILL) }, 0);
owner.wait().expect("reap owner fixture");
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let remaining = project_processes(root);
if remaining.is_empty() {
break;
}
assert!(
std::time::Instant::now() < deadline,
"Runner owner SIGKILL 后 sandbox 子进程仍存在:pids={remaining:?}"
);
thread::sleep(Duration::from_millis(25));
}
}
}