diff --git a/.github/workflows/spike-windows-acp-native-agent.yml b/.github/workflows/spike-windows-acp-native-agent.yml new file mode 100644 index 000000000..6ed9251ee --- /dev/null +++ b/.github/workflows/spike-windows-acp-native-agent.yml @@ -0,0 +1,252 @@ +name: Windows shell and process-tree gate + +on: + push: + branches: + - spike/windows-acp-native-agent + - main + paths: + - "crates/openab-core/src/acp/**" + - "crates/openab-mcp/src/auth.rs" + - "crates/openab-mcp/Cargo.toml" + - "openab-agent/src/tools.rs" + - "openab-agent/Cargo.toml" + - "scripts/windows-acp-native-canary.ps1" + - "docs/windows-acp-native-canary.md" + - ".github/workflows/spike-windows-acp-native-agent.yml" + pull_request: + paths: + - "crates/openab-core/src/acp/**" + - "crates/openab-mcp/src/auth.rs" + - "crates/openab-mcp/Cargo.toml" + - "openab-agent/src/tools.rs" + - "openab-agent/Cargo.toml" + - "scripts/windows-acp-native-canary.ps1" + - "docs/windows-acp-native-canary.md" + - ".github/workflows/spike-windows-acp-native-agent.yml" + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: windows-native-lifecycle-${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: false + +env: + CARGO_TERM_COLOR: always + TARGET: x86_64-pc-windows-msvc + UPSTREAM_PIN: 280db4db9452e7e108b52720fddf18dd8c3f5181 + +jobs: + test-auth-windows: + runs-on: windows-latest + timeout-minutes: 40 + steps: + - name: Checkout production auth source + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + + - name: Record toolchain + shell: pwsh + run: | + rustc --version --verbose + cargo --version --verbose + rustup target add $env:TARGET + + - name: Test Windows auth lock and atomic replace + shell: pwsh + run: | + cargo test --locked --target $env:TARGET -p openab-mcp auth::tests:: -- --nocapture + + test-agent-tools-windows: + runs-on: windows-latest + timeout-minutes: 40 + steps: + - name: Checkout shell and process-tree source + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + + - name: Record toolchain + shell: pwsh + run: | + rustc --version --verbose + cargo --version --verbose + rustup target add $env:TARGET + + - name: Test Windows shell and Job Object cleanup + shell: pwsh + run: | + cargo test --locked --target $env:TARGET --manifest-path openab-agent/Cargo.toml tools::tests:: -- --nocapture --test-threads=1 + + build-openab: + runs-on: windows-latest + timeout-minutes: 40 + steps: + - name: Checkout production auth source + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + + - name: Record toolchain + shell: pwsh + run: | + rustc --version --verbose + cargo --version --verbose + rustup target add $env:TARGET + + - name: Build ACP-enabled OpenAB + shell: pwsh + run: | + cargo build --locked --release --target $env:TARGET --no-default-features --features acp + + - name: Upload ACP-enabled OpenAB binary + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: windows-openab-${{ github.sha }} + path: target/${{ env.TARGET }}/release/openab.exe + if-no-files-found: error + retention-days: 1 + compression-level: 0 + + build-openab-agent: + runs-on: windows-latest + timeout-minutes: 40 + steps: + - name: Checkout production auth source + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + + - name: Record toolchain + shell: pwsh + run: | + rustc --version --verbose + cargo --version --verbose + rustup target add $env:TARGET + + - name: Build repo-native OpenAB agent + shell: pwsh + run: | + cargo build --locked --release --target $env:TARGET --manifest-path openab-agent/Cargo.toml + + - name: Upload repo-native OpenAB agent binary + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: windows-openab-agent-${{ github.sha }} + path: openab-agent/target/${{ env.TARGET }}/release/openab-agent.exe + if-no-files-found: error + retention-days: 1 + compression-level: 0 + + windows-dual-layer-canary: + needs: + - test-auth-windows + - test-agent-tools-windows + - build-openab + - build-openab-agent + runs-on: windows-latest + timeout-minutes: 50 + steps: + - name: Checkout canary source + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + + - name: Record toolchain + shell: pwsh + run: | + rustc --version --verbose + cargo --version --verbose + rustup target add $env:TARGET + + - name: Test controller-owned Job Object cleanup while connection is locked + shell: pwsh + run: | + cargo test --locked --target $env:TARGET -p openab-core --no-default-features controller_guard_terminates_windows_job_descendants_while_connection_is_locked -- --nocapture + + - name: Download ACP-enabled OpenAB binary + uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0 + with: + name: windows-openab-${{ github.sha }} + path: dist/binaries + + - name: Download repo-native OpenAB agent binary + uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0 + with: + name: windows-openab-agent-${{ github.sha }} + path: dist/binaries + + - name: Run no-provider-call dual-layer runtime canary + shell: pwsh + run: | + ./scripts/windows-acp-native-canary.ps1 ` + -OpenAbExe "dist/binaries/openab.exe" ` + -AgentExe "dist/binaries/openab-agent.exe" ` + -WorkDir $PWD + + # This step is deliberately after every runtime assertion. If any canary + # fails, the job never creates an uploadable archive and artifact count + # remains zero. + - name: Package bounded non-release canary + id: package + shell: pwsh + run: | + $short = $env:GITHUB_SHA.Substring(0, 7) + $name = "openab-windows-canary-$short" + $stage = Join-Path $PWD "dist/$name" + New-Item -ItemType Directory -Path $stage -Force | Out-Null + Copy-Item "dist/binaries/openab.exe" $stage + Copy-Item "dist/binaries/openab-agent.exe" $stage + Copy-Item "scripts/windows-acp-native-canary.ps1" $stage + Copy-Item "docs/windows-acp-native-canary.md" (Join-Path $stage "README.md") + + $records = @() + foreach ($file in Get-ChildItem -LiteralPath $stage -File | Sort-Object Name) { + $hash = (Get-FileHash -LiteralPath $file.FullName -Algorithm SHA256).Hash.ToLowerInvariant() + $records += [ordered]@{ + name = $file.Name + bytes = $file.Length + sha256 = $hash + } + } + $repository = $env:GITHUB_REPOSITORY + $branch = if ($env:GITHUB_HEAD_REF) { $env:GITHUB_HEAD_REF } else { $env:GITHUB_REF_NAME } + [ordered]@{ + schema = "openab.windows-canary.v1" + repository = $repository + branch = $branch + commit = $env:GITHUB_SHA + upstream_pin = $env:UPSTREAM_PIN + target = $env:TARGET + release = $false + provider_credentials = $false + files = $records + } | ConvertTo-Json -Depth 5 | Set-Content -LiteralPath (Join-Path $stage "MANIFEST.json") -Encoding utf8NoBOM + + Get-ChildItem -LiteralPath $stage -File | Sort-Object Name | ForEach-Object { + $hash = (Get-FileHash -LiteralPath $_.FullName -Algorithm SHA256).Hash.ToLowerInvariant() + "$hash $($_.Name)" + } | Set-Content -LiteralPath (Join-Path $stage "SHA256SUMS.txt") -Encoding ascii + + $archive = Join-Path $PWD "dist/$name.zip" + Compress-Archive -Path (Join-Path $stage "*") -DestinationPath $archive -CompressionLevel Optimal + $archiveHash = (Get-FileHash -LiteralPath $archive -Algorithm SHA256).Hash.ToLowerInvariant() + $archiveBytes = (Get-Item -LiteralPath $archive).Length + "archive=$archive" | Out-File -FilePath $env:GITHUB_OUTPUT -Append -Encoding utf8 + "name=$name" | Out-File -FilePath $env:GITHUB_OUTPUT -Append -Encoding utf8 + "### Non-release Windows canary" | Out-File -FilePath $env:GITHUB_STEP_SUMMARY -Append -Encoding utf8 + "- SHA-256: ``$archiveHash``" | Out-File -FilePath $env:GITHUB_STEP_SUMMARY -Append -Encoding utf8 + "- Bytes: ``$archiveBytes``" | Out-File -FilePath $env:GITHUB_STEP_SUMMARY -Append -Encoding utf8 + "- Commit: ``$env:GITHUB_SHA``" | Out-File -FilePath $env:GITHUB_STEP_SUMMARY -Append -Encoding utf8 + + - name: Upload non-release canary artifact + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: ${{ steps.package.outputs.name }} + path: ${{ steps.package.outputs.archive }} + if-no-files-found: error + retention-days: 3 + compression-level: 0 diff --git a/Cargo.lock b/Cargo.lock index fc11ec7e5..53457bebc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2546,6 +2546,7 @@ dependencies = [ "http 1.4.2", "image", "libc", + "process-wrap", "pulldown-cmark", "rand 0.8.6", "regex", @@ -2654,6 +2655,7 @@ dependencies = [ "tracing", "url", "urlencoding", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/openab-core/Cargo.toml b/crates/openab-core/Cargo.toml index 1ede398f1..eae0d30b8 100644 --- a/crates/openab-core/Cargo.toml +++ b/crates/openab-core/Cargo.toml @@ -54,6 +54,11 @@ http = { version = "1", optional = true } # axum listener) was removed with the per-session proxy. rmcp = { version = "1.7", default-features = false, optional = true } tokio-util = { version = "0.7", optional = true } +process-wrap = { version = "9.1.0", default-features = false, features = [ + "tokio1", + "kill-on-drop", + "job-object", +] } [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/crates/openab-core/src/acp/connection.rs b/crates/openab-core/src/acp/connection.rs index 5f5d83747..9f37f8be4 100644 --- a/crates/openab-core/src/acp/connection.rs +++ b/crates/openab-core/src/acp/connection.rs @@ -3,6 +3,8 @@ use crate::acp::protocol::{ JsonRpcResponse, UsageReport, }; use anyhow::{anyhow, Result}; +#[cfg(not(windows))] +use process_wrap::tokio::ChildWrapper as _; use serde_json::{json, Value}; use std::collections::HashMap; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; @@ -13,6 +15,10 @@ use tokio::sync::{mpsc, oneshot, Mutex}; use tokio::task::JoinHandle; use tracing::{debug, error, info, trace}; +pub use crate::acp::process_tree::ProcessTreeGuard; +#[cfg(windows)] +use process_wrap::tokio::{CommandWrap, JobObject, KillOnDrop}; + /// Pick the most permissive selectable permission option from ACP options. fn pick_best_option(options: &[Value]) -> Option { let mut fallback: Option<&Value> = None; @@ -171,7 +177,13 @@ impl SessionActivity { } pub struct AcpConnection { + #[cfg(not(windows))] _proc: Child, + /// Cloneable controller handle stored outside the connection mutex by the + /// pool. The background owner holds the Job Object child and executes + /// explicit termination requests; `KILL_ON_JOB_CLOSE` is only a fallback. + #[cfg(windows)] + process_tree_guard: ProcessTreeGuard, /// PID of the direct child, used as the process group ID for cleanup. child_pgid: Option, stdin: Arc>, @@ -361,10 +373,6 @@ impl AcpConnection { Ok(()) }); } - #[cfg(windows)] - { - cmd.creation_flags(0x00000200); // CREATE_NEW_PROCESS_GROUP - } // Clear inherited env to prevent credential leakage (e.g. DISCORD_BOT_TOKEN). // Only [agent].env values + essential baseline vars are passed through. cmd.env_clear(); @@ -401,8 +409,16 @@ impl AcpConnection { if let Ok(v) = std::env::var("SystemRoot") { cmd.env("SystemRoot", v); } - if let Ok(v) = std::env::var("SystemDrive") { - cmd.env("SystemDrive", v); + // PowerShell fixtures and normal Windows CLI tools need the standard + // runtime locations after env_clear(). Per-user application-data + // paths are intentionally excluded: npmrc/pip.ini and similar files + // may contain credentials and must not become an implicit agent read + // surface after prompt injection. The key list is shared with + // `openab-agent` via `crate::acp::WINDOWS_RUNTIME_ENV_KEYS`. + for key in crate::acp::WINDOWS_RUNTIME_ENV_KEYS { + if let Ok(v) = std::env::var(key) { + cmd.env(key, v); + } } } for (k, v) in env { @@ -423,18 +439,27 @@ impl AcpConnection { "[agent].env/inherit_env is set -- these values are accessible to the agent and could be exfiltrated via prompt injection" ); } + #[cfg(windows)] + let mut proc = { + let mut wrapped = CommandWrap::from(cmd); + wrapped.wrap(KillOnDrop).wrap(JobObject); + wrapped + .spawn() + .map_err(|e| anyhow!("failed to spawn {command} in Windows Job Object: {e}"))? + }; + #[cfg(not(windows))] let mut proc = cmd .spawn() .map_err(|e| anyhow!("failed to spawn {command}: {e}"))?; let child_pgid = proc.id().and_then(|pid| i32::try_from(pid).ok()); - let stdout = proc.stdout.take().ok_or_else(|| anyhow!("no stdout"))?; - let stdin = proc.stdin.take().ok_or_else(|| anyhow!("no stdin"))?; + let stdout = proc.stdout().take().ok_or_else(|| anyhow!("no stdout"))?; + let stdin = proc.stdin().take().ok_or_else(|| anyhow!("no stdin"))?; let stdin = Arc::new(Mutex::new(stdin)); // Capture agent stderr and log it (ACP spec: agents MAY write to stderr // for logging; clients MAY capture or ignore it). - let stderr_handle = if let Some(stderr) = proc.stderr.take() { + let stderr_handle = if let Some(stderr) = proc.stderr().take() { let cmd_name = command.to_string(); Some(tokio::spawn(async move { let mut reader = BufReader::new(stderr); @@ -463,6 +488,9 @@ impl AcpConnection { None }; + #[cfg(windows)] + let process_tree_guard = ProcessTreeGuard::new(proc); + let pending: Arc>>> = Arc::new(Mutex::new(HashMap::new())); let notify_tx: Arc>>> = @@ -478,7 +506,10 @@ impl AcpConnection { let activity = Arc::new(SessionActivity::new()); Ok(Self { + #[cfg(not(windows))] _proc: proc, + #[cfg(windows)] + process_tree_guard, child_pgid, stdin, next_id: AtomicU64::new(1), @@ -787,6 +818,17 @@ impl AcpConnection { self.child_pgid } + pub fn process_tree_guard(&self) -> ProcessTreeGuard { + #[cfg(windows)] + { + self.process_tree_guard.clone() + } + #[cfg(not(windows))] + { + ProcessTreeGuard + } + } + pub fn alive(&self) -> bool { !self._reader_handle.is_finished() } @@ -855,6 +897,123 @@ mod tests { use super::{build_agent_env, build_permission_response, pick_best_option}; use serde_json::json; + #[cfg(windows)] + fn powershell_path() -> std::path::PathBuf { + std::path::PathBuf::from(std::env::var_os("SystemRoot").expect("SystemRoot")) + .join("System32") + .join("WindowsPowerShell") + .join("v1.0") + .join("powershell.exe") + } + + #[cfg(windows)] + fn process_exists(powershell: &std::path::Path, pid: u32) -> bool { + std::process::Command::new(powershell) + .args([ + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + &format!( + "if (Get-Process -Id {pid} -ErrorAction SilentlyContinue) {{ exit 0 }} else {{ exit 1 }}" + ), + ]) + .status() + .map(|status| status.success()) + .unwrap_or(false) + } + + #[cfg(windows)] + #[tokio::test] + async fn controller_guard_terminates_windows_job_descendants_while_connection_is_locked() { + use super::AcpConnection; + use std::collections::HashMap; + + let temp = tempfile::tempdir().unwrap(); + let pid_file = temp.path().join("descendant.pid"); + let fixture = temp.path().join("agent-fixture.ps1"); + let pid_literal = pid_file.to_string_lossy().replace('\'', "''"); + std::fs::write( + &fixture, + format!( + r#" +$descendant = Start-Process -FilePath "$env:SystemRoot\System32\ping.exe" -ArgumentList @("127.0.0.1", "-n", "120") -PassThru +Set-Content -LiteralPath '{pid_literal}' -Value $descendant.Id -NoNewline +while (($line = [Console]::In.ReadLine()) -ne $null) {{ + $request = $line | ConvertFrom-Json + if ($request.method -eq "initialize") {{ + @{{jsonrpc="2.0"; id=$request.id; result=@{{protocolVersion=1; agentInfo=@{{name="job-fixture"; version="0"}}; agentCapabilities=@{{loadSession=$false}}}}}} | ConvertTo-Json -Compress -Depth 8 + }} +}} +Wait-Process -Id $descendant.Id +"# + ), + ) + .unwrap(); + + let powershell = powershell_path(); + let args = vec![ + "-NoLogo".to_string(), + "-NoProfile".to_string(), + "-NonInteractive".to_string(), + "-ExecutionPolicy".to_string(), + "Bypass".to_string(), + "-File".to_string(), + fixture.to_string_lossy().into_owned(), + ]; + let powershell_command = powershell.to_string_lossy().into_owned(); + let working_dir = temp.path().to_string_lossy().into_owned(); + let mut conn = AcpConnection::spawn( + &powershell_command, + &args, + &working_dir, + &HashMap::new(), + &[], + ) + .await + .unwrap(); + conn.initialize().await.unwrap(); + let direct_pid = u32::try_from(conn.child_pgid.unwrap()).unwrap(); + + let descendant_pid = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + if let Ok(raw) = std::fs::read_to_string(&pid_file) { + if let Ok(pid) = raw.trim().parse::() { + break pid; + } + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + }) + .await + .expect("fixture did not publish descendant pid"); + assert!(process_exists(&powershell, direct_pid)); + assert!(process_exists(&powershell, descendant_pid)); + + let guard = conn.process_tree_guard(); + let conn = std::sync::Arc::new(tokio::sync::Mutex::new(conn)); + let connection_lock = conn.lock().await; + guard.terminate().await.unwrap(); + + let gone = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if !process_exists(&powershell, direct_pid) + && !process_exists(&powershell, descendant_pid) + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + }) + .await; + assert!(gone.is_ok(), "Windows Job Object left a process behind"); + guard + .terminate() + .await + .expect("second terminate must succeed after the controller already killed the tree"); + drop(connection_lock); + } + #[test] fn picks_allow_always_over_other_options() { let options = vec![ diff --git a/crates/openab-core/src/acp/mod.rs b/crates/openab-core/src/acp/mod.rs index 7919c6d3d..a9cb60e57 100644 --- a/crates/openab-core/src/acp/mod.rs +++ b/crates/openab-core/src/acp/mod.rs @@ -2,8 +2,12 @@ pub mod agentcore; pub mod connection; pub mod pool; +pub mod process_tree; +mod windows_env; pub mod protocol; pub use connection::ContentBlock; pub use pool::SessionPool; pub use protocol::{classify_notification, parse_turn_result, AcpEvent, TurnResult}; + +pub use windows_env::WINDOWS_RUNTIME_ENV_KEYS; diff --git a/crates/openab-core/src/acp/pool.rs b/crates/openab-core/src/acp/pool.rs index 86b2ee989..0ecfa7516 100644 --- a/crates/openab-core/src/acp/pool.rs +++ b/crates/openab-core/src/acp/pool.rs @@ -1,4 +1,4 @@ -use crate::acp::connection::{AcpConnection, SessionActivity}; +use crate::acp::connection::{AcpConnection, ProcessTreeGuard, SessionActivity}; use crate::acp::protocol::ConfigOption; use crate::config::AgentConfig; use anyhow::{anyhow, Result}; @@ -35,6 +35,10 @@ struct PoolState { /// Child process-group ids, captured at insert time so hung eviction can /// kill the agent process without ever locking the connection. pgids: HashMap, + /// Cloneable process-tree controllers, stored outside the connection + /// mutex so Windows shutdown and hung eviction can terminate a Job Object + /// even while a streaming turn owns the connection lock. + process_tree_guards: HashMap, /// Suspended sessions: thread_key → ACP sessionId. /// Used at runtime to decide which thread can be resumed via `session/load` /// because it no longer has a live in-memory connection. @@ -49,6 +53,10 @@ struct PoolState { /// Per-session working directory overrides (from control directives). /// thread_key → canonical workspace path. session_workdirs: HashMap, + /// Once shutdown starts, no new session may be inserted. This closes the + /// snapshot/terminate race where reset_session or get_or_create could + /// replace a guard after shutdown captured the old one. + shutting_down: bool, } pub struct SessionPool { @@ -69,6 +77,7 @@ pub struct SessionPool { type CancelHandle = (Arc>, String); type ActiveSnapshot = Vec<(String, Arc>)>; +type ProcessTreeSnapshot = Vec<(String, ProcessTreeGuard)>; type EvictionCandidate = (String, Arc>, Instant, Option); fn remove_if_same_handle( @@ -177,6 +186,7 @@ fn purge_session_entries(state: &mut PoolState, key: &str) { state.cancel_handles.remove(key); state.activity.remove(key); state.pgids.remove(key); + state.process_tree_guards.remove(key); state.suspended.remove(key); state.persisted.remove(key); // Do NOT remove the creating gate: it is concurrency control, not session @@ -190,13 +200,26 @@ fn purge_session_entries(state: &mut PoolState, key: &str) { /// session/cancel attempt, SIGTERM, wait 2s, SIGKILL. Mirrors /// `AcpConnection::kill_process_group`, which cannot run here because the /// hung task never drops its connection Arc. -async fn kill_pgid_after_grace(pgid: Option) { - let Some(pgid) = pgid.filter(|p| *p > 0) else { - return; - }; +async fn kill_process_tree_after_grace( + pgid: Option, + process_tree_guard: Option, +) { + #[cfg(not(windows))] tokio::time::sleep(std::time::Duration::from_secs(10)).await; + #[cfg(not(windows))] + let _ = &process_tree_guard; + #[cfg(windows)] + if let Some(guard) = process_tree_guard { + if let Err(e) = guard.terminate().await { + warn!(error = %e, "failed to terminate hung Windows agent Job Object"); + } + return; + } #[cfg(unix)] { + let Some(pgid) = pgid.filter(|p| *p > 0) else { + return; + }; unsafe { libc::kill(-pgid, libc::SIGTERM); } @@ -205,11 +228,9 @@ async fn kill_pgid_after_grace(pgid: Option) { libc::kill(-pgid, libc::SIGKILL); } } - #[cfg(not(unix))] + #[cfg(not(any(unix, windows)))] { - // No process-group kill on non-unix; rely on AcpConnection::Drop's - // Windows handling if/when the hung task eventually unwinds. - let _ = pgid; + let _ = (pgid, process_tree_guard); } } @@ -295,10 +316,12 @@ impl SessionPool { facade_tokens: HashMap::new(), activity: HashMap::new(), pgids: HashMap::new(), + process_tree_guards: HashMap::new(), persisted: suspended.clone(), suspended, creating: HashMap::new(), session_workdirs, + shutting_down: false, }), config, max_sessions, @@ -400,6 +423,9 @@ impl SessionPool { ) -> Result { let create_gate = { let mut state = self.state.write().await; + if state.shutting_down { + return Err(anyhow!("session pool is shutting down")); + } get_or_insert_gate(&mut state.creating, thread_id) }; let _create_guard = create_gate.lock().await; @@ -613,6 +639,7 @@ impl SessionPool { let cancel_handle = new_conn.cancel_handle(); let activity_handle = new_conn.activity_handle(); let child_pgid = new_conn.child_pgid(); + let process_tree_guard = new_conn.process_tree_guard(); let cancel_session_id = new_conn.acp_session_id.clone().unwrap_or_default(); #[cfg(feature = "acp-mcp")] new_conn.set_facade_token_guard(facade_token_guard); @@ -620,6 +647,10 @@ impl SessionPool { let mut state = self.state.write().await; + if state.shutting_down { + return Err(anyhow!("session pool is shutting down")); + } + // Another task may have created a healthy connection while we were // initializing this one. if let Some(existing) = state.active.get(thread_id).cloned() { @@ -635,6 +666,7 @@ impl SessionPool { state.cancel_handles.remove(thread_id); state.activity.remove(thread_id); state.pgids.remove(thread_id); + state.process_tree_guards.remove(thread_id); } if state.active.len() >= self.max_sessions { @@ -643,6 +675,7 @@ impl SessionPool { state.cancel_handles.remove(&key); state.activity.remove(&key); state.pgids.remove(&key); + state.process_tree_guards.remove(&key); #[cfg(feature = "acp-mcp")] revoke_facade_token_for_key(&mut state, &key, self.session_registrar.as_ref()); info!(evicted = %crate::redact::redact_session_ids(&key), "pool full, suspending oldest idle session"); @@ -683,6 +716,9 @@ impl SessionPool { if let Some(pgid) = child_pgid { state.pgids.insert(thread_id.to_string(), pgid); } + state + .process_tree_guards + .insert(thread_id.to_string(), process_tree_guard); if !cancel_session_id.is_empty() { state .cancel_handles @@ -815,6 +851,10 @@ impl SessionPool { /// Arc reference is dropped (after streaming finishes). The next message will /// trigger a fresh `get_or_create` with a new ACP session. pub async fn reset_session(&self, thread_id: &str) -> Result<()> { + let process_tree_guard = { + let state = self.state.read().await; + state.process_tree_guards.get(thread_id).cloned() + }; // Send session/cancel via the lock-free stdin handle first. // This stops in-flight streaming even while with_connection() holds the // connection mutex, so the old process finishes promptly. @@ -835,6 +875,15 @@ impl SessionPool { let _ = w.flush().await; } + #[cfg(windows)] + if let Some(guard) = process_tree_guard { + if let Err(e) = guard.terminate().await { + warn!(thread_id = %crate::redact::redact_session_ids(thread_id), error = %e, "reset process-tree shutdown failed"); + } + } + #[cfg(not(windows))] + let _ = process_tree_guard; + let mut state = self.state.write().await; let had_active = state.active.remove(thread_id).is_some(); // Everything else a reset clears is exactly what hung eviction clears, including the rule @@ -860,7 +909,7 @@ impl SessionPool { let cutoff = Instant::now() - std::time::Duration::from_secs(ttl_secs); let hung_threshold = std::time::Duration::from_secs(self.hung_threshold_secs); - let (snapshot, activity_map, cancel_map, pgid_map) = { + let (snapshot, activity_map, cancel_map, pgid_map, process_tree_guard_map) = { let state = self.state.read().await; let snapshot: ActiveSnapshot = state .active @@ -872,6 +921,7 @@ impl SessionPool { state.activity.clone(), state.cancel_handles.clone(), state.pgids.clone(), + state.process_tree_guards.clone(), ) }; @@ -895,10 +945,11 @@ impl SessionPool { // handle, detached so a wedged stdin can never block // cleanup (and never while holding `state`). The hung // task never unwinds, so AcpConnection::Drop never - // fires; after the cancel attempt, kill the child - // process group directly or the agent leaks forever (F4). + // fires; after the cancel attempt, use the lock-free + // process-tree controller or the agent leaks forever (F4). let stdin_handle = cancel_map.get(&key).map(|(stdin, _)| Arc::clone(stdin)); let pgid = pgid_map.get(&key).copied(); + let process_tree_guard = process_tree_guard_map.get(&key).cloned(); tokio::spawn(async move { if let (Some(stdin), Some(session_id)) = (stdin_handle, session_id) { let _ = tokio::time::timeout( @@ -921,7 +972,7 @@ impl SessionPool { ) .await; } - kill_pgid_after_grace(pgid).await; + kill_process_tree_after_grace(pgid, process_tree_guard).await; }); hung.push((key, conn_handle)); } @@ -954,6 +1005,7 @@ impl SessionPool { state.cancel_handles.remove(&key); state.activity.remove(&key); state.pgids.remove(&key); + state.process_tree_guards.remove(&key); #[cfg(feature = "acp-mcp")] revoke_facade_token_for_key(&mut state, &key, self.session_registrar.as_ref()); if let Some(sid) = sid { @@ -986,17 +1038,47 @@ impl SessionPool { // Snapshot active handles, then drop state lock before awaiting // per-connection mutexes (lock ordering: never hold state while // awaiting a connection lock). - let snapshot: Vec<(String, Arc>)> = { - let state = self.state.read().await; - state - .active - .iter() - .map(|(k, v)| (k.clone(), Arc::clone(v))) - .collect() + let (snapshot, process_tree_guards): (ActiveSnapshot, ProcessTreeSnapshot) = { + let mut state = self.state.write().await; + state.shutting_down = true; + ( + state + .active + .iter() + .map(|(k, v)| (k.clone(), Arc::clone(v))) + .collect(), + state + .process_tree_guards + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + ) }; + // Terminate from controller-owned handles before awaiting any + // connection mutex. A streaming turn can hold that mutex indefinitely. + for (key, guard) in process_tree_guards { + if let Err(e) = guard.terminate().await { + warn!(thread_id = %crate::redact::redact_session_ids(&key), error = %e, "agent process-tree shutdown failed"); + } + } + let mut session_ids: Vec<(String, String)> = Vec::new(); for (key, conn) in snapshot { + #[cfg(windows)] + let conn = match tokio::time::timeout( + std::time::Duration::from_secs(10), + conn.lock(), + ) + .await + { + Ok(conn) => conn, + Err(_) => { + warn!(thread_id = %crate::redact::redact_session_ids(&key), "timed out waiting for connection after process-tree shutdown"); + continue; + } + }; + #[cfg(not(windows))] let conn = conn.lock().await; if let Some(sid) = conn.acp_session_id.clone() { session_ids.push((key, sid)); @@ -1014,6 +1096,7 @@ impl SessionPool { state.cancel_handles.clear(); state.activity.clear(); state.pgids.clear(); + state.process_tree_guards.clear(); info!(count, "pool shutdown complete"); } } @@ -1065,10 +1148,12 @@ mod tests { facade_tokens: HashMap::new(), activity: HashMap::new(), pgids: HashMap::new(), + process_tree_guards: HashMap::new(), suspended: HashMap::new(), persisted: HashMap::new(), creating: HashMap::new(), session_workdirs: HashMap::new(), + shutting_down: false, } } @@ -1338,6 +1423,7 @@ mod tests { ("other".to_string(), Arc::new(SessionActivity::new())), ]), pgids: HashMap::from([("hung".to_string(), 1234), ("other".to_string(), 5678)]), + process_tree_guards: HashMap::new(), suspended: HashMap::from([ ("hung".to_string(), "session-hung".to_string()), ("other".to_string(), "session-other".to_string()), @@ -1348,6 +1434,7 @@ mod tests { ]), creating: HashMap::from([("hung".to_string(), Arc::new(Mutex::new(())))]), session_workdirs: HashMap::from([("hung".to_string(), "/tmp/ws".to_string())]), + shutting_down: false, }; purge_session_entries(&mut state, "hung"); diff --git a/crates/openab-core/src/acp/process_tree.rs b/crates/openab-core/src/acp/process_tree.rs new file mode 100644 index 000000000..be31baf78 --- /dev/null +++ b/crates/openab-core/src/acp/process_tree.rs @@ -0,0 +1,97 @@ +//! Platform-specific process-tree lifecycle control for ACP child processes. +//! +//! Keeping the controller here gives the ACP connection and pool one platform-neutral +//! handle. The Windows implementation owns the Job Object and waits asynchronously for +//! normal child exit; it never uses a periodic liveness poll. + +#[cfg(windows)] +use anyhow::anyhow; +use anyhow::Result; + +#[cfg(windows)] +use process_wrap::tokio::ChildWrapper; +#[cfg(windows)] +use tokio::sync::{mpsc, oneshot}; +#[cfg(windows)] +use tracing::error; + +#[cfg(windows)] +type AgentChild = Box; + +#[cfg(windows)] +#[derive(Clone)] +pub struct ProcessTreeGuard { + terminate_tx: mpsc::UnboundedSender>>, +} + +#[cfg(not(windows))] +#[derive(Clone, Default)] +pub struct ProcessTreeGuard; + +#[cfg(not(windows))] +impl ProcessTreeGuard { + pub async fn terminate(&self) -> Result<()> { + Ok(()) + } +} + +#[cfg(windows)] +impl ProcessTreeGuard { + pub(crate) fn new(mut child: AgentChild) -> Self { + let (terminate_tx, mut terminate_rx) = + mpsc::unbounded_channel::>>(); + + tokio::spawn(async move { + // Service every terminate request. The first one kills; later callers + // receive Ok(()) instead of "controller exited before replying". + // Keep wait() scoped to select! so kill() never aliases that borrow. + let mut killed = false; + loop { + if killed { + match terminate_rx.recv().await { + Some(reply) => { + let _ = reply.send(Ok(())); + } + None => return, + } + continue; + } + + let request = tokio::select! { + request = terminate_rx.recv() => request, + result = child.wait() => { + if let Err(e) = result { + error!(error = %e, "failed to wait for Windows agent Job Object"); + } + while let Ok(reply) = terminate_rx.try_recv() { + let _ = reply.send(Ok(())); + } + return; + } + }; + + let Some(reply) = request else { + return; + }; + let result = Box::into_pin(child.kill()).await.map_err(|e| e.to_string()); + killed = true; + let _ = reply.send(result); + } + }); + + Self { terminate_tx } + } + + /// Terminate the Job Object without acquiring the connection mutex. + pub async fn terminate(&self) -> Result<()> { + let (reply_tx, reply_rx) = oneshot::channel(); + if self.terminate_tx.send(reply_tx).is_err() { + return Ok(()); + } + tokio::time::timeout(std::time::Duration::from_secs(10), reply_rx) + .await + .map_err(|_| anyhow!("timeout terminating Windows agent Job Object"))? + .map_err(|_| anyhow!("Windows agent process controller exited before replying"))? + .map_err(|e| anyhow!("failed to terminate Windows agent Job Object: {e}")) + } +} diff --git a/crates/openab-core/src/acp/windows_env.rs b/crates/openab-core/src/acp/windows_env.rs new file mode 100644 index 000000000..dbb60ced0 --- /dev/null +++ b/crates/openab-core/src/acp/windows_env.rs @@ -0,0 +1,24 @@ +//! Shared Windows process environment allow-list for ACP spawn and the shell tool. + +/// Shared Windows runtime environment keys copied into child processes. +/// +/// ACP agent spawn (`openab-core`) and `openab-agent` shell execution must use +/// this same allow-list so a future addition cannot silently diverge between +/// the two layers. APP/ LOCALAPPData are intentionally excluded: per-user +/// application-data paths may contain credentials and must not become an +/// implicit agent read surface after prompt injection. +pub const WINDOWS_RUNTIME_ENV_KEYS: &[&str] = &[ + "SystemDrive", + "WINDIR", + "ComSpec", + "PATHEXT", + "TEMP", + "TMP", + "HOMEDRIVE", + "HOMEPATH", + "ProgramData", + "ProgramFiles", + "ProgramFiles(x86)", + "ProgramW6432", + "PSModulePath", +]; diff --git a/crates/openab-core/src/acp_mcp.rs b/crates/openab-core/src/acp_mcp.rs index ab82c3091..7ff8ffd74 100644 --- a/crates/openab-core/src/acp_mcp.rs +++ b/crates/openab-core/src/acp_mcp.rs @@ -401,17 +401,6 @@ mod facade_config_writer { #[cfg(test)] mod tests { - use super::{ - }; - - - - - - - - - /// Unique throwaway workdir with a `.kiro/agents/` tree. async fn tmp_workdir(tag: &str) -> std::path::PathBuf { let dir = std::env::temp_dir().join(format!( diff --git a/crates/openab-core/src/setup/wizard.rs b/crates/openab-core/src/setup/wizard.rs index 67d4acb07..c8a9905e1 100644 --- a/crates/openab-core/src/setup/wizard.rs +++ b/crates/openab-core/src/setup/wizard.rs @@ -160,7 +160,7 @@ fn print_box(lines: &[&str]) { "╔".to_string() + &BORDER.to_string().repeat(width + 2) + "╗" ); for line in lines { - let padded = format!(" {:) -> Result<()> { - let dir = path.parent().unwrap_or(Path::new(".")); - std::fs::create_dir_all(dir)?; - let data = serde_json::to_string_pretty(map)?; - #[cfg(unix)] - { - use std::fs::{File, OpenOptions}; - use std::io::Write as _; - use std::os::unix::fs::OpenOptionsExt; - use std::sync::atomic::{AtomicU64, Ordering}; - static TMP_COUNTER: AtomicU64 = AtomicU64::new(0); +/// Write a complete, synced replacement in the destination directory. Unique +/// `create_new` names plus retry mean a stale temp left by a crashed process +/// cannot wedge a later save even if the OS eventually reuses its process id. +fn write_synced_auth_temp(dir: &Path, data: &str) -> Result { + use std::sync::atomic::{AtomicU64, Ordering}; + + static TMP_COUNTER: AtomicU64 = AtomicU64::new(0); + for _ in 0..1024 { let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed); let tmp = dir.join(format!("auth.json.tmp.{}.{seq}", std::process::id())); - let write_and_sync = || -> Result<()> { - let mut file = OpenOptions::new() - .write(true) - .create_new(true) - .mode(0o600) - .open(&tmp)?; - file.write_all(data.as_bytes())?; - file.sync_all()?; - Ok(()) + let mut file = match create_auth_temp(&tmp) { + Ok(file) => file, + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(e) => return Err(e.into()), }; - if let Err(e) = write_and_sync() { - let _ = std::fs::remove_file(&tmp); - return Err(e); - } - if let Err(e) = std::fs::rename(&tmp, path) { + if let Err(e) = file + .write_all(data.as_bytes()) + .and_then(|_| file.sync_all()) + { + drop(file); let _ = std::fs::remove_file(&tmp); return Err(e.into()); } - // fsync the parent dir so the rename itself is durable; without - // this, the inode swap can be reordered after a power loss even - // though the tmp's contents were synced. - if let Ok(dir_handle) = File::open(dir) { - let _ = dir_handle.sync_all(); + drop(file); + return Ok(tmp); + } + Err(anyhow!( + "unable to allocate a unique auth.json temp file after 1024 attempts" + )) +} + +#[cfg(unix)] +fn create_auth_temp(path: &Path) -> std::io::Result { + use std::fs::OpenOptions; + use std::os::unix::fs::OpenOptionsExt; + OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(path) +} + +#[cfg(windows)] +fn create_auth_temp(path: &Path) -> std::io::Result { + create_restricted_auth_temp(path) +} + +#[cfg(not(any(unix, windows)))] +fn create_auth_temp(path: &Path) -> std::io::Result { + use std::fs::OpenOptions; + OpenOptions::new().write(true).create_new(true).open(path) +} + +/// Process-token SID for the current user. The backing buffer must outlive any +/// `PSID` derived from it. +#[cfg(windows)] +struct CurrentUserSid(Vec); + +#[cfg(windows)] +impl CurrentUserSid { + fn query() -> std::io::Result { + use windows_sys::Win32::Foundation::{CloseHandle, HANDLE}; + use windows_sys::Win32::Security::{ + GetTokenInformation, TokenUser, TOKEN_QUERY, TOKEN_USER, + }; + use windows_sys::Win32::System::Threading::{GetCurrentProcess, OpenProcessToken}; + + let mut token: HANDLE = std::ptr::null_mut(); + if unsafe { OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &mut token) } == 0 { + return Err(std::io::Error::last_os_error()); + } + struct TokenHandle(HANDLE); + impl Drop for TokenHandle { + fn drop(&mut self) { + if !self.0.is_null() { + unsafe { + CloseHandle(self.0); + } + } + } + } + let token = TokenHandle(token); + + let mut needed = 0u32; + unsafe { + GetTokenInformation(token.0, TokenUser, std::ptr::null_mut(), 0, &mut needed); + } + if needed == 0 { + return Err(std::io::Error::last_os_error()); + } + + let mut buffer = vec![0u8; needed as usize]; + if unsafe { + GetTokenInformation( + token.0, + TokenUser, + buffer.as_mut_ptr().cast(), + needed, + &mut needed, + ) + } == 0 + { + return Err(std::io::Error::last_os_error()); + } + if buffer.len() < std::mem::size_of::() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "TokenUser buffer was truncated", + )); + } + Ok(Self(buffer)) + } + + fn as_psid(&self) -> windows_sys::Win32::Security::PSID { + use windows_sys::Win32::Security::TOKEN_USER; + unsafe { (*self.0.as_ptr().cast::()).User.Sid } + } +} + +#[cfg(windows)] +struct LocalAcl(*mut windows_sys::Win32::Security::ACL); + +#[cfg(windows)] +impl Drop for LocalAcl { + fn drop(&mut self) { + if !self.0.is_null() { + unsafe { + windows_sys::Win32::Foundation::LocalFree(self.0.cast()); + } + self.0 = std::ptr::null_mut(); + } + } +} + +#[cfg(windows)] +impl LocalAcl { + fn for_current_user(sid: &CurrentUserSid) -> std::io::Result { + use windows_sys::Win32::Foundation::ERROR_SUCCESS; + use windows_sys::Win32::Security::Authorization::{ + BuildTrusteeWithSidW, SetEntriesInAclW, EXPLICIT_ACCESS_W, SET_ACCESS, TRUSTEE_IS_USER, + }; + use windows_sys::Win32::Security::NO_INHERITANCE; + use windows_sys::Win32::Storage::FileSystem::FILE_ALL_ACCESS; + + let mut access = EXPLICIT_ACCESS_W { + grfAccessPermissions: FILE_ALL_ACCESS, + grfAccessMode: SET_ACCESS, + grfInheritance: NO_INHERITANCE, + Trustee: Default::default(), + }; + unsafe { + BuildTrusteeWithSidW(&mut access.Trustee, sid.as_psid()); + } + access.Trustee.TrusteeType = TRUSTEE_IS_USER; + + let mut acl = std::ptr::null_mut(); + let status = unsafe { SetEntriesInAclW(1, &access, std::ptr::null(), &mut acl) }; + if status != ERROR_SUCCESS { + return Err(std::io::Error::from_raw_os_error(status as i32)); + } + Ok(Self(acl)) + } +} + +/// Create the auth temp with an explicit current-user DACL so the file never +/// inherits a permissive parent ACL. Trustee is the process token SID, not +/// `USERNAME` / `USERDOMAIN`. +#[cfg(windows)] +fn create_restricted_auth_temp(path: &Path) -> std::io::Result { + use std::os::windows::ffi::OsStrExt; + use std::os::windows::io::{FromRawHandle, RawHandle}; + use windows_sys::Win32::Foundation::{ + ERROR_ALREADY_EXISTS, ERROR_FILE_EXISTS, GENERIC_WRITE, INVALID_HANDLE_VALUE, + }; + use windows_sys::Win32::Security::{ + InitializeSecurityDescriptor, SetSecurityDescriptorDacl, SECURITY_ATTRIBUTES, + SECURITY_DESCRIPTOR, + }; + use windows_sys::Win32::Storage::FileSystem::{ + CreateFileW, CREATE_NEW, FILE_ATTRIBUTE_NORMAL, FILE_SHARE_NONE, + }; + + let sid = CurrentUserSid::query()?; + let acl = LocalAcl::for_current_user(&sid)?; + let mut sd: SECURITY_DESCRIPTOR = unsafe { std::mem::zeroed() }; + if unsafe { InitializeSecurityDescriptor((&mut sd as *mut SECURITY_DESCRIPTOR).cast(), 1) } == 0 + { + return Err(std::io::Error::last_os_error()); + } + if unsafe { + SetSecurityDescriptorDacl((&mut sd as *mut SECURITY_DESCRIPTOR).cast(), 1, acl.0, 0) + } == 0 + { + return Err(std::io::Error::last_os_error()); + } + let sa = SECURITY_ATTRIBUTES { + nLength: std::mem::size_of::() as u32, + lpSecurityDescriptor: (&mut sd as *mut SECURITY_DESCRIPTOR).cast(), + bInheritHandle: 0, + }; + let path_w: Vec = path.as_os_str().encode_wide().chain(Some(0)).collect(); + let handle = unsafe { + CreateFileW( + path_w.as_ptr(), + GENERIC_WRITE, + FILE_SHARE_NONE, + &sa, + CREATE_NEW, + FILE_ATTRIBUTE_NORMAL, + std::ptr::null_mut(), + ) + }; + if handle == INVALID_HANDLE_VALUE { + let err = std::io::Error::last_os_error(); + return Err(match err.raw_os_error() { + Some(code) + if code == ERROR_FILE_EXISTS as i32 || code == ERROR_ALREADY_EXISTS as i32 => + { + std::io::Error::new(std::io::ErrorKind::AlreadyExists, err) + } + _ => err, + }); + } + Ok(unsafe { std::fs::File::from_raw_handle(handle as RawHandle) }) +} + +/// Re-apply the restricted current-user DACL. Required after `ReplaceFileW`, +/// which preserves the replaced file's security descriptor. +#[cfg(windows)] +fn restrict_auth_dacl(path: &Path) -> std::io::Result<()> { + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Foundation::ERROR_SUCCESS; + use windows_sys::Win32::Security::Authorization::{SetNamedSecurityInfoW, SE_FILE_OBJECT}; + use windows_sys::Win32::Security::{ + DACL_SECURITY_INFORMATION, PROTECTED_DACL_SECURITY_INFORMATION, + }; + + let sid = CurrentUserSid::query()?; + let acl = LocalAcl::for_current_user(&sid)?; + let path_w: Vec = path.as_os_str().encode_wide().chain(Some(0)).collect(); + let status = unsafe { + SetNamedSecurityInfoW( + path_w.as_ptr(), + SE_FILE_OBJECT, + DACL_SECURITY_INFORMATION | PROTECTED_DACL_SECURITY_INFORMATION, + std::ptr::null_mut(), + std::ptr::null_mut(), + acl.0, + std::ptr::null(), + ) + }; + if status != ERROR_SUCCESS { + return Err(std::io::Error::from_raw_os_error(status as i32)); + } + Ok(()) +} + +/// Commit a synced temp file over `auth.json` without exposing a partially +/// written destination. Both backends keep the replacement in the same +/// directory/volume so the namespace switch is atomic. +fn replace_auth_file(tmp: &Path, path: &Path) -> std::io::Result<()> { + #[cfg(unix)] + { + std::fs::rename(tmp, path)?; + // Persist the directory entry swap after the temp contents themselves + // were synced. Best-effort because not every Unix filesystem supports + // syncing directory handles. + if let Some(dir) = path.parent() { + if let Ok(dir_handle) = std::fs::File::open(dir) { + let _ = dir_handle.sync_all(); + } + } + Ok(()) + } + + #[cfg(windows)] + { + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Storage::FileSystem::{ + MoveFileExW, ReplaceFileW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, + REPLACEFILE_WRITE_THROUGH, + }; + + fn wide(path: &Path) -> Vec { + path.as_os_str().encode_wide().chain(Some(0)).collect() + } + + let replacement = wide(tmp); + let destination = wide(path); + // Try ReplaceFileW first, without a path.exists() check: that check is + // a TOCTOU race with another writer. MoveFileExW is only the + // first-creation fallback when ReplaceFileW reports FILE_NOT_FOUND. + let replace_error = unsafe { + if ReplaceFileW( + destination.as_ptr(), + replacement.as_ptr(), + std::ptr::null(), + REPLACEFILE_WRITE_THROUGH, + std::ptr::null_mut(), + std::ptr::null_mut(), + ) == 0 + { + Some(std::io::Error::last_os_error()) + } else { + None + } + }; + match replace_error { + None => restrict_auth_dacl(path), + Some(error) + if error.raw_os_error() + == Some(windows_sys::Win32::Foundation::ERROR_FILE_NOT_FOUND as i32) => + { + let ok = unsafe { + MoveFileExW( + replacement.as_ptr(), + destination.as_ptr(), + MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, + ) + }; + if ok == 0 { + Err(std::io::Error::last_os_error()) + } else { + restrict_auth_dacl(path) + } + } + Some(error) => Err(error), } } - #[cfg(not(unix))] + + #[cfg(not(any(unix, windows)))] { - std::fs::write(path, &data)?; + std::fs::rename(tmp, path) + } +} + +/// Atomically replace `auth.json` with a complete, synced new map. A crash +/// before `replace_auth_file` leaves the prior file untouched; after the atomic +/// namespace switch, readers observe only the complete new file. This protects +/// refresh-token rotation from half-written credential state on both Unix and +/// Windows. +fn write_auth_file(path: &Path, map: &HashMap) -> Result<()> { + let dir = path.parent().unwrap_or(Path::new(".")); + std::fs::create_dir_all(dir)?; + let data = serde_json::to_string_pretty(map)?; + let tmp = write_synced_auth_temp(dir, &data)?; + if let Err(e) = replace_auth_file(&tmp, path) { + let _ = std::fs::remove_file(&tmp); + return Err(e.into()); } Ok(()) } @@ -555,86 +855,88 @@ fn auth_subcommand(namespace: &str) -> &'static str { // refresh per tenant so concurrent processes present a rotated `RT_old` // only once, never tripping OAuth 2.1 §10.4 token-family revocation. // -// `flock(2)` (not a sentinel lockfile) so the kernel auto-releases on fd close / -// process death — no stale lock, no orphan cleanup. The lock lives on a sidecar, -// never on `auth.json` itself, because the atomic tmp+rename swaps that inode out -// from under any lock held on it. `#[cfg(unix)]`; a non-unix build is a no-op -// (openab-agent is de-facto unix-only — see `write_auth_file`). +// The standard-library API maps to OS advisory file locks, so fd/handle close +// and process death release a lock without stale sentinel cleanup. The lock +// lives on a sidecar, never on `auth.json` itself, because atomic replacement +// swaps that inode/file entry. /// Sidecar lock path `auth.json..lock`, next to the auth file so a /// test-injected tempdir locks its own sidecar rather than the real `$HOME` one. -#[cfg(unix)] fn lock_path_for(auth: &Path, suffix: &str) -> PathBuf { let dir = auth.parent().unwrap_or_else(|| Path::new(".")); dir.join(format!("auth.json.{suffix}.lock")) } -/// RAII guard releasing the advisory lock on drop. The kernel also drops it on -/// fd close / process death, so a crashed holder never wedges the file. -#[cfg(unix)] +/// RAII guard releasing the advisory lock on drop. Supported Unix and Windows +/// builds share this API and the kernel also releases the lock on process death. +#[cfg(any(unix, windows))] pub(crate) struct AuthFileLock { file: std::fs::File, } -#[cfg(unix)] +#[cfg(not(any(unix, windows)))] +pub(crate) struct AuthFileLock; + +#[cfg(any(unix, windows))] impl Drop for AuthFileLock { fn drop(&mut self) { - use std::os::unix::io::AsRawFd; - // SAFETY: `self.file` owns a valid fd; flock has no memory effects. - unsafe { libc::flock(self.file.as_raw_fd(), libc::LOCK_UN) }; + let _ = self.file.unlock(); } } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn open_lock_file(lock: &Path) -> Result { + #[cfg(unix)] use std::os::unix::fs::OpenOptionsExt; if let Some(dir) = lock.parent() { std::fs::create_dir_all(dir)?; } - Ok(std::fs::OpenOptions::new() - .write(true) - .create(true) - .truncate(false) - .mode(0o600) - .open(lock)?) + let mut options = std::fs::OpenOptions::new(); + options.read(true).write(true).create(true).truncate(false); + #[cfg(unix)] + options.mode(0o600); + Ok(options.open(lock)?) } /// Blocking exclusive lock. Used ONLY for the global file RMW, which performs no /// network I/O while held, so acquisition blocks at most for another process's /// fast tmp+rename — never for a slow refresh (those take the per-tenant lock). -#[cfg(unix)] -fn flock_exclusive(lock: &Path) -> Result { - use std::os::unix::io::AsRawFd; +#[cfg(any(unix, windows))] +fn lock_exclusive(lock: &Path) -> Result { let file = open_lock_file(lock)?; - // SAFETY: valid fd held by `file`; flock has no memory effects. - let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) }; - if rc != 0 { - return Err(std::io::Error::last_os_error().into()); - } + file.lock()?; Ok(AuthFileLock { file }) } -/// Acquire the global `auth.json` write lock (a no-op `None` guard off-unix). +#[cfg(any(unix, windows))] +fn try_lock_exclusive(file: &std::fs::File) -> std::io::Result { + match file.try_lock() { + Ok(()) => Ok(true), + Err(std::fs::TryLockError::WouldBlock) => Ok(false), + Err(std::fs::TryLockError::Error(e)) => Err(e), + } +} + +/// Acquire the global `auth.json` write lock (a no-op only on unsupported +/// platforms that are neither Unix nor Windows). /// Both `with_auth_locked` and `McpCredentialStore::clear` — which needs a /// delete-on-empty tail the funnel can't express — acquire here, so the /// `"global"` sidecar name and the acquire policy live in exactly one place. fn lock_global(path: &Path) -> Result> { - #[cfg(unix)] + #[cfg(any(unix, windows))] { - Ok(Some(flock_exclusive(&lock_path_for(path, "global"))?)) + Ok(Some(lock_exclusive(&lock_path_for(path, "global"))?)) } - #[cfg(not(unix))] + #[cfg(not(any(unix, windows)))] { - // No flock(2) off-unix: every writer runs unprotected, so concurrent - // processes can silently corrupt auth.json (ADR §5.4). openab-agent is - // de-facto unix-only; warn once rather than fail silently so a non-unix - // build with concurrent processes is at least diagnosable. + // Keep unsupported targets diagnosable instead of silently claiming a + // lock guarantee they do not have. use std::sync::Once; static WARN_NO_LOCK: Once = Once::new(); WARN_NO_LOCK.call_once(|| { tracing::warn!( - "auth.json cross-process file locking is unavailable on this non-unix platform; \ - concurrent openab-agent processes may corrupt stored credentials (ADR §5.4)" + "auth.json cross-process file locking is unavailable on this platform; \ + concurrent processes may corrupt stored credentials (ADR §5.4)" ); }); let _ = path; @@ -682,7 +984,7 @@ fn gc_stale_pending(map: &mut HashMap) { pub(crate) const REFRESH_HTTP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(8); /// Outcome of acquiring a tenant's refresh lock. See [`lock_tenant_refresh`]. -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) enum RefreshLock { /// Lock acquired — hold across the refresh. Held(AuthFileLock), @@ -699,7 +1001,7 @@ pub(crate) enum RefreshLock { /// the MCP path makes two — rmcp's `initialize_from_store()` (authorization-server /// discovery) then `get_access_token()` (the refresh) — each bounded by /// [`REFRESH_HTTP_TIMEOUT`]. -#[cfg(unix)] +#[cfg(any(unix, windows))] const MAX_REFRESH_ROUND_TRIPS: u64 = 2; /// Lock-acquire deadline. Sized strictly above the worst-case lock-hold @@ -708,7 +1010,7 @@ const MAX_REFRESH_ROUND_TRIPS: u64 = 2; /// bounded — and, on the MCP path, multi-call — refresh; only a genuinely stuck /// holder trips the timeout. Derived from `REFRESH_HTTP_TIMEOUT` so the relationship /// can't silently drift if that bound changes. -#[cfg(unix)] +#[cfg(any(unix, windows))] const REFRESH_LOCK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(REFRESH_HTTP_TIMEOUT.as_secs() * MAX_REFRESH_ROUND_TRIPS + 4); @@ -741,23 +1043,22 @@ const REFRESH_LOCK_TIMEOUT: std::time::Duration = /// token is already fresh). `force_refresh` intentionally skips that optimisation and /// always refreshes (it runs on a 401, where the clock-fresh token is already /// known-bad); it stays reuse-safe because it, too, loads inside the lock. -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) async fn lock_tenant_refresh(auth: &Path, tenant: &str) -> RefreshLock { lock_tenant_refresh_until(auth, tenant, REFRESH_LOCK_TIMEOUT).await } /// [`lock_tenant_refresh`] with an injectable deadline so tests can drive the /// fail-closed timeout path in milliseconds instead of [`REFRESH_LOCK_TIMEOUT`]. -#[cfg(unix)] +#[cfg(any(unix, windows))] async fn lock_tenant_refresh_until( auth: &Path, tenant: &str, timeout: std::time::Duration, ) -> RefreshLock { - use std::os::unix::io::AsRawFd; let lock = lock_path_for(auth, &format!("refresh.{tenant}")); - // Open the lock fd once; re-issue `flock` on it each retry instead of - // re-opening (and re-`create_dir_all`-ing) the same file every 100 ms. + // Open the sidecar once; retry the same kernel lock every 100 ms instead of + // re-opening (and re-`create_dir_all`-ing) the file on each attempt. let file = match open_lock_file(&lock) { Ok(f) => f, Err(e) => { @@ -767,17 +1068,15 @@ async fn lock_tenant_refresh_until( }; let deadline = std::time::Instant::now() + timeout; loop { - // SAFETY: valid fd held by `file`; flock has no memory effects. - let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) }; - if rc == 0 { - return RefreshLock::Held(AuthFileLock { file }); - } - let err = std::io::Error::last_os_error(); - // EWOULDBLOCK/EAGAIN (both `ErrorKind::WouldBlock`) = another holder is - // refreshing; any other errno is a real failure we degrade on. - if err.kind() != std::io::ErrorKind::WouldBlock { - tracing::warn!(tenant, error = %err, "refresh lock unavailable; proceeding unserialised"); - return RefreshLock::Unavailable; + match try_lock_exclusive(&file) { + Ok(true) => { + return RefreshLock::Held(AuthFileLock { file }); + } + Ok(false) => {} + Err(err) => { + tracing::warn!(tenant, error = %err, "refresh lock unavailable; proceeding unserialised"); + return RefreshLock::Unavailable; + } } if std::time::Instant::now() >= deadline { // Fail-closed (see fn doc): the refresh is HTTP-bounded shorter than this @@ -940,7 +1239,7 @@ pub async fn get_valid_token_for(namespace: &str) -> Result { // second process does not present the same RT_old (§5.4 (b)). Fail closed // on a contended-lock timeout: surface a retryable error rather than // refresh unserialised (which would risk §10.4 family revocation). - #[cfg(unix)] + #[cfg(any(unix, windows))] let _refresh_guard = match lock_tenant_refresh(&auth_path(), namespace).await { RefreshLock::Held(g) => Some(g), RefreshLock::Unavailable => None, @@ -965,7 +1264,7 @@ pub async fn get_valid_token_for(namespace: &str) -> Result { pub async fn force_refresh_for(namespace: &str) -> Result { // Serialise even a forced refresh so two of them can't both rotate RT_old. // Fail closed on timeout (see get_valid_token_for) rather than refresh unserialised. - #[cfg(unix)] + #[cfg(any(unix, windows))] let _refresh_guard = match lock_tenant_refresh(&auth_path(), namespace).await { RefreshLock::Held(g) => Some(g), RefreshLock::Unavailable => None, @@ -1924,9 +2223,7 @@ mod tests { #[test] fn test_auth_path() { - assert!(auth_path() - .to_string_lossy() - .contains(".openab/agent/auth.json")); + assert!(auth_path().ends_with(Path::new(".openab").join("agent").join("auth.json"))); } #[test] @@ -2037,6 +2334,134 @@ mod tests { assert_eq!(token_of(map.get("mcp:github")).expires_at, 42); } + #[cfg(windows)] + #[test] + fn windows_auth_acl_uses_process_token_not_env_name() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("auth.json"); + let mut input = HashMap::new(); + input.insert("codex".to_string(), AuthEntry::Token(make_store(7))); + temp_env::with_vars( + [ + ("USERNAME", Some("definitely-not-a-windows-account")), + ("USERDOMAIN", Some("NO-SUCH-DOMAIN")), + ], + || { + write_auth_file(&path, &input).unwrap(); + write_auth_file(&path, &input).unwrap(); + }, + ); + let map = read_auth_file(&path).unwrap(); + assert_eq!(token_of(map.get("codex")).expires_at, 7); + } + + #[test] + fn atomic_auth_replace_exposes_complete_old_or_new_state() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("auth.json"); + + let mut old = HashMap::new(); + old.insert("codex".to_string(), AuthEntry::Token(make_store(1))); + write_auth_file(&path, &old).unwrap(); + + let mut new = HashMap::new(); + new.insert("codex".to_string(), AuthEntry::Token(make_store(2))); + new.insert("github".to_string(), AuthEntry::Mcp(make_mcp_creds())); + let data = serde_json::to_string_pretty(&new).unwrap(); + let tmp = write_synced_auth_temp(dir.path(), &data).unwrap(); + + // Simulated crash before commit: the synced temp may remain, but the + // canonical file is still the complete prior generation. + let before = read_auth_file(&path).unwrap(); + assert_eq!(token_of(before.get("codex")).expires_at, 1); + assert!(before.get("github").is_none()); + + replace_auth_file(&tmp, &path).unwrap(); + assert!(!tmp.exists(), "atomic commit consumes the temp name"); + let after = read_auth_file(&path).unwrap(); + assert_eq!(token_of(after.get("codex")).expires_at, 2); + assert!(matches!(after.get("github"), Some(AuthEntry::Mcp(_)))); + } + + #[cfg(any(unix, windows))] + #[test] + fn auth_file_lock_is_exclusive_and_released_on_drop() { + let dir = tempfile::tempdir().unwrap(); + let auth = dir.path().join("auth.json"); + let lock_path = lock_path_for(&auth, "global"); + + let held = lock_exclusive(&lock_path).unwrap(); + let contender = open_lock_file(&lock_path).unwrap(); + assert!( + !try_lock_exclusive(&contender).unwrap(), + "second handle must observe contention" + ); + + drop(held); + assert!( + try_lock_exclusive(&contender).unwrap(), + "lock must be acquirable after the RAII guard drops" + ); + contender.unlock().unwrap(); + } + + #[cfg(any(unix, windows))] + #[test] + fn auth_lock_child_process_holder() { + let Some(lock_path) = std::env::var_os("OPENAB_AUTH_LOCK_CHILD_PATH") else { + return; + }; + let ready_path = std::env::var_os("OPENAB_AUTH_LOCK_CHILD_READY") + .expect("child ready path is required when child lock path is set"); + let _held = lock_exclusive(Path::new(&lock_path)).unwrap(); + std::fs::write(ready_path, b"ready").unwrap(); + loop { + std::thread::sleep(std::time::Duration::from_secs(60)); + } + } + + #[cfg(any(unix, windows))] + #[test] + fn auth_file_lock_is_released_after_holder_process_is_killed() { + let dir = tempfile::tempdir().unwrap(); + let auth = dir.path().join("auth.json"); + let lock_path = lock_path_for(&auth, "crash-test"); + let ready_path = dir.path().join("child.ready"); + + let mut child = std::process::Command::new(std::env::current_exe().unwrap()) + .arg("--exact") + .arg("auth::tests::auth_lock_child_process_holder") + .arg("--nocapture") + .env("OPENAB_AUTH_LOCK_CHILD_PATH", &lock_path) + .env("OPENAB_AUTH_LOCK_CHILD_READY", &ready_path) + .spawn() + .unwrap(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + while !ready_path.exists() && std::time::Instant::now() < deadline { + std::thread::sleep(std::time::Duration::from_millis(20)); + } + if !ready_path.exists() { + let _ = child.kill(); + let _ = child.wait(); + panic!("child did not acquire the auth lock before the deadline"); + } + + let contender = open_lock_file(&lock_path).unwrap(); + assert!( + !try_lock_exclusive(&contender).unwrap(), + "parent must observe the child process lock" + ); + + child.kill().unwrap(); + child.wait().unwrap(); + assert!( + try_lock_exclusive(&contender).unwrap(), + "kernel must release the lock when the holder process dies" + ); + contender.unlock().unwrap(); + } + #[cfg(unix)] #[test] fn write_auth_file_creates_file_with_0600_mode() { @@ -2365,7 +2790,7 @@ mod tests { assert!(map.get("codex").is_some(), "real tenant untouched"); } - #[cfg(unix)] + #[cfg(any(unix, windows))] #[tokio::test] async fn lock_tenant_refresh_fails_closed_when_contended() { // §5.4 (b), fail-closed: while one holder keeps the tenant refresh lock, a @@ -2418,7 +2843,7 @@ mod tests { assert_eq!(token_of(map.get("codex")).expires_at, 1); } - #[cfg(unix)] + #[cfg(any(unix, windows))] #[tokio::test] async fn lock_tenant_refresh_fails_closed_for_anthropic_and_is_per_tenant() { // §5.4 (b) proven for the `anthropic-oauth` tenant: while one holder keeps diff --git a/crates/openab-mcp/src/mcp/runtime.rs b/crates/openab-mcp/src/mcp/runtime.rs index b45ff785f..6b0373d80 100644 --- a/crates/openab-mcp/src/mcp/runtime.rs +++ b/crates/openab-mcp/src/mcp/runtime.rs @@ -669,7 +669,8 @@ impl McpRuntimeManager { // implicit: `mgr.initialize_from_store()` (called below) re-`load()`s // `auth.json` from disk, after which `get_access_token` skips the network // refresh when the loaded token is already fresh, so a process that loses - // the race adopts the token the winner wrote. Non-unix = no-op. + // the race adopts the token the winner wrote. Unsupported platforms + // outside Unix/Windows retain the legacy no-op fallback. // // Cross-module invariant: the per-tenant refresh lock and the credential // entry must key off the *same* server identifier. `name` is passed both @@ -679,7 +680,7 @@ impl McpRuntimeManager { // Fail closed on a contended-lock timeout: a `Transient` error so the // caller retries WITHOUT forcing re-login (NeedsAuth) or tripping the // breaker. `Held`/`Unavailable` both proceed to drive the refresh. - #[cfg(unix)] + #[cfg(any(unix, windows))] let _refresh_guard = match crate::auth::lock_tenant_refresh(&self.auth_path, name).await { crate::auth::RefreshLock::Held(g) => Some(g), crate::auth::RefreshLock::Unavailable => None, diff --git a/docs/windows-acp-native-canary.md b/docs/windows-acp-native-canary.md new file mode 100644 index 000000000..abf881294 --- /dev/null +++ b/docs/windows-acp-native-canary.md @@ -0,0 +1,50 @@ +# Windows ACP native canary + +This is a non-release, no-provider-call canary for the two Windows binaries built from the +`spike/windows-acp-native-agent` branch. + +## Safety boundary + +- Use only the `openab.exe` and `openab-agent.exe` shipped in the same canary archive. +- Verify the archive and binary SHA-256 values against `MANIFEST.json` and `SHA256SUMS.txt`. +- The script creates an isolated temporary home and uses a non-secret sentinel credential with + an explicit Anthropic provider/model selection so provider construction is deterministic. +- All provider HTTP(S) proxy variables point to the closed loopback port `127.0.0.1:9`. +- It does not log in, read an existing auth store, call a usable provider endpoint, publish a + release, install anything, or modify the repository. +- Temporary files are deleted in `finally`. The script force-stops any canary process left after a + failed assertion. + +## Run on Windows 11 + +Open PowerShell in the extracted directory and run: + +```powershell +Get-FileHash .\openab.exe -Algorithm SHA256 +Get-FileHash .\openab-agent.exe -Algorithm SHA256 + +PowerShell -NoProfile -ExecutionPolicy Bypass -File .\windows-acp-native-canary.ps1 ` + -OpenAbExe .\openab.exe ` + -AgentExe .\openab-agent.exe ` + -WorkDir $PWD +``` + +Success ends with `WINDOWS_DUAL_LAYER_CANARY_PASS`. Any missing response, timeout, unexpected +process count, or surviving downstream agent fails closed with a non-zero exit. + +## What this proves + +1. Both PE binaries start on Windows. +2. `openab-agent.exe` completes ACP `initialize` and `session/new` over stdio without a provider + network request. +3. `openab.exe` serves loopback `/acp`, completes upstream `initialize` and `session/new`, and the + first prompt crosses the full `openab.exe -> openab-agent.exe` process boundary. +4. Gateway `session/cancel` settles the upstream waiter as `cancelled`. +5. The CI-only controller regression terminates the Job Object while the connection mutex is held, + proving cleanup does not depend on the streaming future unwinding or on implicit connection + drop. +6. Abruptly terminating the root process closes its `KILL_ON_JOB_CLOSE` fallback and removes the + downstream agent. + +It does **not** prove a real model/provider turn, installation UX, Windows 10 compatibility, or +that gateway cancellation stops downstream model computation. Those remain separate gates. diff --git a/openab-agent/Cargo.lock b/openab-agent/Cargo.lock index a12cd2167..4db4f5972 100644 --- a/openab-agent/Cargo.lock +++ b/openab-agent/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "ahash" version = "0.8.12" @@ -250,6 +256,24 @@ version = "0.6.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e" +[[package]] +name = "bytemuck" +version = "1.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec" + +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + +[[package]] +name = "byteorder-lite" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" + [[package]] name = "bytes" version = "1.11.1" @@ -305,6 +329,16 @@ dependencies = [ "windows-link", ] +[[package]] +name = "chrono-tz" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6139a8597ed92cf816dfb33f5dd6cf0bb93a6adc938f11039f371bc5bcd26c3" +dependencies = [ + "chrono", + "phf 0.12.1", +] + [[package]] name = "clap" version = "4.6.1" @@ -354,6 +388,12 @@ dependencies = [ "cc", ] +[[package]] +name = "color_quant" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d7b894f5411737b7867f4827955924d7c254fc9f4d91a6aad6b097804b1018b" + [[package]] name = "colorchoice" version = "1.0.5" @@ -410,6 +450,27 @@ dependencies = [ "libc", ] +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "cron" +version = "0.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "089df96cf6a25253b4b6b6744d86f91150a3d4df546f31a95def47976b8cba97" +dependencies = [ + "chrono", + "once_cell", + "phf 0.11.3", + "winnow", +] + [[package]] name = "crypto-common" version = "0.1.7" @@ -488,6 +549,15 @@ dependencies = [ "serde", ] +[[package]] +name = "emojis" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0a4d5d50b0b58df5173d8ff1192b4d1422ceae5d981b30d4b6f8ed1d673a2bc4" +dependencies = [ + "phf 0.13.1", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -521,12 +591,31 @@ version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +[[package]] +name = "fdeflate" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e6853b52649d4ac5c0bd02320cddc5ba956bdb407c4b75a2c6b75bf51500f8c" +dependencies = [ + "simd-adler32", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flate2" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" +dependencies = [ + "crc32fast", + "miniz_oxide", +] + [[package]] name = "fluent-uri" version = "0.4.1" @@ -714,6 +803,16 @@ dependencies = [ "wasip3", ] +[[package]] +name = "gif" +version = "0.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee8cfcc411d9adbbaba82fb72661cc1bcca13e8bba98b364e62b2dba8f960159" +dependencies = [ + "color_quant", + "weezl", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -830,11 +929,11 @@ dependencies = [ "http", "hyper", "hyper-util", - "rustls", + "rustls 0.23.40", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.4", "tower-service", - "webpki-roots", + "webpki-roots 1.0.7", ] [[package]] @@ -993,6 +1092,34 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "image" +version = "0.25.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85ab80394333c02fe689eaf900ab500fbd0c2213da414687ebf995a65d5a6104" +dependencies = [ + "bytemuck", + "byteorder-lite", + "color_quant", + "gif", + "image-webp", + "moxcms", + "num-traits", + "png", + "zune-core", + "zune-jpeg", +] + +[[package]] +name = "image-webp" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "525e9ff3e1a4be2fbea1fdf0e98686a6d98b4d8f937e1bf7402245af1909e8c3" +dependencies = [ + "byteorder-lite", + "quick-error", +] + [[package]] name = "indexmap" version = "2.14.0" @@ -1224,6 +1351,26 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "mime_guess" +version = "2.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e" +dependencies = [ + "mime", + "unicase", +] + +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.2.1" @@ -1235,6 +1382,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "moxcms" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb85c154ba489f01b25c0d36ae69a87e4a1c73a72631fc6c0eb6dde34a73e44b" +dependencies = [ + "num-traits", + "pxfm", +] + [[package]] name = "nix" version = "0.31.3" @@ -1385,8 +1542,9 @@ dependencies = [ "async-trait", "base64", "clap", - "libc", + "openab-core", "openab-mcp", + "process-wrap", "reqwest 0.12.28", "rmcp", "serde", @@ -1400,6 +1558,43 @@ dependencies = [ "uuid", ] +[[package]] +name = "openab-core" +version = "0.8.5" +dependencies = [ + "anyhow", + "async-trait", + "base64", + "bytes", + "chrono", + "chrono-tz", + "clap", + "cron", + "emojis", + "futures-util", + "image", + "libc", + "process-wrap", + "pulldown-cmark", + "rand 0.8.6", + "regex", + "reqwest 0.12.28", + "rpassword", + "serde", + "serde_json", + "sha2 0.10.9", + "tempfile", + "tokio", + "tokio-tungstenite", + "toml", + "toml_edit", + "tracing", + "tracing-subscriber", + "unicode-segmentation", + "unicode-width", + "uuid", +] + [[package]] name = "openab-mcp" version = "0.1.0" @@ -1423,6 +1618,7 @@ dependencies = [ "tracing", "url", "urlencoding", + "windows-sys 0.61.2", ] [[package]] @@ -1478,12 +1674,103 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "phf" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd6780a80ae0c52cc120a26a1a42c1ae51b247a253e4e06113d23d2c2edd078" +dependencies = [ + "phf_macros", + "phf_shared 0.11.3", +] + +[[package]] +name = "phf" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "913273894cec178f401a31ec4b656318d95473527be05c0752cc41cdc32be8b7" +dependencies = [ + "phf_shared 0.12.1", +] + +[[package]] +name = "phf" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1562dc717473dbaa4c1f85a36410e03c047b2e7df7f45ee938fbef64ae7fadf" +dependencies = [ + "phf_shared 0.13.1", +] + +[[package]] +name = "phf_generator" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d" +dependencies = [ + "phf_shared 0.11.3", + "rand 0.8.6", +] + +[[package]] +name = "phf_macros" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f84ac04429c13a7ff43785d75ad27569f2951ce0ffd30a3321230db2fc727216" +dependencies = [ + "phf_generator", + "phf_shared 0.11.3", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "phf_shared" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67eabc2ef2a60eb7faa00097bd1ffdb5bd28e62bf39990626a582201b7a754e5" +dependencies = [ + "siphasher", +] + +[[package]] +name = "phf_shared" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06005508882fb681fd97892ecff4b7fd0fee13ef1aa569f8695dae7ab9099981" +dependencies = [ + "siphasher", +] + +[[package]] +name = "phf_shared" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e57fef6bc5981e38c2ce2d63bfa546861309f875b8a75f092d1d54ae2d64f266" +dependencies = [ + "siphasher", +] + [[package]] name = "pin-project-lite" version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "png" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "60769b8b31b2a9f263dae2776c37b1b28ae246943cf719eb6946a1db05128a61" +dependencies = [ + "bitflags", + "crc32fast", + "fdeflate", + "flate2", + "miniz_oxide", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -1535,6 +1822,29 @@ dependencies = [ "windows", ] +[[package]] +name = "pulldown-cmark" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f068eba8e7071c5f9511831b44f32c740d5adf574e990f946ddb53db2f314e" +dependencies = [ + "bitflags", + "memchr", + "unicase", +] + +[[package]] +name = "pxfm" +version = "0.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0c5ccf5294c6ccd63a74f1565028353830a9c2f5eb0c682c355c471726a6e3f" + +[[package]] +name = "quick-error" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3" + [[package]] name = "quinn" version = "0.11.9" @@ -1547,7 +1857,7 @@ dependencies = [ "quinn-proto", "quinn-udp", "rustc-hash", - "rustls", + "rustls 0.23.40", "socket2", "thiserror 2.0.18", "tokio", @@ -1568,7 +1878,7 @@ dependencies = [ "rand 0.9.4", "ring", "rustc-hash", - "rustls", + "rustls 0.23.40", "rustls-pki-types", "slab", "thiserror 2.0.18", @@ -1771,7 +2081,9 @@ checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ "base64", "bytes", + "futures-channel", "futures-core", + "futures-util", "http", "http-body", "http-body-util", @@ -1780,25 +2092,28 @@ dependencies = [ "hyper-util", "js-sys", "log", + "mime_guess", "percent-encoding", "pin-project-lite", "quinn", - "rustls", + "rustls 0.23.40", "rustls-pki-types", "serde", "serde_json", "serde_urlencoded", "sync_wrapper", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.4", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams 0.4.2", "web-sys", - "webpki-roots", + "webpki-roots 1.0.7", ] [[package]] @@ -1822,14 +2137,14 @@ dependencies = [ "percent-encoding", "pin-project-lite", "quinn", - "rustls", + "rustls 0.23.40", "rustls-pki-types", "rustls-platform-verifier", "serde", "serde_json", "sync_wrapper", "tokio", - "tokio-rustls", + "tokio-rustls 0.26.4", "tokio-util", "tower", "tower-http", @@ -1837,7 +2152,7 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", + "wasm-streams 0.5.0", "web-sys", ] @@ -1888,6 +2203,27 @@ dependencies = [ "uuid", ] +[[package]] +name = "rpassword" +version = "7.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2da316a15f47e3d053de9cb2c439650bd8fa4aaeb9365f2e5f27f492ff73c196" +dependencies = [ + "libc", + "rtoolbox", + "windows-sys 0.61.2", +] + +[[package]] +name = "rtoolbox" +version = "0.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50a0e551c1e27e1731aba276dbeaeac73f53c7cd34d1bda485d02bd1e0f36844" +dependencies = [ + "libc", + "windows-sys 0.59.0", +] + [[package]] name = "rustc-hash" version = "2.1.2" @@ -1916,6 +2252,20 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustls" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf4ef73721ac7bcd79b2b315da7779d8fc09718c6b3d2d1b2d94850eb8c18432" +dependencies = [ + "log", + "ring", + "rustls-pki-types", + "rustls-webpki 0.102.8", + "subtle", + "zeroize", +] + [[package]] name = "rustls" version = "0.23.40" @@ -1926,7 +2276,7 @@ dependencies = [ "once_cell", "ring", "rustls-pki-types", - "rustls-webpki", + "rustls-webpki 0.103.13", "subtle", "zeroize", ] @@ -1964,10 +2314,10 @@ dependencies = [ "jni", "log", "once_cell", - "rustls", + "rustls 0.23.40", "rustls-native-certs", "rustls-platform-verifier-android", - "rustls-webpki", + "rustls-webpki 0.103.13", "security-framework", "security-framework-sys", "webpki-root-certs", @@ -1980,6 +2330,17 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" +[[package]] +name = "rustls-webpki" +version = "0.102.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustls-webpki" version = "0.103.13" @@ -2148,6 +2509,15 @@ dependencies = [ "serde_core", ] +[[package]] +name = "serde_spanned" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" +dependencies = [ + "serde", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -2160,6 +2530,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest 0.10.7", +] + [[package]] name = "sha2" version = "0.10.9" @@ -2207,6 +2588,12 @@ dependencies = [ "libc", ] +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "simd_cesu8" version = "1.1.1" @@ -2223,6 +2610,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" +[[package]] +name = "siphasher" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" + [[package]] name = "slab" version = "0.4.12" @@ -2432,13 +2825,24 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-rustls" +version = "0.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "775e0c0f0adb3a2f22a00c4745d728b479985fc15ee7ca6a2608388c5569860f" +dependencies = [ + "rustls 0.22.4", + "rustls-pki-types", + "tokio", +] + [[package]] name = "tokio-rustls" version = "0.26.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" dependencies = [ - "rustls", + "rustls 0.23.40", "tokio", ] @@ -2453,6 +2857,22 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c83b561d025642014097b66e6c1bb422783339e0909e4429cde4749d1990bc38" +dependencies = [ + "futures-util", + "log", + "rustls 0.22.4", + "rustls-pki-types", + "tokio", + "tokio-rustls 0.25.0", + "tungstenite", + "webpki-roots 0.26.11", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -2466,6 +2886,47 @@ dependencies = [ "tokio", ] +[[package]] +name = "toml" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" +dependencies = [ + "serde", + "serde_spanned", + "toml_datetime", + "toml_edit", +] + +[[package]] +name = "toml_datetime" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +dependencies = [ + "serde", +] + +[[package]] +name = "toml_edit" +version = "0.22.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" +dependencies = [ + "indexmap", + "serde", + "serde_spanned", + "toml_datetime", + "toml_write", + "winnow", +] + +[[package]] +name = "toml_write" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" + [[package]] name = "tower" version = "0.5.3" @@ -2554,6 +3015,16 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + [[package]] name = "tracing-subscriber" version = "0.3.23" @@ -2564,12 +3035,15 @@ dependencies = [ "nu-ansi-term", "once_cell", "regex-automata", + "serde", + "serde_json", "sharded-slab", "smallvec", "thread_local", "tracing", "tracing-core", "tracing-log", + "tracing-serde", ] [[package]] @@ -2578,12 +3052,39 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ef1a641ea34f399a848dea702823bbecfb4c486f911735368f1f137cb8257e1" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.8.6", + "rustls 0.22.4", + "rustls-pki-types", + "sha1", + "thiserror 1.0.69", + "url", + "utf-8", +] + [[package]] name = "typenum" version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "unicase" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" + [[package]] name = "unicode-general-category" version = "1.1.0" @@ -2596,6 +3097,18 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-segmentation" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" + +[[package]] +name = "unicode-width" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" + [[package]] name = "unicode-xid" version = "0.2.6" @@ -2627,6 +3140,12 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" @@ -2798,6 +3317,19 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasm-streams" version = "0.5.0" @@ -2852,6 +3384,15 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.7", +] + [[package]] name = "webpki-roots" version = "1.0.7" @@ -2861,6 +3402,12 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "weezl" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a28ac98ddc8b9274cb41bb4d9d4d5c425b6020c50c46f25559911905610b4a88" + [[package]] name = "winapi-util" version = "0.1.11" @@ -2980,6 +3527,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-sys" version = "0.60.2" @@ -3136,6 +3692,15 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" +[[package]] +name = "winnow" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945" +dependencies = [ + "memchr", +] + [[package]] name = "wit-bindgen" version = "0.51.0" @@ -3344,3 +3909,18 @@ name = "zmij" version = "1.0.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" + +[[package]] +name = "zune-core" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb8a0807f7c01457d0379ba880ba6322660448ddebc890ce29bb64da71fb40f9" + +[[package]] +name = "zune-jpeg" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27bc9d5b815bc103f142aa054f561d9187d191692ec7c2d1e2b4737f8dbd7296" +dependencies = [ + "zune-core", +] diff --git a/openab-agent/Cargo.toml b/openab-agent/Cargo.toml index 4afe1d691..951b107aa 100644 --- a/openab-agent/Cargo.toml +++ b/openab-agent/Cargo.toml @@ -10,6 +10,7 @@ description = "Native Rust coding agent with built-in ACP support" # extracted so the broker can host the OAB MCP Facade in-process (ADR §6.2). # This crate re-exports them as `crate::{mcp,auth}` and from `crate::{llm,acp}`. openab-mcp = { path = "../crates/openab-mcp" } +openab-core = { path = "../crates/openab-core", default-features = false } tokio = { version = "1", features = ["full"] } serde = { version = "1", features = ["derive"] } serde_json = "1" @@ -22,12 +23,15 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } base64 = "0.22.1" url = "2.5.8" +process-wrap = { version = "9.1.0", default-features = false, features = [ + "tokio1", + "kill-on-drop", + "process-session", + "job-object", +] } # rmcp model/service types still surface in the ACP layer (sampling bridge). rmcp = { version = "1.7", default-features = false, features = ["client"] } -[target.'cfg(unix)'.dependencies] -libc = "0.2" - [dev-dependencies] tempfile = "3" temp-env = { version = "0.3.6", features = ["async_closure"] } diff --git a/openab-agent/src/tools.rs b/openab-agent/src/tools.rs index e0f898f4f..36c46db72 100644 --- a/openab-agent/src/tools.rs +++ b/openab-agent/src/tools.rs @@ -1,12 +1,27 @@ use anyhow::{anyhow, Result}; +use process_wrap::tokio::{CommandWrap, KillOnDrop}; use serde_json::{json, Value}; use std::collections::HashMap; use std::path::{Path, PathBuf}; +use std::process::{ExitStatus, Stdio}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::io::{AsyncRead, AsyncReadExt}; use tokio::process::Command; +use tokio::sync::oneshot; +use tokio::task::{AbortHandle, JoinHandle}; use tracing::debug; +#[cfg(windows)] +use process_wrap::tokio::JobObject; +#[cfg(unix)] +use process_wrap::tokio::ProcessSession; + use crate::llm::ToolDef; +#[cfg(any(windows, test))] +use base64::Engine as _; + /// Validate that a path is within the allowed working directory. /// This function has NO side-effects — it never creates directories or files. fn validate_path(path: &str, working_dir: &Path) -> Result { @@ -54,10 +69,29 @@ fn validate_path(path: &str, working_dir: &Path) -> Result { )) } -/// Build a filtered environment for bash tool execution. +/// Build a filtered environment for shell tool execution. fn build_env(allow_list: &[String]) -> HashMap { let mut env = HashMap::new(); - for key in &["PATH", "HOME", "USER", "LANG", "TERM", "SHELL"] { + #[cfg(unix)] + let baseline = ["PATH", "HOME", "USER", "LANG", "TERM", "SHELL"]; + #[cfg(windows)] + let baseline: Vec<&str> = { + // General + identity keys stay local; the shared Windows runtime keys + // come from `openab-core` so this allow-list cannot silently diverge + // from the ACP spawn environment. + let mut keys = vec![ + "PATH", + "HOME", + "LANG", + "SystemRoot", + "USERPROFILE", + "USERNAME", + ]; + keys.extend_from_slice(openab_core::acp::WINDOWS_RUNTIME_ENV_KEYS); + keys + }; + + for key in baseline { if let Ok(val) = std::env::var(key) { env.insert(key.to_string(), val); } @@ -181,7 +215,360 @@ fn tool_edit(input: &Value, working_dir: &Path) -> Result { )) } -/// Execute a shell command with process group isolation and env filtering. +/// Prefix PowerShell scripts with deterministic UTF-8 console encodings, then encode the +/// complete script as UTF-16LE for `-EncodedCommand`. This keeps user input out of the Windows +/// command-line quoting layer. +#[cfg(any(windows, test))] +fn powershell_encoded_command(command: &str) -> String { + const PREFIX: &str = concat!( + "[Console]::InputEncoding = [System.Text.UTF8Encoding]::new($false); ", + "[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new($false); ", + "$OutputEncoding = [Console]::OutputEncoding;\n" + ); + let script = format!("{PREFIX}{command}"); + let utf16le: Vec = script.encode_utf16().flat_map(u16::to_le_bytes).collect(); + base64::engine::general_purpose::STANDARD.encode(utf16le) +} + +#[cfg(unix)] +fn platform_shell_command(command: &str) -> Result { + let mut cmd = Command::new("/bin/sh"); + cmd.arg("-c").arg(command); + Ok(cmd) +} + +#[cfg(windows)] +fn platform_shell_command(command: &str) -> Result { + let system_root = std::env::var_os("SystemRoot") + .or_else(|| std::env::var_os("WINDIR")) + .ok_or_else(|| { + anyhow!("bash: SystemRoot is unavailable; refusing PATH-based shell lookup") + })?; + let powershell = PathBuf::from(system_root) + .join("System32") + .join("WindowsPowerShell") + .join("v1.0") + .join("powershell.exe"); + let mut cmd = Command::new(powershell); + cmd.args([ + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-EncodedCommand", + &powershell_encoded_command(command), + ]); + Ok(cmd) +} + +fn format_shell_output(status: ExitStatus, stdout: &[u8], stderr: &[u8]) -> String { + let stdout = String::from_utf8_lossy(stdout); + let stderr = String::from_utf8_lossy(stderr); + let code = status.code().unwrap_or(-1); + + let mut result = String::new(); + if !stdout.is_empty() { + result.push_str(&stdout); + } + if !stderr.is_empty() { + if !result.is_empty() { + result.push('\n'); + } + result.push_str("[stderr]\n"); + result.push_str(&stderr); + } + if code != 0 { + result.push_str(&format!("\n[exit code: {code}]")); + } + result +} + +struct ShellExecution { + status: ExitStatus, + stdout: Vec, + stderr: Vec, +} + +enum ShellSupervisorOutcome { + Completed(ShellExecution), + Cancelled, +} + +/// Controller-owned cancellation guard for one shell process tree. +/// +/// The supervisor task, rather than the caller's future, owns the child handle. Dropping an +/// in-flight caller therefore sends an explicit cancellation request while the supervisor remains +/// alive to kill and reap the complete process tree. Callers with an orderly shutdown path should +/// use `cancel_and_wait` so cleanup failures are observable. +struct ShellCommandController { + cancel_tx: Option>, + outcome_rx: oneshot::Receiver>, +} + +impl ShellCommandController { + fn request_cancel(&mut self) { + if let Some(cancel_tx) = self.cancel_tx.take() { + let _ = cancel_tx.send(()); + } + } + + async fn wait(&mut self) -> Result { + let outcome = (&mut self.outcome_rx) + .await + .map_err(|_| anyhow!("bash: process supervisor stopped before reporting cleanup"))?; + self.cancel_tx.take(); + outcome + } + + async fn cancel_and_wait(&mut self) -> Result { + self.request_cancel(); + self.wait().await + } +} + +impl Drop for ShellCommandController { + fn drop(&mut self) { + self.request_cancel(); + } +} + +async fn read_shell_pipe_into(mut pipe: R, sink: Arc>>) -> std::io::Result<()> +where + R: AsyncRead + Unpin, +{ + let mut chunk = [0u8; 8192]; + loop { + let n = pipe.read(&mut chunk).await?; + if n == 0 { + return Ok(()); + } + sink.lock() + .expect("shell pipe buffer lock") + .extend_from_slice(&chunk[..n]); + } +} + +const POST_KILL_PIPE_JOIN: Duration = Duration::from_secs(3); +const POST_EXIT_PIPE_JOIN: Duration = Duration::from_secs(6); +const POST_TIMEOUT_CLEANUP: Duration = Duration::from_secs(5); + +const PIPE_TRUNCATION_MARKER: &[u8] = b"\n[output truncated: pipe held open past deadline]\n"; + +async fn join_shell_pipe(name: &str, task: JoinHandle>) -> Result<()> { + task.await + .map_err(|e| anyhow!("bash: {name} reader task failed: {e}"))? + .map_err(|e| anyhow!("bash: {name} read failed: {e}")) +} + +fn take_pipe_buffer(sink: &Arc>>, truncated: bool) -> Vec { + let mut output = sink.lock().expect("shell pipe buffer lock").clone(); + if truncated { + output.extend_from_slice(PIPE_TRUNCATION_MARKER); + } + output +} + +async fn join_shell_pipes( + stdout_task: JoinHandle>, + stderr_task: JoinHandle>, + stdout_abort: AbortHandle, + stderr_abort: AbortHandle, + stdout_buf: Arc>>, + stderr_buf: Arc>>, + limit: Option, +) -> Result<(Vec, Vec)> { + let join_both = async { + join_shell_pipe("stdout", stdout_task).await?; + join_shell_pipe("stderr", stderr_task).await?; + Ok::<(), anyhow::Error>(()) + }; + match limit { + None => { + join_both.await?; + Ok(( + take_pipe_buffer(&stdout_buf, false), + take_pipe_buffer(&stderr_buf, false), + )) + } + Some(deadline) => match tokio::time::timeout(deadline, join_both).await { + Ok(result) => { + result?; + Ok(( + take_pipe_buffer(&stdout_buf, false), + take_pipe_buffer(&stderr_buf, false), + )) + } + Err(_) => { + stdout_abort.abort(); + stderr_abort.abort(); + Ok(( + take_pipe_buffer(&stdout_buf, true), + take_pipe_buffer(&stderr_buf, true), + )) + } + }, + } +} + +async fn supervise_shell_process( + mut child: Box, + stdout_task: JoinHandle>, + stderr_task: JoinHandle>, + stdout_buf: Arc>>, + stderr_buf: Arc>>, + mut cancel_rx: oneshot::Receiver<()>, +) -> Result { + enum ProcessOutcome { + Completed(ExitStatus), + Cancelled, + } + + let stdout_abort = stdout_task.abort_handle(); + let stderr_abort = stderr_task.abort_handle(); + + let process_outcome = tokio::select! { + status = child.wait() => match status { + Ok(status) => ProcessOutcome::Completed(status), + Err(wait_error) => { + return match Box::into_pin(child.kill()).await { + Ok(()) => Err(anyhow!("bash: execution error: {wait_error}")), + Err(cleanup_error) => Err(anyhow!( + "bash: execution error: {wait_error}; process-tree cleanup failed: {cleanup_error}" + )), + }; + } + }, + _ = &mut cancel_rx => { + Box::into_pin(child.kill()) + .await + .map_err(|e| anyhow!("bash: process-tree cleanup failed: {e}"))?; + ProcessOutcome::Cancelled + } + }; + + let pipe_limit = match process_outcome { + ProcessOutcome::Completed(_) => Some(POST_EXIT_PIPE_JOIN), + ProcessOutcome::Cancelled => Some(POST_KILL_PIPE_JOIN), + }; + let (stdout, stderr) = join_shell_pipes( + stdout_task, + stderr_task, + stdout_abort, + stderr_abort, + stdout_buf, + stderr_buf, + pipe_limit, + ) + .await?; + + Ok(match process_outcome { + ProcessOutcome::Completed(status) => ShellSupervisorOutcome::Completed(ShellExecution { + status, + stdout, + stderr, + }), + ProcessOutcome::Cancelled => ShellSupervisorOutcome::Cancelled, + }) +} + +/// Spawn the platform shell inside a dedicated Unix session or Windows Job Object, then transfer +/// process-tree ownership to a supervisor that survives cancellation of the calling future. +fn spawn_shell_command( + command: &str, + working_dir: &Path, + env: &HashMap, +) -> Result { + let mut cmd = platform_shell_command(command)?; + cmd.current_dir(working_dir) + .env_clear() + .envs(env) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let mut wrapped = CommandWrap::from(cmd); + wrapped.wrap(KillOnDrop); + #[cfg(unix)] + wrapped.wrap(ProcessSession); + #[cfg(windows)] + wrapped.wrap(JobObject); + + let mut child = wrapped + .spawn() + .map_err(|e| anyhow!("bash: spawn failed: {e}"))?; + let stdout = child + .stdout() + .take() + .ok_or_else(|| anyhow!("bash: stdout pipe unavailable"))?; + let stderr = child + .stderr() + .take() + .ok_or_else(|| anyhow!("bash: stderr pipe unavailable"))?; + + let stdout_buf = Arc::new(Mutex::new(Vec::new())); + let stderr_buf = Arc::new(Mutex::new(Vec::new())); + let stdout_task = tokio::spawn(read_shell_pipe_into(stdout, Arc::clone(&stdout_buf))); + let stderr_task = tokio::spawn(read_shell_pipe_into(stderr, Arc::clone(&stderr_buf))); + let (cancel_tx, cancel_rx) = oneshot::channel(); + let (outcome_tx, outcome_rx) = oneshot::channel(); + tokio::spawn(async move { + let outcome = supervise_shell_process( + child, + stdout_task, + stderr_task, + stdout_buf, + stderr_buf, + cancel_rx, + ) + .await; + let _ = outcome_tx.send(outcome); + }); + + Ok(ShellCommandController { + cancel_tx: Some(cancel_tx), + outcome_rx, + }) +} + +/// Run a supervised shell command. Timeout is an orderly controller cancellation: kill and reap +/// must complete before this function returns. +async fn run_shell_command( + command: &str, + working_dir: &Path, + timeout_duration: Duration, + env: &HashMap, +) -> Result { + let mut controller = spawn_shell_command(command, working_dir, env)?; + + match tokio::time::timeout(timeout_duration, controller.wait()).await { + Ok(Ok(ShellSupervisorOutcome::Completed(execution))) => Ok(format_shell_output( + execution.status, + &execution.stdout, + &execution.stderr, + )), + Ok(Ok(ShellSupervisorOutcome::Cancelled)) => Err(anyhow!("bash: command cancelled")), + Ok(Err(e)) => Err(e), + Err(_) => match tokio::time::timeout(POST_TIMEOUT_CLEANUP, controller.cancel_and_wait()) + .await + { + Ok(Ok(ShellSupervisorOutcome::Cancelled | ShellSupervisorOutcome::Completed(_))) => { + Err(anyhow!( + "bash: command timed out after {}s", + timeout_duration.as_secs() + )) + } + Ok(Err(e)) => Err(anyhow!( + "bash: command timed out after {}s; {e}", + timeout_duration.as_secs() + )), + Err(_) => Err(anyhow!( + "bash: command timed out after {}s; process-tree cleanup exceeded the post-kill deadline", + timeout_duration.as_secs() + )), + }, + } +} + +/// Execute a shell command with process-tree isolation and env filtering. async fn tool_bash(input: &Value, working_dir: &Path) -> Result { let command = input .get("command") @@ -191,13 +578,8 @@ async fn tool_bash(input: &Value, working_dir: &Path) -> Result { let cmd_working_dir = input .get("working_dir") .and_then(|v| v.as_str()) - .map(|p| { - if Path::new(p).is_absolute() { - PathBuf::from(p) - } else { - working_dir.join(p) - } - }) + .map(|p| validate_path(p, working_dir)) + .transpose()? .unwrap_or_else(|| working_dir.to_path_buf()); let timeout_secs = std::env::var("OPENAB_AGENT_TIMEOUT_SECS") @@ -215,78 +597,13 @@ async fn tool_bash(input: &Value, working_dir: &Path) -> Result { let env = build_env(&env_allow); debug!("bash: executing '{}' in {:?}", command, cmd_working_dir); - - let mut cmd = Command::new("sh"); - cmd.arg("-c") - .arg(command) - .current_dir(&cmd_working_dir) - .env_clear() - .envs(&env) - .stdin(std::process::Stdio::null()) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()); - - // Create new process group on Unix for clean cleanup - #[cfg(unix)] - unsafe { - #[allow(unused_imports)] - use std::os::unix::process::CommandExt; - cmd.pre_exec(|| { - if libc::setsid() == -1 { - return Err(std::io::Error::last_os_error()); - } - Ok(()) - }); - } - - let child = cmd - .spawn() - .map_err(|e| anyhow!("bash: spawn failed: {e}"))?; - - // Capture pid before wait_with_output takes ownership - #[cfg(unix)] - let child_pid = child.id(); - - let result = tokio::time::timeout( - std::time::Duration::from_secs(timeout_secs), - child.wait_with_output(), + run_shell_command( + command, + &cmd_working_dir, + Duration::from_secs(timeout_secs), + &env, ) - .await; - - match result { - Ok(Ok(output)) => { - let stdout = String::from_utf8_lossy(&output.stdout); - let stderr = String::from_utf8_lossy(&output.stderr); - let code = output.status.code().unwrap_or(-1); - - let mut result = String::new(); - if !stdout.is_empty() { - result.push_str(&stdout); - } - if !stderr.is_empty() { - if !result.is_empty() { - result.push('\n'); - } - result.push_str("[stderr]\n"); - result.push_str(&stderr); - } - if code != 0 { - result.push_str(&format!("\n[exit code: {code}]")); - } - Ok(result) - } - Ok(Err(e)) => Err(anyhow!("bash: execution error: {e}")), - Err(_) => { - // Timeout — kill the process group - #[cfg(unix)] - if let Some(pid) = child_pid { - unsafe { - libc::kill(-(pid as i32), libc::SIGKILL); - } - } - Err(anyhow!("bash: command timed out after {timeout_secs}s")) - } - } + .await } /// Return tool definitions for the LLM. @@ -366,6 +683,23 @@ mod tests { assert!(result.unwrap_err().to_string().contains("path traversal")); } + #[test] + fn test_powershell_encoded_command_preserves_unicode_and_quotes() { + let command = "Write-Output 'héllo 世界'; Write-Output \"quoted\""; + let encoded = powershell_encoded_command(command); + let bytes = base64::engine::general_purpose::STANDARD + .decode(encoded) + .unwrap(); + assert_eq!(bytes.len() % 2, 0); + let utf16: Vec = bytes + .chunks_exact(2) + .map(|pair| u16::from_le_bytes([pair[0], pair[1]])) + .collect(); + let decoded = String::from_utf16(&utf16).unwrap(); + assert!(decoded.ends_with(command)); + assert!(decoded.contains("UTF8Encoding")); + } + #[test] #[ignore] // Integration test: filesystem access fn test_tool_write_and_read() { @@ -413,8 +747,8 @@ mod tests { assert!(result.contains("subdir/")); } + #[cfg(unix)] #[tokio::test] - #[ignore] // Integration test: subprocess execution async fn test_tool_bash_simple() { let tmp = TempDir::new().unwrap(); let input = json!({ "command": "echo hello" }); @@ -422,6 +756,249 @@ mod tests { assert_eq!(result.trim(), "hello"); } + #[cfg(windows)] + #[tokio::test] + async fn test_tool_bash_windows_unicode_and_quoting() { + let tmp = TempDir::new().unwrap(); + let input = + json!({ "command": "Write-Output 'héllo 世界'; Write-Output \"quoted value\"" }); + let result = tool_bash(&input, tmp.path()).await.unwrap(); + assert!(result.contains("héllo 世界")); + assert!(result.contains("quoted value")); + } + + #[cfg(windows)] + async fn warm_windows_shell(working_dir: &Path) { + let result = run_shell_command( + "Write-Output 'warm'", + working_dir, + Duration::from_secs(60), + &build_env(&[]), + ) + .await + .expect("Windows PowerShell warmup failed"); + assert!(result.contains("warm")); + } + + #[cfg(unix)] + #[tokio::test] + async fn test_unix_future_drop_requests_supervised_process_tree_cleanup() { + let tmp = TempDir::new().unwrap(); + let command = concat!( + "(sleep 2; printf escaped > drop-orphan.txt) & ", + "printf ready > drop-ready.txt; ", + "sleep 120" + ); + let task_dir = tmp.path().to_path_buf(); + let task = tokio::spawn(async move { + run_shell_command( + command, + &task_dir, + Duration::from_secs(120), + &build_env(&[]), + ) + .await + }); + + tokio::time::timeout(Duration::from_secs(5), async { + while !tmp.path().join("drop-ready.txt").exists() { + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .expect("parent shell did not start"); + task.abort(); + let _ = task.await; + + tokio::time::sleep(Duration::from_secs(3)).await; + assert!( + !tmp.path().join("drop-orphan.txt").exists(), + "descendant escaped supervised cleanup when the shell future was dropped" + ); + } + + #[cfg(windows)] + #[tokio::test] + async fn test_windows_timeout_kills_descendant_process() { + let tmp = TempDir::new().unwrap(); + warm_windows_shell(tmp.path()).await; + let descendant = powershell_encoded_command( + "Start-Sleep -Seconds 10; [IO.File]::WriteAllText('orphan.txt', 'escaped')", + ); + let command = format!( + concat!( + "$child = Join-Path $PSHOME 'powershell.exe'; ", + "Start-Process -FilePath $child -ArgumentList @(", + "'-NoLogo','-NoProfile','-NonInteractive','-EncodedCommand','{descendant}'); ", + "[IO.File]::WriteAllText('ready.txt', 'ready'); ", + "Start-Sleep -Seconds 120" + ), + descendant = descendant + ); + let error = run_shell_command( + &command, + tmp.path(), + Duration::from_secs(5), + &build_env(&[]), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("timed out")); + assert!(tmp.path().join("ready.txt").exists()); + + tokio::time::sleep(Duration::from_secs(12)).await; + assert!( + !tmp.path().join("orphan.txt").exists(), + "descendant escaped the Windows Job Object" + ); + } + + #[cfg(windows)] + #[tokio::test] + async fn test_windows_future_drop_kills_descendant_process() { + let tmp = TempDir::new().unwrap(); + warm_windows_shell(tmp.path()).await; + let descendant = powershell_encoded_command( + "Start-Sleep -Seconds 10; [IO.File]::WriteAllText('drop-orphan.txt', 'escaped')", + ); + let command = format!( + concat!( + "$child = Join-Path $PSHOME 'powershell.exe'; ", + "Start-Process -FilePath $child -ArgumentList @(", + "'-NoLogo','-NoProfile','-NonInteractive','-EncodedCommand','{descendant}'); ", + "[IO.File]::WriteAllText('drop-ready.txt', 'ready'); ", + "Start-Sleep -Seconds 120" + ), + descendant = descendant + ); + let task_dir = tmp.path().to_path_buf(); + let task = tokio::spawn(async move { + run_shell_command( + &command, + &task_dir, + Duration::from_secs(120), + &build_env(&[]), + ) + .await + }); + + tokio::time::timeout(Duration::from_secs(10), async { + while !tmp.path().join("drop-ready.txt").exists() { + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("parent shell did not start"); + task.abort(); + let _ = task.await; + + tokio::time::sleep(Duration::from_secs(12)).await; + assert!( + !tmp.path().join("drop-orphan.txt").exists(), + "descendant escaped cleanup when the shell future was dropped" + ); + } + + #[cfg(windows)] + #[tokio::test] + async fn test_windows_controller_cancel_waits_for_descendant_cleanup() { + let tmp = TempDir::new().unwrap(); + warm_windows_shell(tmp.path()).await; + let descendant = powershell_encoded_command( + "Start-Sleep -Seconds 10; [IO.File]::WriteAllText('cancel-orphan.txt', 'escaped')", + ); + let command = format!( + concat!( + "$child = Join-Path $PSHOME 'powershell.exe'; ", + "Start-Process -FilePath $child -ArgumentList @(", + "'-NoLogo','-NoProfile','-NonInteractive','-EncodedCommand','{descendant}'); ", + "[IO.File]::WriteAllText('cancel-ready.txt', 'ready'); ", + "Start-Sleep -Seconds 120" + ), + descendant = descendant + ); + let mut controller = spawn_shell_command(&command, tmp.path(), &build_env(&[])).unwrap(); + + tokio::time::timeout(Duration::from_secs(10), async { + while !tmp.path().join("cancel-ready.txt").exists() { + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("parent shell did not start"); + let outcome = controller.cancel_and_wait().await.unwrap(); + assert!(matches!(outcome, ShellSupervisorOutcome::Cancelled)); + + tokio::time::sleep(Duration::from_secs(12)).await; + assert!( + !tmp.path().join("cancel-orphan.txt").exists(), + "controller returned before the Windows process tree was cleaned up" + ); + } + + #[cfg(unix)] + #[tokio::test] + async fn test_unix_completed_returns_when_setsid_descendant_holds_pipes() { + let tmp = TempDir::new().unwrap(); + let command = "setsid sh -c 'echo $$ > escaped.pid; exec sleep 30' & exit 0"; + let started = std::time::Instant::now(); + let result = tokio::time::timeout( + Duration::from_secs(8), + run_shell_command( + command, + tmp.path(), + Duration::from_secs(10), + &build_env(&[]), + ), + ) + .await + .expect("completed path hung waiting for escaped descendant pipes"); + assert!( + result.is_ok(), + "normal exit must not hang on leftover pipes: {result:?}" + ); + assert!( + started.elapsed() < Duration::from_secs(8), + "completed-path pipe join must be bounded" + ); + + if let Ok(raw) = std::fs::read_to_string(tmp.path().join("escaped.pid")) { + if let Ok(pid) = raw.trim().parse::() { + let _ = std::process::Command::new("kill") + .args(["-9", &pid.to_string()]) + .status(); + } + } + } + + #[cfg(unix)] + #[tokio::test] + async fn test_unix_timeout_returns_when_setsid_descendant_holds_pipes() { + let tmp = TempDir::new().unwrap(); + let command = "setsid sh -c 'echo $$ > escaped.pid; exec sleep 30' & exec sleep 120"; + let started = std::time::Instant::now(); + let error = tokio::time::timeout( + Duration::from_secs(8), + run_shell_command(command, tmp.path(), Duration::from_secs(1), &build_env(&[])), + ) + .await + .expect("timeout path hung waiting for escaped descendant pipes") + .unwrap_err(); + assert!(error.to_string().contains("timed out")); + assert!( + started.elapsed() < Duration::from_secs(8), + "post-kill pipe join must be bounded" + ); + + if let Ok(raw) = std::fs::read_to_string(tmp.path().join("escaped.pid")) { + if let Ok(pid) = raw.trim().parse::() { + let _ = std::process::Command::new("kill") + .args(["-9", &pid.to_string()]) + .status(); + } + } + } + #[tokio::test] #[ignore] // Integration test: subprocess execution async fn test_tool_bash_env_filtered() { diff --git a/operator/src/ingress.rs b/operator/src/ingress.rs index 57761ea3d..3da3c684c 100644 --- a/operator/src/ingress.rs +++ b/operator/src/ingress.rs @@ -1176,7 +1176,7 @@ mod tests { fn cloud_map_service_id_parses_from_arn() { assert_eq!( cloud_map_service_id_from_arn( - "arn:aws:servicediscovery:us-east-1:903779448426:service/srv-abc123" + "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-abc123" ), Some("srv-abc123".to_string()) ); diff --git a/operator/src/secrets.rs b/operator/src/secrets.rs index 5c17be05a..392063a48 100644 --- a/operator/src/secrets.rs +++ b/operator/src/secrets.rs @@ -165,20 +165,20 @@ mod tests { // (which used to hand this straight to GetSecretValue and fail, // since that API has no knowledge of the trailing ECS suffix). let (base, key) = split_ecs_json_key_suffix( - "arn:aws:secretsmanager:us-east-1:903779448426:secret:oab/telegram/pahudxbot-AC80TP:TELEGRAM_BOT_TOKEN::", + "arn:aws:secretsmanager:us-east-1:123456789012:secret:example/telegram/bot-AbCdEf:TELEGRAM_BOT_TOKEN::", ) .unwrap(); - assert_eq!(base, "arn:aws:secretsmanager:us-east-1:903779448426:secret:oab/telegram/pahudxbot-AC80TP"); + assert_eq!(base, "arn:aws:secretsmanager:us-east-1:123456789012:secret:example/telegram/bot-AbCdEf"); assert_eq!(key, Some("TELEGRAM_BOT_TOKEN")); } #[test] fn split_ecs_json_key_suffix_unchanged_for_plain_arn() { let (base, key) = split_ecs_json_key_suffix( - "arn:aws:secretsmanager:us-east-1:903779448426:secret:oab/telegram/pahudxbot-AC80TP", + "arn:aws:secretsmanager:us-east-1:123456789012:secret:example/telegram/bot-AbCdEf", ) .unwrap(); - assert_eq!(base, "arn:aws:secretsmanager:us-east-1:903779448426:secret:oab/telegram/pahudxbot-AC80TP"); + assert_eq!(base, "arn:aws:secretsmanager:us-east-1:123456789012:secret:example/telegram/bot-AbCdEf"); assert_eq!(key, None); } @@ -196,8 +196,8 @@ mod tests { // json-key="mysecret" — "mysecret" here is part of the secret // name/base ARN, not a suffix field. let (base, key) = - split_ecs_json_key_suffix("arn:aws:secretsmanager:us-east-1:903779448426:secret:mysecret::").unwrap(); - assert_eq!(base, "arn:aws:secretsmanager:us-east-1:903779448426:secret:mysecret"); + split_ecs_json_key_suffix("arn:aws:secretsmanager:us-east-1:123456789012:secret:mysecret::").unwrap(); + assert_eq!(base, "arn:aws:secretsmanager:us-east-1:123456789012:secret:mysecret"); assert_eq!(key, None); } @@ -207,7 +207,7 @@ mod tests { // scope for in-process resolution — fail closed with a clear error // instead of silently mishandling it. let err = split_ecs_json_key_suffix( - "arn:aws:secretsmanager:us-east-1:903779448426:secret:appauthexample-AbCdEf::AWSPREVIOUS:", + "arn:aws:secretsmanager:us-east-1:123456789012:secret:appauthexample-AbCdEf::AWSPREVIOUS:", ) .unwrap_err(); assert!(err.to_string().contains("version")); @@ -216,7 +216,7 @@ mod tests { #[test] fn split_ecs_json_key_suffix_rejects_version_id() { let err = split_ecs_json_key_suffix( - "arn:aws:secretsmanager:us-east-1:903779448426:secret:appauthexample-AbCdEf:::9d4cb84b-ad69-40c0-a0ab-cead3EXAMPLE", + "arn:aws:secretsmanager:us-east-1:123456789012:secret:appauthexample-AbCdEf:::9d4cb84b-ad69-40c0-a0ab-cead3EXAMPLE", ) .unwrap_err(); assert!(err.to_string().contains("version")); @@ -225,7 +225,7 @@ mod tests { #[test] fn split_ecs_json_key_suffix_rejects_key_and_version_stage_together() { let err = split_ecs_json_key_suffix( - "arn:aws:secretsmanager:us-east-1:903779448426:secret:appauthexample-AbCdEf:username1:AWSPREVIOUS:", + "arn:aws:secretsmanager:us-east-1:123456789012:secret:appauthexample-AbCdEf:username1:AWSPREVIOUS:", ) .unwrap_err(); assert!(err.to_string().contains("version")); @@ -233,16 +233,16 @@ mod tests { #[test] fn parse_aws_sm_uri_extracts_id_and_key() { - let (id, key) = parse_aws_sm_uri("aws-sm://oab/telegram/pahudxbot#TELEGRAM_BOT_TOKEN") + let (id, key) = parse_aws_sm_uri("aws-sm://example/telegram/bot#TELEGRAM_BOT_TOKEN") .unwrap() .unwrap(); - assert_eq!(id, "oab/telegram/pahudxbot"); + assert_eq!(id, "example/telegram/bot"); assert_eq!(key, "TELEGRAM_BOT_TOKEN"); } #[test] fn parse_aws_sm_uri_rejects_missing_hash() { - assert!(parse_aws_sm_uri("aws-sm://oab/telegram/pahudxbot").unwrap().is_err()); + assert!(parse_aws_sm_uri("aws-sm://example/telegram/bot").unwrap().is_err()); } #[test] diff --git a/scripts/windows-acp-native-canary.ps1 b/scripts/windows-acp-native-canary.ps1 new file mode 100644 index 000000000..3922c1678 --- /dev/null +++ b/scripts/windows-acp-native-canary.ps1 @@ -0,0 +1,475 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$OpenAbExe, + + [Parameter(Mandatory = $true)] + [string]$AgentExe, + + [string]$WorkDir = (Get-Location).Path +) + +$ErrorActionPreference = "Stop" +Set-StrictMode -Version Latest + +function Assert-Canary { + param([bool]$Condition, [string]$Message) + if (-not $Condition) { + throw "CANARY FAIL: $Message" + } +} + +function Redact-CanaryText { + param([AllowNull()][string]$Text) + if ($null -eq $Text) { + return "" + } + return $Text ` + -replace '(?i)Bearer\s+[A-Za-z0-9._~+/=-]+', 'Bearer ' ` + -replace '(?i)(sk-ant-|sk-)[A-Za-z0-9._~-]+', '' ` + -replace '(?i)(api[_-]?key\s*[:=]\s*)[^,\s}]+', '$1' +} + +function Format-CanaryFrame { + param([AllowNull()][object]$Frame) + if ($null -eq $Frame) { + return "" + } + if ($Frame -is [string]) { + return Redact-CanaryText ([string]$Frame) + } + try { + return Redact-CanaryText ($Frame | ConvertTo-Json -Compress -Depth 20) + } + catch { + return Redact-CanaryText ([string]$Frame) + } +} + +function Assert-RpcSuccess { + param( + [AllowNull()][object]$Frame, + [int]$ExpectedId, + [string]$Method + ) + Assert-Canary -Condition ($null -ne $Frame) -Message "$Method returned no JSON-RPC frame" + $properties = @($Frame.PSObject.Properties.Name) + if (-not ($properties -contains "id") -or [int]$Frame.id -ne $ExpectedId) { + throw "CANARY FAIL: $Method response id mismatch; frame=$(Format-CanaryFrame $Frame)" + } + if ($properties -contains "error") { + throw "CANARY FAIL: $Method returned JSON-RPC error; frame=$(Format-CanaryFrame $Frame)" + } + if (-not ($properties -contains "result")) { + throw "CANARY FAIL: $Method response has no result; frame=$(Format-CanaryFrame $Frame)" + } + return $Frame.result +} + +function Add-BaselineEnvironment { + param([System.Diagnostics.ProcessStartInfo]$StartInfo) + $StartInfo.EnvironmentVariables.Clear() + foreach ($name in @( + "SystemRoot", "SystemDrive", "PATH", "PATHEXT", "TEMP", "TMP", + "USERPROFILE", "USERNAME", "HOME", "APPDATA", "LOCALAPPDATA" + )) { + $value = [Environment]::GetEnvironmentVariable($name) + if (-not [string]::IsNullOrWhiteSpace($value)) { + $StartInfo.EnvironmentVariables[$name] = $value + } + } +} + +function New-RedirectedProcess { + param( + [string]$FilePath, + [string[]]$ArgumentList, + [hashtable]$Environment, + [string]$WorkingDirectory + ) + $start = [System.Diagnostics.ProcessStartInfo]::new() + $start.FileName = $FilePath + $start.WorkingDirectory = $WorkingDirectory + $start.UseShellExecute = $false + $start.CreateNoWindow = $true + $start.RedirectStandardInput = $true + $start.RedirectStandardOutput = $true + $start.RedirectStandardError = $true + Add-BaselineEnvironment $start + # ProcessStartInfo.ArgumentList is unavailable in Windows PowerShell 5.1. + # Windows paths cannot contain a literal double quote, so quoting every + # argument gives one implementation that works on both 5.1 and PowerShell 7. + Assert-Canary -Condition (-not ($ArgumentList | Where-Object { $_.Contains('"') })) -Message "argument contains a double quote" + $start.Arguments = (($ArgumentList | ForEach-Object { '"' + $_ + '"' }) -join " ") + foreach ($entry in $Environment.GetEnumerator()) { + $start.EnvironmentVariables[$entry.Key] = [string]$entry.Value + } + $process = [System.Diagnostics.Process]::new() + $process.StartInfo = $start + Assert-Canary -Condition ($process.Start()) -Message "could not start $FilePath" + return $process +} + +function Send-StdioJson { + param([System.Diagnostics.Process]$Process, [hashtable]$Frame) + $line = $Frame | ConvertTo-Json -Compress -Depth 20 + $Process.StandardInput.WriteLine($line) + $Process.StandardInput.Flush() +} + +function Receive-StdioJson { + param([System.Diagnostics.Process]$Process, [int]$TimeoutSeconds = 15) + $read = $Process.StandardOutput.ReadLineAsync() + Assert-Canary -Condition ($read.Wait([TimeSpan]::FromSeconds($TimeoutSeconds))) -Message "stdio response timed out" + Assert-Canary -Condition ($null -ne $read.Result) -Message "stdio closed before response" + $raw = [string]$read.Result + try { + return $raw | ConvertFrom-Json -ErrorAction Stop + } + catch { + throw "CANARY FAIL: stdio emitted invalid JSON; frame=$(Redact-CanaryText $raw)" + } +} + +function Send-WebSocketJson { + param([System.Net.WebSockets.ClientWebSocket]$Socket, [hashtable]$Frame) + $bytes = [Text.Encoding]::UTF8.GetBytes(($Frame | ConvertTo-Json -Compress -Depth 20)) + $segment = [ArraySegment[byte]]::new($bytes) + $Socket.SendAsync( + $segment, + [System.Net.WebSockets.WebSocketMessageType]::Text, + $true, + [Threading.CancellationToken]::None + ).GetAwaiter().GetResult() +} + +function Receive-WebSocketJson { + param([System.Net.WebSockets.ClientWebSocket]$Socket, [int]$TimeoutSeconds = 15) + $buffer = [byte[]]::new(65536) + $stream = [System.IO.MemoryStream]::new() + $cancel = [Threading.CancellationTokenSource]::new([TimeSpan]::FromSeconds($TimeoutSeconds)) + try { + do { + $segment = [ArraySegment[byte]]::new($buffer) + $result = $Socket.ReceiveAsync($segment, $cancel.Token).GetAwaiter().GetResult() + Assert-Canary -Condition ( + $result.MessageType -ne [System.Net.WebSockets.WebSocketMessageType]::Close + ) -Message "WebSocket closed before response" + $stream.Write($buffer, 0, $result.Count) + } while (-not $result.EndOfMessage) + return [Text.Encoding]::UTF8.GetString($stream.ToArray()) | ConvertFrom-Json + } + finally { + $cancel.Dispose() + $stream.Dispose() + } +} + +function Receive-WebSocketResponse { + param( + [System.Net.WebSockets.ClientWebSocket]$Socket, + [int]$Id, + [int]$TimeoutSeconds = 30 + ) + $watch = [Diagnostics.Stopwatch]::StartNew() + while ($watch.Elapsed.TotalSeconds -lt $TimeoutSeconds) { + $remaining = [Math]::Max(1, $TimeoutSeconds - [int]$watch.Elapsed.TotalSeconds) + $frame = Receive-WebSocketJson $Socket $remaining + if (($frame.PSObject.Properties.Name -contains "id") -and [int]$frame.id -eq $Id) { + return $frame + } + } + throw "CANARY FAIL: no WebSocket response for id $Id" +} + +function Get-CanaryUpdateText { + param([AllowNull()][object]$Frame) + if ($null -eq $Frame) { + return "" + } + try { + if ([string]$Frame.method -ne "session/update") { + return "" + } + $text = [string]$Frame.params.update.content.text + if ([string]::IsNullOrWhiteSpace($text)) { + return "" + } + return $text + } + catch { + return "" + } +} + +function Receive-WebSocketPromptSettlement { + param( + [System.Net.WebSockets.ClientWebSocket]$Socket, + [int]$Id, + [int]$TimeoutSeconds = 45 + ) + $updates = New-Object System.Text.StringBuilder + $watch = [Diagnostics.Stopwatch]::StartNew() + while ($watch.Elapsed.TotalSeconds -lt $TimeoutSeconds) { + $remaining = [Math]::Max(1, $TimeoutSeconds - [int]$watch.Elapsed.TotalSeconds) + $frame = Receive-WebSocketJson $Socket $remaining + $chunk = Get-CanaryUpdateText $frame + if (-not [string]::IsNullOrWhiteSpace($chunk)) { + [void]$updates.Append($chunk) + continue + } + if (($frame.PSObject.Properties.Name -contains "id") -and [int]$frame.id -eq $Id) { + return [pscustomobject]@{ + Frame = $frame + UpdateText = $updates.ToString() + } + } + } + throw "CANARY FAIL: no WebSocket response for id $Id; updates=$(Redact-CanaryText $updates.ToString())" +} + +function Test-NoProviderFailureShape { + param( + [AllowNull()][object]$Frame, + [AllowNull()][string]$UpdateText + ) + $blob = New-Object System.Text.StringBuilder + if ($null -ne $Frame -and @($Frame.PSObject.Properties.Name) -contains "error") { + [void]$blob.AppendLine((Format-CanaryFrame $Frame.error)) + } + if (-not [string]::IsNullOrWhiteSpace($UpdateText)) { + [void]$blob.AppendLine($UpdateText) + } + $joined = $blob.ToString() + if ([string]::IsNullOrWhiteSpace($joined)) { + return $false + } + # /acp settles a failed backend turn as result {stopReason} plus streamed + # text ("⚠️ agent error: LLM error: ..."). JSON-RPC error is sufficient + # but not required. A successful model completion without these markers + # is a canary failure. + return [bool]($joined -match '(?i)(⚠️|LLM error|API error|agent error:|error sending request|error trying to connect|connection refused|invalid.?api.?key|authentication|credential|401|403|timed out waiting|_\(no response\)_|backend configuration issue|provider/model/auth)') +} + +function Wait-TcpPort { + param([int]$Port, [int]$TimeoutSeconds = 20) + $watch = [Diagnostics.Stopwatch]::StartNew() + while ($watch.Elapsed.TotalSeconds -lt $TimeoutSeconds) { + $client = [Net.Sockets.TcpClient]::new() + try { + $connect = $client.ConnectAsync("127.0.0.1", $Port) + if ($connect.Wait(250) -and $client.Connected) { + return + } + } + catch { + } + finally { + $client.Dispose() + } + Start-Sleep -Milliseconds 100 + } + throw "CANARY FAIL: OpenAB did not listen on port $Port" +} + +function Test-ProcessGone { + param([int]$Id) + return $null -eq (Get-Process -Id $Id -ErrorAction SilentlyContinue) +} + +$OpenAbExe = (Resolve-Path -LiteralPath $OpenAbExe).Path +$AgentExe = (Resolve-Path -LiteralPath $AgentExe).Path +$WorkDir = (Resolve-Path -LiteralPath $WorkDir).Path + +$tempRoot = Join-Path ([IO.Path]::GetTempPath()) ("openab-windows-canary-" + [Guid]::NewGuid()) +$homeDir = Join-Path $tempRoot "home" +$configPath = Join-Path $tempRoot "config.toml" +New-Item -ItemType Directory -Path $homeDir -Force | Out-Null + +$agent = $null +$root = $null +$socket = $null +$rootStdout = $null +$rootStderr = $null +$agentPid = $null + +try { + & $AgentExe --help | Out-Null + Assert-Canary -Condition ($LASTEXITCODE -eq 0) -Message "openab-agent.exe --help" + & $OpenAbExe --help | Out-Null + Assert-Canary -Condition ($LASTEXITCODE -eq 0) -Message "openab.exe --help" + Write-Host "[PASS] both binaries answer --help" + + $isolatedAgentEnv = @{ + HOME = $homeDir + USERPROFILE = $homeDir + OPENAB_AGENT_PROVIDER = "anthropic" + OPENAB_AGENT_MODEL = "claude-sonnet-4-6" + ANTHROPIC_API_KEY = "ci-non-secret-no-provider-call" + HTTPS_PROXY = "http://127.0.0.1:9" + HTTP_PROXY = "http://127.0.0.1:9" + ALL_PROXY = "http://127.0.0.1:9" + NO_PROXY = "" + } + $agent = New-RedirectedProcess -FilePath $AgentExe -ArgumentList ([string[]]@()) -Environment $isolatedAgentEnv -WorkingDirectory $WorkDir + Send-StdioJson $agent @{ + jsonrpc = "2.0" + id = 1 + method = "initialize" + params = @{ protocolVersion = 1; clientCapabilities = @{}; clientInfo = @{ name = "windows-canary"; version = "1" } } + } + $initialize = Receive-StdioJson $agent + $initializeResult = Assert-RpcSuccess $initialize 1 "agent initialize" + Assert-Canary -Condition ($initializeResult.agentInfo.name -eq "openab-agent") -Message "agent initialize agentInfo" + + Send-StdioJson $agent @{ + jsonrpc = "2.0" + id = 2 + method = "session/new" + params = @{ cwd = $WorkDir; mcpServers = @() } + } + $newSession = Receive-StdioJson $agent + $newSessionResult = Assert-RpcSuccess $newSession 2 "agent session/new" + Assert-Canary -Condition (-not [string]::IsNullOrWhiteSpace($newSessionResult.sessionId)) -Message "agent session/new sessionId" + $agent.StandardInput.Close() + Assert-Canary -Condition ($agent.WaitForExit(10000)) -Message "agent did not stop after stdin EOF" + Assert-Canary -Condition ($agent.ExitCode -eq 0) -Message "agent returned non-zero after stdio smoke" + Write-Host "[PASS] openab-agent ACP stdio initialize + session/new + EOF shutdown" + + $listener = [Net.Sockets.TcpListener]::new([Net.IPAddress]::Loopback, 0) + $listener.Start() + $port = ([Net.IPEndPoint]$listener.LocalEndpoint).Port + $listener.Stop() + + $agentToml = $AgentExe.Replace("'", "''") + $workToml = $WorkDir.Replace("'", "''") + $config = @" +[agent] +command = '$agentToml' +working_dir = '$workToml' +env = { OPENAB_AGENT_PROVIDER = "anthropic", OPENAB_AGENT_MODEL = "claude-sonnet-4-6", ANTHROPIC_API_KEY = "ci-non-secret-no-provider-call", HTTPS_PROXY = "http://127.0.0.1:9", HTTP_PROXY = "http://127.0.0.1:9", ALL_PROXY = "http://127.0.0.1:9", NO_PROXY = "" } + +[pool] +max_sessions = 2 +"@ + [IO.File]::WriteAllText($configPath, $config, [Text.UTF8Encoding]::new($false)) + + $rootEnv = @{ + HOME = $homeDir + USERPROFILE = $homeDir + OPENAB_ACP_ENABLED = "1" + GATEWAY_LISTEN = "127.0.0.1:$port" + GATEWAY_ALLOW_ALL_USERS = "true" + RUST_LOG = "info" + } + $root = New-RedirectedProcess -FilePath $OpenAbExe -ArgumentList @("run", "-c", $configPath) -Environment $rootEnv -WorkingDirectory $WorkDir + $rootStdout = $root.StandardOutput.ReadToEndAsync() + $rootStderr = $root.StandardError.ReadToEndAsync() + Wait-TcpPort $port + + $socket = [System.Net.WebSockets.ClientWebSocket]::new() + $socket.ConnectAsync( + [Uri]"ws://127.0.0.1:$port/acp", + [Threading.CancellationToken]::None + ).GetAwaiter().GetResult() + + Send-WebSocketJson $socket @{ + jsonrpc = "2.0" + id = 1 + method = "initialize" + params = @{ protocolVersion = 1; clientCapabilities = @{}; clientInfo = @{ name = "windows-canary"; version = "1" } } + } + $rootInitialize = Receive-WebSocketResponse $socket 1 + $rootInitializeResult = Assert-RpcSuccess $rootInitialize 1 "root /acp initialize" + Assert-Canary -Condition ($rootInitializeResult.protocolVersion -eq 1) -Message "root /acp initialize protocolVersion" + + Send-WebSocketJson $socket @{ + jsonrpc = "2.0" + id = 2 + method = "session/new" + params = @{ cwd = $WorkDir; mcpServers = @() } + } + $rootNew = Receive-WebSocketResponse $socket 2 + $rootNewResult = Assert-RpcSuccess $rootNew 2 "root /acp session/new" + $sessionId = [string]$rootNewResult.sessionId + Assert-Canary -Condition ($sessionId.StartsWith("sess_")) -Message "root /acp session/new" + + Send-WebSocketJson $socket @{ + jsonrpc = "2.0" + id = 3 + method = "session/prompt" + params = @{ sessionId = $sessionId; prompt = @(@{ type = "text"; text = "NO_NETWORK_CANARY" }) } + } + $promptSettlement = Receive-WebSocketPromptSettlement $socket 3 45 + $promptResult = $promptSettlement.Frame + $promptUpdateText = [string]$promptSettlement.UpdateText + Write-Host "[canary] prompt frame=$(Format-CanaryFrame $promptResult)" + Write-Host "[canary] prompt update=$(Redact-CanaryText $promptUpdateText)" + # NO_NETWORK_CANARY runs with a non-secret key and a proxy pointed at a + # dead loopback port (127.0.0.1:9). /acp still settles session/prompt as + # result {stopReason} when the backend fails; the failure shape lives in + # JSON-RPC error and/or streamed session/update text. A completed model + # turn without those markers is a canary failure. + $promptSettled = (@($promptResult.PSObject.Properties.Name) -contains "result") -or + (@($promptResult.PSObject.Properties.Name) -contains "error") + Assert-Canary -Condition $promptSettled -Message "root prompt did not settle; frame=$(Format-CanaryFrame $promptResult)" + Assert-Canary -Condition (Test-NoProviderFailureShape $promptResult $promptUpdateText) -Message "root no-network prompt did not show a no-provider/no-credentials failure shape; frame=$(Format-CanaryFrame $promptResult); update=$(Redact-CanaryText $promptUpdateText)" + + $children = @(Get-CimInstance Win32_Process -Filter "ParentProcessId = $($root.Id)" | + Where-Object { $_.Name -ieq "openab-agent.exe" }) + Assert-Canary -Condition ($children.Count -eq 1) -Message "root did not own exactly one openab-agent.exe" + $agentPid = [int]$children[0].ProcessId + + Send-WebSocketJson $socket @{ + jsonrpc = "2.0" + id = 4 + method = "session/prompt" + params = @{ sessionId = $sessionId; prompt = @(@{ type = "text"; text = "CANCEL_CANARY" }) } + } + Send-WebSocketJson $socket @{ + jsonrpc = "2.0" + method = "session/cancel" + params = @{ sessionId = $sessionId } + } + $cancelResult = Receive-WebSocketResponse $socket 4 15 + $cancelResultResult = Assert-RpcSuccess $cancelResult 4 "gateway cancellation" + Assert-Canary -Condition ($cancelResultResult.stopReason -eq "cancelled") -Message "gateway cancellation did not settle as cancelled" + Write-Host "[PASS] openab.exe /acp -> openab-agent.exe full-chain spawn and gateway cancel" + + $socket.Dispose() + $socket = $null + Stop-Process -Id $root.Id -Force + Assert-Canary -Condition ($root.WaitForExit(10000)) -Message "root did not terminate" + + $watch = [Diagnostics.Stopwatch]::StartNew() + while ($watch.Elapsed.TotalSeconds -lt 10 -and -not (Test-ProcessGone $agentPid)) { + Start-Sleep -Milliseconds 100 + } + Assert-Canary -Condition (Test-ProcessGone $agentPid) -Message "root termination left openab-agent.exe behind" + Write-Host "[PASS] Windows Job Object removed the downstream agent on root termination" + Write-Host "WINDOWS_DUAL_LAYER_CANARY_PASS" +} +finally { + if ($null -ne $socket) { + $socket.Dispose() + } + foreach ($process in @($agent, $root)) { + if ($null -ne $process) { + try { + if (-not $process.HasExited) { + Stop-Process -Id $process.Id -Force -ErrorAction SilentlyContinue + [void]$process.WaitForExit(5000) + } + } + catch { + } + $process.Dispose() + } + } + if ($null -ne $agentPid -and -not (Test-ProcessGone $agentPid)) { + Stop-Process -Id $agentPid -Force -ErrorAction SilentlyContinue + } + if (Test-Path -LiteralPath $tempRoot) { + Remove-Item -LiteralPath $tempRoot -Recurse -Force + } +}