From 801242ef93b736ec2ee8bdad662e1648e607bf0b Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Thu, 20 Aug 2026 12:55:49 -0300 Subject: [PATCH 1/6] feat(kw): add KwActor skeleton and message protocol This commit introduces KwActor with the full message surface and the immediate-reply contract for Start requests. The skeleton refuses unimplemented starts instantly, exposes idle status over a watch channel, and already routes RecordApply and GetReadiness through the shared history store and probes. This commit is part of the kw integration's step 4. Signed-off-by: lorenzoberts --- src/kw/actor.rs | 506 +++++++++++++++++++++++++++++++++++++++++++++ src/kw/errors.rs | 38 ++++ src/kw/handle.rs | 124 +++++++++++ src/kw/messages.rs | 88 ++++++++ src/kw/mod.rs | 10 +- src/kw/status.rs | 63 ++++++ 6 files changed, 827 insertions(+), 2 deletions(-) create mode 100644 src/kw/actor.rs create mode 100644 src/kw/errors.rs create mode 100644 src/kw/handle.rs create mode 100644 src/kw/messages.rs create mode 100644 src/kw/status.rs diff --git a/src/kw/actor.rs b/src/kw/actor.rs new file mode 100644 index 0000000..071fc80 --- /dev/null +++ b/src/kw/actor.rs @@ -0,0 +1,506 @@ +//! kw actor: owns kw build/deploy job state and the kw history store. +//! +//! All kw operations go through [`KwHandle`](crate::kw::handle::KwHandle) as +//! typed request/reply messages. `Start*` messages reply immediately with an +//! accept/refuse verdict — a job accepted by the actor keeps running after +//! the caller has been answered, so the AppActor loop never blocks on a +//! kernel build. The actor composes the readiness probes from +//! [`crate::kw::readiness`] and records applies through the shared +//! [`KwHistoryStore`](crate::kw::history::KwHistoryStore). + +// No production caller until the actor is wired into the app; the allow +// marks the module as a live root so the message/status/handle types it +// references stay live too. Removed once main.rs spawns the actor. Kept +// per the CachePolicy precedent (src/lore/application/cache.rs). +#![allow(dead_code)] + +use std::{ops::ControlFlow, path::PathBuf, sync::Arc}; + +use tokio::{ + spawn, + sync::{mpsc, oneshot, watch}, +}; + +use crate::{ + config::KernelTree, + infrastructure::{ + env::EnvTrait, + file_system::FileSystemTrait, + process::ProcessTrait, + shell::{ShellCommand, ShellTrait}, + }, + kw::{ + errors::{KwError, KwStartError}, + handle::KwHandle, + history::KwHistoryStore, + messages::KwMessage, + readiness::{self, KwReadiness}, + status::KwStatusSnapshot, + }, +}; + +pub const DEFAULT_KW_CHANNEL_SIZE: usize = 16; + +pub struct KwActor { + rx: mpsc::Receiver, + status_tx: watch::Sender, + history: Arc, + shell: Arc, + fs: Arc, + env: Arc, + // Read once job execution lands; kept per the CachePolicy precedent + // (src/lore/application/cache.rs). + #[allow(dead_code)] + process: Arc, + #[allow(dead_code)] + kw_log_dir: PathBuf, +} + +impl KwActor { + pub fn new( + rx: mpsc::Receiver, + history: Arc, + process: Arc, + shell: Arc, + fs: Arc, + env: Arc, + kw_log_dir: PathBuf, + ) -> Self { + let (status_tx, _) = watch::channel(KwStatusSnapshot::idle()); + Self { + rx, + status_tx, + history, + shell, + fs, + env, + process, + kw_log_dir, + } + } + + #[allow(clippy::too_many_arguments)] + pub fn spawn( + history: Arc, + process: Arc, + shell: Arc, + fs: Arc, + env: Arc, + kw_log_dir: PathBuf, + ) -> KwHandle { + let (tx, rx) = mpsc::channel(DEFAULT_KW_CHANNEL_SIZE); + tracing::debug!(channel_size = DEFAULT_KW_CHANNEL_SIZE, "spawning kw actor"); + spawn(Self::new(rx, history, process, shell, fs, env, kw_log_dir).run()); + KwHandle::new(tx) + } + + pub async fn run(mut self) { + tracing::info!("kw actor started"); + while let Some(message) = self.rx.recv().await { + if let ControlFlow::Break(()) = self.handle_message(message) { + break; + } + } + tracing::info!("kw actor stopped"); + } + + fn handle_message(&mut self, message: KwMessage) -> ControlFlow<()> { + let message_name = message.name(); + tracing::debug!(message = message_name, "kw request received"); + + match message { + KwMessage::RecordApply { record, reply } => { + send_kw_reply( + message_name, + reply, + self.history.record_apply(record).map_err(KwError::from), + ); + ControlFlow::Continue(()) + } + // Job execution lands with the build step; the immediate-reply + // contract holds from the skeleton onward, so callers never + // learn to depend on a blocking reply. + KwMessage::StartBuild { reply, .. } + | KwMessage::StartDeploy { reply, .. } + | KwMessage::StartBuildThenDeploy { reply, .. } => { + send_start_reply(message_name, reply, Err(KwStartError::NotImplemented)); + ControlFlow::Continue(()) + } + KwMessage::Cancel { reply } => { + send_kw_reply(message_name, reply, Err(KwError::NoJobRunning)); + ControlFlow::Continue(()) + } + KwMessage::GetStatus { reply } => { + send_value_reply(message_name, reply, self.status_tx.borrow().clone()); + ControlFlow::Continue(()) + } + KwMessage::WatchStatus { reply } => { + send_value_reply(message_name, reply, self.status_tx.subscribe()); + ControlFlow::Continue(()) + } + KwMessage::GetReadiness { + kernel_tree_id, + tree, + reply, + } => { + send_kw_reply( + message_name, + reply, + self.evaluate_readiness(&kernel_tree_id, &tree), + ); + ControlFlow::Continue(()) + } + // Restoring needs the dirty-worktree check and `git switch`, + // which arrive with the checkout policy in the build step. + KwMessage::RestorePreviousBranch { reply } => { + send_kw_reply(message_name, reply, Err(KwError::NoRecordedBranch)); + ControlFlow::Continue(()) + } + KwMessage::Shutdown => { + tracing::debug!("kw actor shutting down"); + ControlFlow::Break(()) + } + } + } + + fn evaluate_readiness( + &self, + kernel_tree_id: &str, + tree: &KernelTree, + ) -> Result { + let head = self.head_branch(tree); + Ok(readiness::evaluate_readiness( + &*self.fs, + &*self.env, + &*self.shell, + &*self.history, + kernel_tree_id, + tree, + &head, + )?) + } + + /// The tree's current branch, probed via git. An unresolvable HEAD + /// (not a git repo, or a detached HEAD, for which `branch + /// --show-current` prints nothing) yields an empty string: no build + /// record can match it, so deploy-alone readiness refuses — the safe + /// direction for an unknown HEAD. + fn head_branch(&self, tree: &KernelTree) -> String { + let cmd = ShellCommand::new("git").args(["-C", tree.path(), "branch", "--show-current"]); + match self.shell.execute(&cmd) { + Ok(output) if output.success => { + String::from_utf8_lossy(&output.stdout).trim().to_string() + } + Ok(output) => { + tracing::warn!( + tree = tree.path(), + stderr = %String::from_utf8_lossy(&output.stderr), + "failed to probe the kernel tree's HEAD branch" + ); + String::new() + } + Err(error) => { + tracing::warn!(tree = tree.path(), %error, "failed to probe the kernel tree's HEAD branch"); + String::new() + } + } + } +} + +fn send_kw_reply( + message_name: &'static str, + reply: oneshot::Sender>, + result: Result, +) { + if let Err(error) = &result { + tracing::warn!( + message = message_name, + error = %error, + "kw request failed" + ); + } + + if reply.send(result).is_err() { + tracing::warn!( + message = message_name, + "kw reply receiver dropped before response" + ); + } +} + +fn send_start_reply( + message_name: &'static str, + reply: oneshot::Sender>, + result: Result<(), KwStartError>, +) { + if let Err(error) = &result { + tracing::warn!( + message = message_name, + error = %error, + "kw start request refused" + ); + } + + if reply.send(result).is_err() { + tracing::warn!( + message = message_name, + "kw reply receiver dropped before response" + ); + } +} + +fn send_value_reply(message_name: &'static str, reply: oneshot::Sender, value: T) { + if reply.send(value).is_err() { + tracing::warn!( + message = message_name, + "kw reply receiver dropped before response" + ); + } +} + +#[cfg(test)] +mod tests { + use std::{io, path::Path, time::Duration}; + + use crate::{ + infrastructure::{ + env::MockEnvTrait, + file_system::{FileSystemError, MockFileSystemTrait}, + process::FakeProcess, + shell::{MockShellTrait, ShellOutput}, + }, + kw::{ + errors::KwStartError, + history::{KwApplyRecord, MockKwHistoryStore}, + messages::StartRequest, + readiness::{DeployAloneRefusal, TreeReadiness}, + status::KwJobStatus, + }, + }; + + use super::*; + + fn kernel_tree(path: &Path) -> KernelTree { + serde_json::from_value(serde_json::json!({ + "path": path.to_str().unwrap(), + "branch": "master" + })) + .unwrap() + } + + fn start_request() -> StartRequest { + StartRequest { + kernel_tree_id: "mainline".to_string(), + tree: kernel_tree(Path::new("/home/user/linux")), + branch: "patchset-2026-08-01-17-30-00".to_string(), + } + } + + /// Spawns the actor with the already-configured mocks (mockall + /// expectations need `&mut`, so they are set before the mocks move + /// behind `Arc`s). + fn spawn_test_actor( + history: MockKwHistoryStore, + shell: MockShellTrait, + fs: MockFileSystemTrait, + env: MockEnvTrait, + ) -> KwHandle { + KwActor::spawn( + Arc::new(history), + Arc::new(FakeProcess::new()), + Arc::new(shell), + Arc::new(fs), + Arc::new(env), + PathBuf::from("/tmp/patch-hub-test-kw-logs"), + ) + } + + fn apply_record() -> KwApplyRecord { + KwApplyRecord { + message_id: "msg-1".to_string(), + kernel_tree_id: "mainline".to_string(), + tree_path: "/home/user/linux".to_string(), + applied_branch: "patchset-2026-08-01-17-30-00".to_string(), + base_branch: "master".to_string(), + applied_at: "2026-08-01T17:30:00Z".to_string(), + } + } + + #[tokio::test] + async fn record_apply_writes_through_history_store() { + let expected = apply_record(); + let mut history = MockKwHistoryStore::new(); + history + .expect_record_apply() + .withf(move |record| *record == expected) + .times(1) + .returning(|_| Ok(())); + let handle = spawn_test_actor( + history, + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + handle.record_apply(apply_record()).await.unwrap(); + handle.shutdown().await; + } + + #[tokio::test] + async fn record_apply_surfaces_store_errors() { + let mut history = MockKwHistoryStore::new(); + history + .expect_record_apply() + .times(1) + .returning(|_| Err(FileSystemError::IoError(io::Error::other("disk full")))); + let handle = spawn_test_actor( + history, + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let err = handle.record_apply(apply_record()).await.unwrap_err(); + + assert!(matches!(err, KwError::History(_))); + handle.shutdown().await; + } + + #[tokio::test] + async fn get_status_reports_idle_before_any_job() { + let handle = spawn_test_actor( + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let snapshot = handle.get_status().await.unwrap(); + + assert_eq!(KwJobStatus::Idle, snapshot.job); + handle.shutdown().await; + } + + #[tokio::test] + async fn watch_status_receiver_sees_current_snapshot() { + let handle = spawn_test_actor( + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let receiver = handle.watch_status().await.unwrap(); + + assert_eq!(KwJobStatus::Idle, receiver.borrow().job); + handle.shutdown().await; + } + + #[tokio::test] + async fn get_readiness_composes_probes_and_head_branch() { + let tree = kernel_tree(Path::new("/home/user/linux")); + + let mut env = MockEnvTrait::new(); + env.expect_which() + .withf(|name| name == "kw") + .returning(|_| false); + let mut fs = MockFileSystemTrait::new(); + fs.expect_is_file().returning(|_| false); + fs.expect_is_dir().returning(|_| false); + fs.expect_read_dir().returning(|_| { + Err(FileSystemError::IoError(io::Error::new( + io::ErrorKind::NotFound, + "missing", + ))) + }); + let mut history = MockKwHistoryStore::new(); + history + .expect_build_records() + .withf(|kernel_tree_id, branch| kernel_tree_id == "mainline" && branch == "for-next") + .times(1) + .returning(|_, _| Ok((None, None))); + let mut shell = MockShellTrait::new(); + shell + .expect_execute() + .withf(|cmd| { + cmd.program == "git" + && cmd.args == ["-C", "/home/user/linux", "branch", "--show-current"] + }) + .times(1) + .returning(|_| { + Ok(ShellOutput { + stdout: b"for-next\n".to_vec(), + stderr: Vec::new(), + success: true, + }) + }); + let handle = spawn_test_actor(history, shell, fs, env); + + let readiness = handle.get_readiness("mainline", &tree).await.unwrap(); + + assert!(!readiness.kw_binary.available); + assert_eq!(TreeReadiness::Missing, readiness.tree); + assert_eq!( + Err(DeployAloneRefusal::TreeNotReady(TreeReadiness::Missing)), + readiness.deploy_alone + ); + handle.shutdown().await; + } + + #[tokio::test] + async fn start_build_replies_immediately_with_refusal() { + let handle = spawn_test_actor( + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + // The immediate-reply contract: the answer must arrive without any + // job completing — there is not even a job yet. + let result = + tokio::time::timeout(Duration::from_secs(1), handle.start_build(start_request())) + .await + .expect("start_build must reply immediately"); + + assert!(matches!(result, Err(KwStartError::NotImplemented))); + handle.shutdown().await; + } + + #[tokio::test] + async fn cancel_and_restore_without_job_are_immediate_errors() { + let handle = spawn_test_actor( + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let cancel = tokio::time::timeout(Duration::from_secs(1), handle.cancel()) + .await + .expect("cancel must reply immediately"); + let restore = + tokio::time::timeout(Duration::from_secs(1), handle.restore_previous_branch()) + .await + .expect("restore must reply immediately"); + + assert!(matches!(cancel, Err(KwError::NoJobRunning))); + assert!(matches!(restore, Err(KwError::NoRecordedBranch))); + handle.shutdown().await; + } + + #[tokio::test] + async fn shutdown_stops_actor() { + let handle = spawn_test_actor( + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + handle.shutdown().await; + let err = handle.get_status().await.unwrap_err(); + + assert!(matches!(err, KwError::ActorUnavailable(_))); + } +} diff --git a/src/kw/errors.rs b/src/kw/errors.rs new file mode 100644 index 0000000..77aedf4 --- /dev/null +++ b/src/kw/errors.rs @@ -0,0 +1,38 @@ +use thiserror::Error; + +use crate::{ + infrastructure::{file_system::FileSystemError, shell::ShellError}, + kw::readiness::KwReadinessError, +}; + +#[derive(Debug, Error)] +pub enum KwError { + #[error("kw actor unavailable: {0}")] + ActorUnavailable(String), + #[error("no kw job is running")] + NoJobRunning, + #[error("no pre-job branch was recorded")] + NoRecordedBranch, + #[error("history error: {0}")] + History(#[from] FileSystemError), + #[error("readiness error: {0}")] + Readiness(#[from] KwReadinessError), + #[error("shell error: {0}")] + Shell(#[from] ShellError), +} + +/// Accept/refuse verdict for `Start*` messages. The reply is always +/// immediate: an accepted job keeps running inside the actor after the +/// caller has been answered. +#[derive(Debug, Error)] +pub enum KwStartError { + #[error("kw actor unavailable: {0}")] + ActorUnavailable(String), + // Constructed once job execution lands; kept per the CachePolicy + // precedent (src/lore/application/cache.rs). + #[allow(dead_code)] + #[error("a kw job is already running")] + JobAlreadyRunning, + #[error("kw jobs are not supported yet")] + NotImplemented, +} diff --git a/src/kw/handle.rs b/src/kw/handle.rs new file mode 100644 index 0000000..2ef4b2d --- /dev/null +++ b/src/kw/handle.rs @@ -0,0 +1,124 @@ +use tokio::sync::{mpsc, oneshot, watch}; + +use crate::{ + config::KernelTree, + kw::{ + errors::{KwError, KwStartError}, + history::KwApplyRecord, + messages::{KwMessage, StartRequest}, + readiness::KwReadiness, + status::KwStatusSnapshot, + }, +}; + +#[derive(Clone)] +pub struct KwHandle { + tx: mpsc::Sender, +} + +// No production caller until the actor is wired into the app; kept per the +// CachePolicy precedent (src/lore/application/cache.rs). +#[allow(dead_code)] +impl KwHandle { + pub fn new(tx: mpsc::Sender) -> Self { + Self { tx } + } + + pub async fn record_apply(&self, record: KwApplyRecord) -> Result<(), KwError> { + self.request_result(|reply| KwMessage::RecordApply { record, reply }) + .await + } + + /// Resolves as soon as the actor accepts or refuses the job — never + /// when the job finishes. + pub async fn start_build(&self, request: StartRequest) -> Result<(), KwStartError> { + self.start(|reply| KwMessage::StartBuild { request, reply }) + .await + } + + pub async fn start_deploy(&self, request: StartRequest) -> Result<(), KwStartError> { + self.start(|reply| KwMessage::StartDeploy { request, reply }) + .await + } + + pub async fn start_build_then_deploy(&self, request: StartRequest) -> Result<(), KwStartError> { + self.start(|reply| KwMessage::StartBuildThenDeploy { request, reply }) + .await + } + + pub async fn cancel(&self) -> Result<(), KwError> { + self.request_result(|reply| KwMessage::Cancel { reply }) + .await + } + + pub async fn get_status(&self) -> Result { + self.request(|reply| KwMessage::GetStatus { reply }).await + } + + pub async fn watch_status(&self) -> Result, KwError> { + self.request(|reply| KwMessage::WatchStatus { reply }).await + } + + pub async fn get_readiness( + &self, + kernel_tree_id: &str, + tree: &KernelTree, + ) -> Result { + self.request_result(|reply| KwMessage::GetReadiness { + kernel_tree_id: kernel_tree_id.to_string(), + tree: tree.clone(), + reply, + }) + .await + } + + pub async fn restore_previous_branch(&self) -> Result<(), KwError> { + self.request_result(|reply| KwMessage::RestorePreviousBranch { reply }) + .await + } + + /// Signals the actor to stop processing messages and exit its run loop. + /// A running job is killed before the actor stops. + pub async fn shutdown(&self) { + self.tx.send(KwMessage::Shutdown).await.ok(); + } + + async fn request_result( + &self, + build_message: impl FnOnce(oneshot::Sender>) -> KwMessage, + ) -> Result + where + T: Send + 'static, + { + self.request(build_message).await? + } + + async fn request( + &self, + build_message: impl FnOnce(oneshot::Sender) -> KwMessage, + ) -> Result + where + T: Send + 'static, + { + let (reply, rx) = oneshot::channel(); + self.tx + .send(build_message(reply)) + .await + .map_err(|_| KwError::ActorUnavailable("request channel closed".to_string()))?; + rx.await + .map_err(|_| KwError::ActorUnavailable("reply channel closed".to_string())) + } + + async fn start( + &self, + build_message: impl FnOnce(oneshot::Sender>) -> KwMessage, + ) -> Result<(), KwStartError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(build_message(reply)) + .await + .map_err(|_| KwStartError::ActorUnavailable("request channel closed".to_string()))?; + rx.await + .map_err(|_| KwStartError::ActorUnavailable("reply channel closed".to_string()))? + } +} diff --git a/src/kw/messages.rs b/src/kw/messages.rs new file mode 100644 index 0000000..51617e8 --- /dev/null +++ b/src/kw/messages.rs @@ -0,0 +1,88 @@ +use tokio::sync::{oneshot, watch}; + +use crate::{ + config::KernelTree, + kw::{ + errors::{KwError, KwStartError}, + history::KwApplyRecord, + readiness::KwReadiness, + status::KwStatusSnapshot, + }, +}; + +/// Everything the actor needs to start a job. The tree context is resolved +/// by the caller from its config snapshot, keeping KwActor decoupled from +/// ConfigActor. +// Fields are read once job execution lands; kept per the CachePolicy +// precedent (src/lore/application/cache.rs). +#[allow(dead_code)] +#[derive(Debug, Clone)] +pub struct StartRequest { + pub kernel_tree_id: String, + pub tree: KernelTree, + /// Branch the job must run on; the checkout policy that gets the tree + /// onto it lands with the build step. + pub branch: String, +} + +pub enum KwMessage { + RecordApply { + record: KwApplyRecord, + reply: oneshot::Sender>, + }, + StartBuild { + // Read once job execution lands (CachePolicy precedent). + #[allow(dead_code)] + request: StartRequest, + reply: oneshot::Sender>, + }, + StartDeploy { + #[allow(dead_code)] + request: StartRequest, + reply: oneshot::Sender>, + }, + StartBuildThenDeploy { + #[allow(dead_code)] + request: StartRequest, + reply: oneshot::Sender>, + }, + /// Acknowledges that kill was requested; the actual process death is + /// observed via the status snapshot, not this reply. + Cancel { + reply: oneshot::Sender>, + }, + GetStatus { + reply: oneshot::Sender, + }, + /// Called once by the AppActor when it attaches, not per frame: the + /// returned receiver is its wake source for status changes. + WatchStatus { + reply: oneshot::Sender>, + }, + GetReadiness { + kernel_tree_id: String, + tree: KernelTree, + reply: oneshot::Sender>, + }, + RestorePreviousBranch { + reply: oneshot::Sender>, + }, + Shutdown, +} + +impl KwMessage { + pub fn name(&self) -> &'static str { + match self { + KwMessage::RecordApply { .. } => "RecordApply", + KwMessage::StartBuild { .. } => "StartBuild", + KwMessage::StartDeploy { .. } => "StartDeploy", + KwMessage::StartBuildThenDeploy { .. } => "StartBuildThenDeploy", + KwMessage::Cancel { .. } => "Cancel", + KwMessage::GetStatus { .. } => "GetStatus", + KwMessage::WatchStatus { .. } => "WatchStatus", + KwMessage::GetReadiness { .. } => "GetReadiness", + KwMessage::RestorePreviousBranch { .. } => "RestorePreviousBranch", + KwMessage::Shutdown => "Shutdown", + } + } +} diff --git a/src/kw/mod.rs b/src/kw/mod.rs index 2e5b1b9..24f19ea 100644 --- a/src/kw/mod.rs +++ b/src/kw/mod.rs @@ -1,5 +1,11 @@ -//! kw integration: persistence and readiness probes for `kw build` / -//! `kw deploy` jobs. +//! kw integration: persistence, readiness probes, and the actor +//! orchestrating `kw build` / `kw deploy` jobs. +#[cfg(unix)] +pub mod actor; +pub mod errors; +pub mod handle; pub mod history; +pub mod messages; pub mod readiness; +pub mod status; diff --git a/src/kw/status.rs b/src/kw/status.rs new file mode 100644 index 0000000..da012bc --- /dev/null +++ b/src/kw/status.rs @@ -0,0 +1,63 @@ +//! Status snapshot projected out of the kw actor. +//! +//! `AppState` never owns job state; it polls (`GetStatus`) or watches +//! (`WatchStatus`) these snapshots and projects them into the view model. + +use std::path::PathBuf; + +// Most variants are constructed once job execution lands; kept per the +// CachePolicy precedent (src/lore/application/cache.rs). +#[allow(dead_code)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum KwJobKind { + Build, + Deploy, + BuildThenDeploy, +} + +/// Running phase of a job. `BuildThenDeploy` jobs pass through both. +#[allow(dead_code)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum KwPhase { + Building, + Deploying, +} + +#[allow(dead_code)] +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum KwJobStatus { + Idle, + Running { + kind: KwJobKind, + phase: KwPhase, + kernel_tree_id: String, + branch: String, + log_path: PathBuf, + }, + Succeeded { + kind: KwJobKind, + }, + Failed { + kind: KwJobKind, + phase: KwPhase, + exit_code: Option, + log_path: PathBuf, + }, + Cancelled { + kind: KwJobKind, + phase: KwPhase, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct KwStatusSnapshot { + pub job: KwJobStatus, +} + +impl KwStatusSnapshot { + pub fn idle() -> Self { + Self { + job: KwJobStatus::Idle, + } + } +} From 596f12d68756cbf9cc427f1e3281552529f2145e Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Thu, 20 Aug 2026 13:05:34 -0300 Subject: [PATCH 2/6] feat(kw): run background build jobs with cancel and status transitions This commit makes StartBuild accept a job without waiting for it to finish. A detached task owns the process, broadcasts Idle to a terminal status, and cancel or shutdown kill the process group so the actor never wedges on a live job. This commit is part of the kw integration's step 4. Signed-off-by: lorenzoberts --- src/kw/actor.rs | 447 ++++++++++++++++++++++++++++++++++++++++++--- src/kw/errors.rs | 9 +- src/kw/messages.rs | 6 +- src/kw/status.rs | 11 +- 4 files changed, 436 insertions(+), 37 deletions(-) diff --git a/src/kw/actor.rs b/src/kw/actor.rs index 071fc80..3cddbc8 100644 --- a/src/kw/actor.rs +++ b/src/kw/actor.rs @@ -14,7 +14,7 @@ // per the CachePolicy precedent (src/lore/application/cache.rs). #![allow(dead_code)] -use std::{ops::ControlFlow, path::PathBuf, sync::Arc}; +use std::{ops::ControlFlow, path::PathBuf, process::ExitStatus, sync::Arc}; use tokio::{ spawn, @@ -26,34 +26,60 @@ use crate::{ infrastructure::{ env::EnvTrait, file_system::FileSystemTrait, - process::ProcessTrait, + process::{ProcessError, ProcessTrait, RunningProcess}, shell::{ShellCommand, ShellTrait}, }, kw::{ errors::{KwError, KwStartError}, handle::KwHandle, history::KwHistoryStore, - messages::KwMessage, + messages::{KwMessage, StartRequest}, readiness::{self, KwReadiness}, - status::KwStatusSnapshot, + status::{KwJobKind, KwJobStatus, KwPhase, KwStatusSnapshot}, }, }; pub const DEFAULT_KW_CHANNEL_SIZE: usize = 16; +/// How a job's process ended, as observed by the detached task that owns +/// the process handle. +enum JobOutcome { + Exited(ExitStatus), + WaitFailed(ProcessError), + Cancelled, +} + +/// Internal report from a job task back to the actor loop. Kept off the +/// public [`KwMessage`] protocol: no caller can fake a completion. +enum JobEvent { + Finished(JobOutcome), +} + +/// What the actor remembers about the running job while the detached task +/// owns the process itself (see [`run_job`]). +struct JobState { + kind: KwJobKind, + phase: KwPhase, + kernel_tree_id: String, + branch: String, + log_path: PathBuf, + /// `None` once a cancel has been requested; a second `Cancel` is an + /// idempotent ack. + cancel_tx: Option>, +} + pub struct KwActor { rx: mpsc::Receiver, + job_event_rx: mpsc::Receiver, + job_event_tx: mpsc::Sender, status_tx: watch::Sender, history: Arc, shell: Arc, fs: Arc, env: Arc, - // Read once job execution lands; kept per the CachePolicy precedent - // (src/lore/application/cache.rs). - #[allow(dead_code)] process: Arc, - #[allow(dead_code)] kw_log_dir: PathBuf, + job: Option, } impl KwActor { @@ -67,8 +93,11 @@ impl KwActor { kw_log_dir: PathBuf, ) -> Self { let (status_tx, _) = watch::channel(KwStatusSnapshot::idle()); + let (job_event_tx, job_event_rx) = mpsc::channel(DEFAULT_KW_CHANNEL_SIZE); Self { rx, + job_event_rx, + job_event_tx, status_tx, history, shell, @@ -76,6 +105,7 @@ impl KwActor { env, process, kw_log_dir, + job: None, } } @@ -96,9 +126,19 @@ impl KwActor { pub async fn run(mut self) { tracing::info!("kw actor started"); - while let Some(message) = self.rx.recv().await { - if let ControlFlow::Break(()) = self.handle_message(message) { - break; + // The job-event sender is held by the actor itself, so that arm of + // the select never closes while the actor is alive. + loop { + tokio::select! { + message = self.rx.recv() => { + let Some(message) = message else { break }; + if let ControlFlow::Break(()) = self.handle_message(message) { + break; + } + } + Some(event) = self.job_event_rx.recv() => { + self.handle_job_event(event); + } } } tracing::info!("kw actor stopped"); @@ -117,17 +157,24 @@ impl KwActor { ); ControlFlow::Continue(()) } - // Job execution lands with the build step; the immediate-reply - // contract holds from the skeleton onward, so callers never + KwMessage::StartBuild { request, reply } => { + send_start_reply( + message_name, + reply, + self.start_job(KwJobKind::Build, request), + ); + ControlFlow::Continue(()) + } + // Deploy acceptance lands with the deploy step; the + // immediate-reply contract already holds, so callers never // learn to depend on a blocking reply. - KwMessage::StartBuild { reply, .. } - | KwMessage::StartDeploy { reply, .. } + KwMessage::StartDeploy { reply, .. } | KwMessage::StartBuildThenDeploy { reply, .. } => { send_start_reply(message_name, reply, Err(KwStartError::NotImplemented)); ControlFlow::Continue(()) } KwMessage::Cancel { reply } => { - send_kw_reply(message_name, reply, Err(KwError::NoJobRunning)); + send_kw_reply(message_name, reply, self.request_cancel()); ControlFlow::Continue(()) } KwMessage::GetStatus { reply } => { @@ -157,12 +204,134 @@ impl KwActor { ControlFlow::Continue(()) } KwMessage::Shutdown => { + // Quit-prompt cooperation: the app asks whether a job is + // running before quitting; if it quits anyway, the job's + // process group is killed here. The log may capture partial + // output and the tree a partial build. + if let Some(cancel) = self.job.as_mut().and_then(|job| job.cancel_tx.take()) { + tracing::info!("killing kw job process group during shutdown"); + let _ = cancel.send(()); + } tracing::debug!("kw actor shutting down"); ControlFlow::Break(()) } } } + /// Accepts and starts a build job, or refuses. The reply is sent by the + /// caller right after this returns: the job itself keeps running in a + /// detached task and is observed via the status snapshot. + /// + /// The argv is the skeleton's minimal `kw build --alert=n`; the real + /// argv builder (reserved flags, extra-args merge) and the checkout + /// policy land with the build step. + fn start_job(&mut self, kind: KwJobKind, request: StartRequest) -> Result<(), KwStartError> { + if self.job.is_some() { + return Err(KwStartError::JobAlreadyRunning); + } + + self.fs.create_dir_all(&self.kw_log_dir)?; + let log_path = self.kw_log_dir.join(format!( + "build-{}.log", + chrono::Utc::now().format("%Y%m%d-%H%M%S") + )); + let cmd = ShellCommand::new("kw").args(["build", "--alert=n"]); + let cwd = PathBuf::from(request.tree.path()); + let process = self.process.spawn(&cmd, &cwd, &log_path)?; + + let (cancel_tx, cancel_rx) = oneshot::channel(); + spawn(run_job(process, cancel_rx, self.job_event_tx.clone())); + + let phase = KwPhase::Building; + tracing::info!( + kernel_tree_id = request.kernel_tree_id, + branch = request.branch, + log_path = %log_path.display(), + "kw build job started" + ); + self.job = Some(JobState { + kind, + phase, + kernel_tree_id: request.kernel_tree_id.clone(), + branch: request.branch.clone(), + log_path: log_path.clone(), + cancel_tx: Some(cancel_tx), + }); + self.set_status(KwJobStatus::Running { + kind, + phase, + kernel_tree_id: request.kernel_tree_id, + branch: request.branch, + log_path, + }); + Ok(()) + } + + fn request_cancel(&mut self) -> Result<(), KwError> { + match self.job.as_mut() { + Some(job) => { + if let Some(cancel) = job.cancel_tx.take() { + let _ = cancel.send(()); + } + Ok(()) + } + None => Err(KwError::NoJobRunning), + } + } + + fn handle_job_event(&mut self, event: JobEvent) { + match event { + JobEvent::Finished(outcome) => { + let Some(job) = self.job.take() else { + tracing::warn!("kw job finished with no job state recorded"); + return; + }; + let status = match outcome { + JobOutcome::Exited(exit) if exit.success() => { + tracing::info!(branch = job.branch, "kw job succeeded"); + KwJobStatus::Succeeded { kind: job.kind } + } + JobOutcome::Exited(exit) => { + tracing::warn!( + branch = job.branch, + exit_code = exit.code(), + "kw job failed" + ); + KwJobStatus::Failed { + kind: job.kind, + phase: job.phase, + exit_code: exit.code(), + log_path: job.log_path, + } + } + JobOutcome::WaitFailed(error) => { + tracing::warn!(branch = job.branch, %error, "failed to wait on kw job"); + KwJobStatus::Failed { + kind: job.kind, + phase: job.phase, + exit_code: None, + log_path: job.log_path, + } + } + JobOutcome::Cancelled => { + tracing::info!(branch = job.branch, "kw job cancelled"); + KwJobStatus::Cancelled { + kind: job.kind, + phase: job.phase, + } + } + }; + self.set_status(status); + } + } + } + + fn set_status(&mut self, job: KwJobStatus) { + // send_replace, not send: no receiver (nobody called WatchStatus + // yet) is a normal state, not an error. + self.status_tx.send_replace(KwStatusSnapshot { job }); + } + fn evaluate_readiness( &self, kernel_tree_id: &str, @@ -207,6 +376,33 @@ impl KwActor { } } +/// Owns the spawned process until it ends: waits on it, or — when the +/// cancel signal fires — kills the whole process group and reaps it. The +/// `wait()` future is dropped before the cancel arm's body runs, releasing +/// the mutable borrow so `kill()` can be called. Reports the outcome back +/// to the actor over the internal event channel. +async fn run_job( + mut process: Box, + mut cancel_rx: oneshot::Receiver<()>, + events: mpsc::Sender, +) { + let outcome = tokio::select! { + status = process.wait() => match status { + Ok(status) => JobOutcome::Exited(status), + Err(error) => JobOutcome::WaitFailed(error), + }, + // A dropped sender (actor shutting down) cancels the job too. + _ = &mut cancel_rx => { + if let Err(error) = process.kill() { + tracing::warn!(%error, "failed to kill kw job process group"); + } + let _ = process.wait().await; + JobOutcome::Cancelled + } + }; + events.send(JobEvent::Finished(outcome)).await.ok(); +} + fn send_kw_reply( message_name: &'static str, reply: oneshot::Sender>, @@ -260,7 +456,12 @@ fn send_value_reply(message_name: &'static str, reply: oneshot::Sender, va #[cfg(test)] mod tests { - use std::{io, path::Path, time::Duration}; + use std::{ + io, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, + }; use crate::{ infrastructure::{ @@ -274,12 +475,27 @@ mod tests { history::{KwApplyRecord, MockKwHistoryStore}, messages::StartRequest, readiness::{DeployAloneRefusal, TreeReadiness}, - status::KwJobStatus, + status::{KwJobKind, KwJobStatus, KwPhase}, }, }; use super::*; + static TEST_SEQ: AtomicU64 = AtomicU64::new(0); + + /// A real directory: FakeProcess creates the log file on spawn, so the + /// parent must exist even though the fs trait is mocked. + fn tmp_log_dir(test_name: &str) -> PathBuf { + let n = TEST_SEQ.fetch_add(1, Ordering::SeqCst); + let dir = std::env::temp_dir().join(format!( + "patch-hub-kw-actor-{}-{test_name}-{n}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + fn kernel_tree(path: &Path) -> KernelTree { serde_json::from_value(serde_json::json!({ "path": path.to_str().unwrap(), @@ -315,6 +531,24 @@ mod tests { ) } + /// Spawns the actor with a real temp log dir and exposes the + /// [`FakeProcess`] so tests drive the "running" process. + fn spawn_job_actor(test_name: &str) -> (KwHandle, Arc, PathBuf) { + let process = Arc::new(FakeProcess::new()); + let log_dir = tmp_log_dir(test_name); + let mut fs = MockFileSystemTrait::new(); + fs.expect_create_dir_all().returning(|_| Ok(())); + let handle = KwActor::spawn( + Arc::new(MockKwHistoryStore::new()), + process.clone(), + Arc::new(MockShellTrait::new()), + Arc::new(fs), + Arc::new(MockEnvTrait::new()), + log_dir.clone(), + ); + (handle, process, log_dir) + } + fn apply_record() -> KwApplyRecord { KwApplyRecord { message_id: "msg-1".to_string(), @@ -447,8 +681,177 @@ mod tests { handle.shutdown().await; } + /// Waits until the status leaves `Idle`/`Running` and returns the + /// terminal status. The receiver may have observed the `Running` + /// transition first, so a single `changed()` is not enough. + async fn wait_for_terminal_status( + watch: &mut watch::Receiver, + ) -> KwJobStatus { + tokio::time::timeout(Duration::from_secs(1), async { + loop { + let status = watch.borrow().job.clone(); + if !matches!(status, KwJobStatus::Idle | KwJobStatus::Running { .. }) { + return status; + } + watch.changed().await.unwrap(); + } + }) + .await + .expect("status must reach a terminal state") + } + #[tokio::test] - async fn start_build_replies_immediately_with_refusal() { + async fn start_build_replies_immediately_and_runs_in_background() { + let (handle, process, log_dir) = spawn_job_actor("start-immediate"); + + // The §3.1 reply contract: start_build resolves while the spawned + // process is still running (no finish() was ever signaled). + let result = + tokio::time::timeout(Duration::from_secs(1), handle.start_build(start_request())) + .await + .expect("start_build must reply immediately"); + result.unwrap(); + + let spawned = process.spawned(); + assert_eq!(1, spawned.len()); + assert_eq!("kw", spawned[0].program); + assert_eq!(["build", "--alert=n"], spawned[0].args.as_slice()); + assert_eq!(Path::new("/home/user/linux"), spawned[0].cwd); + assert!(spawned[0].log_path.starts_with(&log_dir)); + + let snapshot = handle.get_status().await.unwrap(); + assert!( + matches!( + snapshot.job, + KwJobStatus::Running { + kind: KwJobKind::Build, + phase: KwPhase::Building, + .. + } + ), + "unexpected status: {:?}", + snapshot.job + ); + + process.last_child().finish(0); + let mut watch = handle.watch_status().await.unwrap(); + let status = wait_for_terminal_status(&mut watch).await; + assert_eq!( + KwJobStatus::Succeeded { + kind: KwJobKind::Build + }, + status + ); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn second_start_while_running_is_refused() { + let (handle, process, log_dir) = spawn_job_actor("busy"); + + handle.start_build(start_request()).await.unwrap(); + let second = handle.start_build(start_request()).await; + + assert!(matches!(second, Err(KwStartError::JobAlreadyRunning))); + + process.last_child().finish(0); + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn failed_build_reports_exit_code_and_log_path() { + let (handle, process, log_dir) = spawn_job_actor("failed"); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + process.last_child().finish(2); + + let status = wait_for_terminal_status(&mut watch).await; + + match status { + KwJobStatus::Failed { + kind, + phase, + exit_code, + log_path, + } => { + assert_eq!(KwJobKind::Build, kind); + assert_eq!(KwPhase::Building, phase); + assert_eq!(Some(2), exit_code); + assert!(log_path.starts_with(&log_dir)); + } + other => panic!("expected Failed, got {other:?}"), + } + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn cancel_kills_process_group_and_reports_cancelled() { + let (handle, process, log_dir) = spawn_job_actor("cancel"); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + // The ack is immediate: process death is observed via the status, + // not the reply. + tokio::time::timeout(Duration::from_secs(1), handle.cancel()) + .await + .expect("cancel must ack immediately") + .unwrap(); + + assert!(process.last_child().was_killed()); + let status = wait_for_terminal_status(&mut watch).await; + assert_eq!( + KwJobStatus::Cancelled { + kind: KwJobKind::Build, + phase: KwPhase::Building, + }, + status + ); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn shutdown_kills_running_job() { + let (handle, process, log_dir) = spawn_job_actor("shutdown-kill"); + + handle.start_build(start_request()).await.unwrap(); + handle.shutdown().await; + + // shutdown() only enqueues the message; the actor being gone + // proves the Shutdown (and its kill) was processed. + let err = handle.get_status().await.unwrap_err(); + assert!(matches!(err, KwError::ActorUnavailable(_))); + assert!(process.last_child().was_killed()); + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn spawn_failure_refuses_start_and_stays_idle() { + let (handle, process, log_dir) = spawn_job_actor("spawn-fail"); + process.refuse_spawns(true); + + let err = handle.start_build(start_request()).await.unwrap_err(); + + assert!(matches!(err, KwStartError::Spawn(_))); + assert_eq!(KwJobStatus::Idle, handle.get_status().await.unwrap().job); + // A refused start must leave the actor able to accept a later one. + process.refuse_spawns(false); + handle.start_build(start_request()).await.unwrap(); + + process.last_child().finish(0); + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn start_deploy_still_refused_until_the_deploy_step() { let handle = spawn_test_actor( MockKwHistoryStore::new(), MockShellTrait::new(), @@ -456,12 +859,10 @@ mod tests { MockEnvTrait::new(), ); - // The immediate-reply contract: the answer must arrive without any - // job completing — there is not even a job yet. let result = - tokio::time::timeout(Duration::from_secs(1), handle.start_build(start_request())) + tokio::time::timeout(Duration::from_secs(1), handle.start_deploy(start_request())) .await - .expect("start_build must reply immediately"); + .expect("start_deploy must reply immediately"); assert!(matches!(result, Err(KwStartError::NotImplemented))); handle.shutdown().await; diff --git a/src/kw/errors.rs b/src/kw/errors.rs index 77aedf4..286f0bb 100644 --- a/src/kw/errors.rs +++ b/src/kw/errors.rs @@ -1,7 +1,7 @@ use thiserror::Error; use crate::{ - infrastructure::{file_system::FileSystemError, shell::ShellError}, + infrastructure::{file_system::FileSystemError, process::ProcessError, shell::ShellError}, kw::readiness::KwReadinessError, }; @@ -28,11 +28,12 @@ pub enum KwError { pub enum KwStartError { #[error("kw actor unavailable: {0}")] ActorUnavailable(String), - // Constructed once job execution lands; kept per the CachePolicy - // precedent (src/lore/application/cache.rs). - #[allow(dead_code)] #[error("a kw job is already running")] JobAlreadyRunning, #[error("kw jobs are not supported yet")] NotImplemented, + #[error("failed to spawn the kw process: {0}")] + Spawn(#[from] ProcessError), + #[error("filesystem error: {0}")] + Fs(#[from] FileSystemError), } diff --git a/src/kw/messages.rs b/src/kw/messages.rs index 51617e8..89f9323 100644 --- a/src/kw/messages.rs +++ b/src/kw/messages.rs @@ -13,9 +13,6 @@ use crate::{ /// Everything the actor needs to start a job. The tree context is resolved /// by the caller from its config snapshot, keeping KwActor decoupled from /// ConfigActor. -// Fields are read once job execution lands; kept per the CachePolicy -// precedent (src/lore/application/cache.rs). -#[allow(dead_code)] #[derive(Debug, Clone)] pub struct StartRequest { pub kernel_tree_id: String, @@ -31,12 +28,11 @@ pub enum KwMessage { reply: oneshot::Sender>, }, StartBuild { - // Read once job execution lands (CachePolicy precedent). - #[allow(dead_code)] request: StartRequest, reply: oneshot::Sender>, }, StartDeploy { + // Read once deploy execution lands (CachePolicy precedent). #[allow(dead_code)] request: StartRequest, reply: oneshot::Sender>, diff --git a/src/kw/status.rs b/src/kw/status.rs index da012bc..d46785a 100644 --- a/src/kw/status.rs +++ b/src/kw/status.rs @@ -5,25 +5,26 @@ use std::path::PathBuf; -// Most variants are constructed once job execution lands; kept per the -// CachePolicy precedent (src/lore/application/cache.rs). -#[allow(dead_code)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum KwJobKind { Build, + // Constructed once deploy execution lands; kept per the CachePolicy + // precedent (src/lore/application/cache.rs). + #[allow(dead_code)] Deploy, + #[allow(dead_code)] BuildThenDeploy, } /// Running phase of a job. `BuildThenDeploy` jobs pass through both. -#[allow(dead_code)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum KwPhase { Building, + // Constructed once deploy execution lands (CachePolicy precedent). + #[allow(dead_code)] Deploying, } -#[allow(dead_code)] #[derive(Debug, Clone, PartialEq, Eq)] pub enum KwJobStatus { Idle, From 6e7a72c93d411bcf2dbaa5431b5273d7bb1f787c Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Thu, 20 Aug 2026 13:13:06 -0300 Subject: [PATCH 3/6] feat(app): wire KwActor into the app and route apply history through it This commit spawns KwActor at startup and moves apply-history writes onto KwHandle so recording serializes with job state inside the actor. The handle is optional because the process module is unix-only; non-unix builds keep writing the store directly. Shutdown follows InputActor so a running job's process group is killed before the runtime tears down. This commit is part of the kw integration's step 4. Signed-off-by: lorenzoberts --- src/app/actor.rs | 2 + .../integration_tests/helpers/app_harness.rs | 1 + src/app/integration_tests/patchset_actions.rs | 18 ++++++- src/app/mod.rs | 54 +++++++++++++------ src/kw/actor.rs | 13 +++-- src/kw/errors.rs | 3 ++ src/kw/handle.rs | 5 +- src/kw/messages.rs | 3 ++ src/kw/status.rs | 3 ++ src/main.rs | 35 ++++++++++-- 10 files changed, 106 insertions(+), 31 deletions(-) diff --git a/src/app/actor.rs b/src/app/actor.rs index 497666a..c0ee9e2 100644 --- a/src/app/actor.rs +++ b/src/app/actor.rs @@ -234,6 +234,7 @@ mod tests { fs: Box::new(MockFileSystemTrait::new()), config: dummy_config_handle(), kw_history: Arc::new(MockKwHistoryStore::new()), + kw: None, }, } } @@ -336,6 +337,7 @@ mod tests { lore_api.clone(), render.clone(), Arc::new(MockKwHistoryStore::new()), + None, ) .expect("App::new must succeed"); diff --git a/src/app/integration_tests/helpers/app_harness.rs b/src/app/integration_tests/helpers/app_harness.rs index ba7531f..6969eac 100644 --- a/src/app/integration_tests/helpers/app_harness.rs +++ b/src/app/integration_tests/helpers/app_harness.rs @@ -56,6 +56,7 @@ pub(crate) fn app_with_bootstrap_and_handles( lore_api, render, Arc::new(MockKwHistoryStore::new()), + None, ) .expect("minimal app should build") } diff --git a/src/app/integration_tests/patchset_actions.rs b/src/app/integration_tests/patchset_actions.rs index 1eadbc7..8070d1e 100644 --- a/src/app/integration_tests/patchset_actions.rs +++ b/src/app/integration_tests/patchset_actions.rs @@ -1,5 +1,6 @@ use std::{ collections::{HashMap, HashSet, VecDeque}, + path::PathBuf, sync::{Arc, Mutex}, }; @@ -16,10 +17,12 @@ use crate::{ }, config::{ConfigSnapshot, ConfigState}, infrastructure::{ + env::MockEnvTrait, file_system::{FileSystemError, MockFileSystemTrait}, + process::FakeProcess, shell::{MockShellTrait, ShellCommand, ShellOutput}, }, - kw::history::MockKwHistoryStore, + kw::{actor::KwActor, history::MockKwHistoryStore}, lore::application::{ cache::BootstrapLoreData, handle::LoreApiHandle, messages::LoreApiMessage, }, @@ -329,6 +332,16 @@ fn app_with_details( config: ConfigSnapshot, kw_history: MockKwHistoryStore, ) -> App { + // Apply history is recorded through the real actor wrapping the mock + // store, mirroring production wiring. + let kw = KwActor::spawn( + Arc::new(kw_history), + Arc::new(FakeProcess::new()), + Arc::new(MockShellTrait::new()), + Arc::new(MockFileSystemTrait::new()), + Arc::new(MockEnvTrait::new()), + PathBuf::from("/tmp/patch-hub-test-kw-logs"), + ); let mut app = App::new( config, dummy_config_handle(), @@ -341,7 +354,8 @@ fn app_with_details( Box::new(shell), lore_api, dummy_render_handle(), - Arc::new(kw_history), + Arc::new(MockKwHistoryStore::new()), + Some(kw), ) .expect("app should build"); diff --git a/src/app/mod.rs b/src/app/mod.rs index 58b418c..c10819c 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -47,7 +47,10 @@ use crate::{ file_system::FileSystemTrait, monitoring::logging::garbage_collector::collect_garbage, shell::ShellTrait, }, - kw::history::{KwApplyRecord, KwHistoryStore}, + kw::{ + handle::KwHandle, + history::{KwApplyRecord, KwHistoryStore}, + }, lore::{ application::{ cache::{BootstrapLoreData, CacheMode}, @@ -76,8 +79,12 @@ pub struct AppServices { pub shell: Box, pub fs: Box, pub config: ConfigHandle, - /// Shared with KwActor once it exists (the actor adopts the same store). + /// Direct access to the store, for the non-unix fallback below. pub kw_history: Arc, + /// `None` on non-unix builds, where ProcessTrait (and thus KwActor) + /// does not exist; apply-history writes then go to `kw_history` + /// directly, as they did before the actor landed. + pub kw: Option, } /// Result type signalling whether a patchset was successfully loaded. @@ -110,6 +117,7 @@ impl App { lore_api: LoreApiHandle, render: RenderHandle, kw_history: Arc, + kw: Option, ) -> Result { event!(Level::INFO, "patch-hub started"); collect_garbage(&config); @@ -147,6 +155,7 @@ impl App { fs, config: config_handle, kw_history, + kw, }, }) } @@ -310,7 +319,7 @@ impl App { debug!("consolidating patchset actions"); self.sync_patchset_bookmark().await?; self.execute_reviewed_reply().await?; - self.execute_apply_patchset(); + self.execute_apply_patchset().await; debug!("patchset actions consolidated"); Ok(()) } @@ -408,7 +417,7 @@ impl App { Ok(()) } - fn execute_apply_patchset(&mut self) { + async fn execute_apply_patchset(&mut self) { let details = self .state .lore @@ -437,18 +446,33 @@ impl App { applied.message ) } - // The git apply itself succeeded; a history-write - // failure must not turn it into a reported failure. - Some(record) => match self.services.kw_history.record_apply(record) { - Ok(()) => applied.message, - Err(e) => { - warn!(error = %e, "failed to record kw apply history"); - format!( - "{}\n\nWarning: the apply was not recorded in the kw history: {e}\nIf this warning keeps appearing, inspect or delete that file.", - applied.message - ) + Some(record) => { + // History writes go through KwActor so apply + // recording serializes with job state; without an + // actor (non-unix), write the store directly. + let recorded = match &self.services.kw { + Some(kw) => { + kw.record_apply(record).await.map_err(|e| e.to_string()) + } + None => self + .services + .kw_history + .record_apply(record) + .map_err(|e| e.to_string()), + }; + // The git apply itself succeeded; a history-write + // failure must not turn it into a reported failure. + match recorded { + Ok(()) => applied.message, + Err(e) => { + warn!(error = %e, "failed to record kw apply history"); + format!( + "{}\n\nWarning: the apply was not recorded in the kw history: {e}\nIf this warning keeps appearing, inspect or delete that file.", + applied.message + ) + } } - }, + } }; popup::AppPopup::info("Patchset Apply Success", popup_body) } diff --git a/src/kw/actor.rs b/src/kw/actor.rs index 3cddbc8..57cc040 100644 --- a/src/kw/actor.rs +++ b/src/kw/actor.rs @@ -8,12 +8,6 @@ //! [`crate::kw::readiness`] and records applies through the shared //! [`KwHistoryStore`](crate::kw::history::KwHistoryStore). -// No production caller until the actor is wired into the app; the allow -// marks the module as a live root so the message/status/handle types it -// references stay live too. Removed once main.rs spawns the actor. Kept -// per the CachePolicy precedent (src/lore/application/cache.rs). -#![allow(dead_code)] - use std::{ops::ControlFlow, path::PathBuf, process::ExitStatus, sync::Arc}; use tokio::{ @@ -288,11 +282,16 @@ impl KwActor { }; let status = match outcome { JobOutcome::Exited(exit) if exit.success() => { - tracing::info!(branch = job.branch, "kw job succeeded"); + tracing::info!( + kernel_tree_id = job.kernel_tree_id, + branch = job.branch, + "kw job succeeded" + ); KwJobStatus::Succeeded { kind: job.kind } } JobOutcome::Exited(exit) => { tracing::warn!( + kernel_tree_id = job.kernel_tree_id, branch = job.branch, exit_code = exit.code(), "kw job failed" diff --git a/src/kw/errors.rs b/src/kw/errors.rs index 286f0bb..aa8c16f 100644 --- a/src/kw/errors.rs +++ b/src/kw/errors.rs @@ -1,3 +1,6 @@ +// The actor that constructs/reads these is unix-only. +#![cfg_attr(not(unix), allow(dead_code))] + use thiserror::Error; use crate::{ diff --git a/src/kw/handle.rs b/src/kw/handle.rs index 2ef4b2d..8e20c77 100644 --- a/src/kw/handle.rs +++ b/src/kw/handle.rs @@ -16,8 +16,9 @@ pub struct KwHandle { tx: mpsc::Sender, } -// No production caller until the actor is wired into the app; kept per the -// CachePolicy precedent (src/lore/application/cache.rs). +// Only record_apply/shutdown have a production caller until the KwOps +// screen lands; kept per the CachePolicy precedent +// (src/lore/application/cache.rs). #[allow(dead_code)] impl KwHandle { pub fn new(tx: mpsc::Sender) -> Self { diff --git a/src/kw/messages.rs b/src/kw/messages.rs index 89f9323..ca99c91 100644 --- a/src/kw/messages.rs +++ b/src/kw/messages.rs @@ -1,3 +1,6 @@ +// The actor that reads these fields is unix-only. +#![cfg_attr(not(unix), allow(dead_code))] + use tokio::sync::{oneshot, watch}; use crate::{ diff --git a/src/kw/status.rs b/src/kw/status.rs index d46785a..f2dbde2 100644 --- a/src/kw/status.rs +++ b/src/kw/status.rs @@ -3,6 +3,9 @@ //! `AppState` never owns job state; it polls (`GetStatus`) or watches //! (`WatchStatus`) these snapshots and projects them into the view model. +// The actor that constructs/reads these is unix-only. +#![cfg_attr(not(unix), allow(dead_code))] + use std::path::PathBuf; #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/src/main.rs b/src/main.rs index fb3e07b..43a9fe3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,6 +16,8 @@ use clap::Parser; use cli::Cli; use color_eyre::{eyre::eyre, Result}; use config::{bootstrap_parts, ConfigActor}; +#[cfg(unix)] +use infrastructure::process::OsProcess; use infrastructure::{ env::OsEnv, file_system::{FileSystemTrait, OsFileSystem}, @@ -25,6 +27,8 @@ use infrastructure::{ terminal::init, }; use input::{actor::InputActor, event::InputEvent}; +#[cfg(unix)] +use kw::actor::KwActor; use kw::history::{FileKwHistoryStore, KwHistoryStore}; use lore::{ application::{actor::LoreApiActor, cache::CacheTtl, service::LoreService}, @@ -101,6 +105,21 @@ async fn main() -> Result<()> { config.data_dir().to_string(), )); + // The kw actor is unix-only because ProcessTrait (process-group kill) + // is; everywhere else there is no handle and App falls back to writing + // apply history directly. + #[cfg(unix)] + let kw_handle = Some(KwActor::spawn( + kw_history.clone(), + Arc::new(OsProcess), + shell_arc.clone(), + fs_arc.clone(), + Arc::new(OsEnv), + format!("{}/kw_logs", config.cache_dir()).into(), + )); + #[cfg(not(unix))] + let kw_handle = None; + let render = RenderActor::spawn(Box::new(ShellRenderService::new(shell_arc.clone()))); let lore_api = LoreApiActor::spawn(LoreService::new( @@ -132,6 +151,7 @@ async fn main() -> Result<()> { lore_api.clone(), render.clone(), kw_history.clone(), + kw_handle.clone(), )?; let (app_input_tx, app_input_rx) = mpsc::channel::(64); let input_handle = InputActor::spawn(terminal_handle.clone(), app.input_context()); @@ -144,11 +164,13 @@ async fn main() -> Result<()> { // Shutdown ordering: // 1. AppActor — exits when the user quits (input channel closes) // 2. InputActor — no further terminal input is needed once App is gone - // 3. ConfigActor — no further configuration requests once App is gone - // 4. LoreApiActor — no further requests once App is gone - // 5. RenderActor — no further requests once App is gone - // 6. UiActor — no further scene builds once App is gone - // 7. TerminalActor — restores the terminal last so the screen stays usable + // 3. KwActor — no further kw requests once App is gone; kills any + // running job's process group before stopping + // 4. ConfigActor — no further configuration requests once App is gone + // 5. LoreApiActor — no further requests once App is gone + // 6. RenderActor — no further requests once App is gone + // 7. UiActor — no further scene builds once App is gone + // 8. TerminalActor — restores the terminal last so the screen stays usable // during the steps above AppActor::spawn( app, @@ -163,6 +185,9 @@ async fn main() -> Result<()> { .shutdown() .await .map_err(|e| eyre!("{e}"))?; + if let Some(kw_handle) = kw_handle { + kw_handle.shutdown().await; + } config_handle.shutdown().await; lore_api.shutdown().await; render.shutdown().await; From d50a1349f69c58197aae363d9bbc64b5e9aabaaf Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Thu, 20 Aug 2026 13:41:30 -0300 Subject: [PATCH 4/6] fix(kw): wait for job kill on shutdown and escalate stuck cancels This commit makes Shutdown wait until kill escalation finishes, so tearing down the runtime cannot orphan kw's process group. Cancel escalates from SIGTERM to SIGKILL, and a process that exits before the signal lands keeps its real outcome instead of being reported as cancelled. The pre-job branch is recorded at accept time so RestorePreviousBranch has a target. This commit is part of the kw integration's step 4. Signed-off-by: lorenzoberts --- src/app/integration_tests/patchset_actions.rs | 40 ++- src/infrastructure/process/fake.rs | 34 ++- src/infrastructure/process/mod.rs | 15 +- src/infrastructure/process/trait.rs | 6 +- src/kw/actor.rs | 241 +++++++++++++++--- src/kw/errors.rs | 6 +- src/kw/handle.rs | 9 +- src/kw/messages.rs | 8 +- src/kw/status.rs | 2 + src/main.rs | 4 +- 10 files changed, 305 insertions(+), 60 deletions(-) diff --git a/src/app/integration_tests/patchset_actions.rs b/src/app/integration_tests/patchset_actions.rs index 8070d1e..72b0d78 100644 --- a/src/app/integration_tests/patchset_actions.rs +++ b/src/app/integration_tests/patchset_actions.rs @@ -67,6 +67,7 @@ async fn apply_success_sets_success_popup_and_resets_apply_action() { "Current branch: 'patchset-", ], ); + shutdown_kw(&app).await; } #[tokio::test] @@ -97,11 +98,14 @@ async fn apply_success_switches_back_when_stay_disabled() { "Patchset Apply Success", &["Current branch: 'feature'"], ); - let calls = calls.lock().unwrap(); - assert_eq!( - command(&["git", "-C", KERNEL_TREE_PATH, "switch", "feature"]), - calls[6] - ); + { + let calls = calls.lock().unwrap(); + assert_eq!( + command(&["git", "-C", KERNEL_TREE_PATH, "switch", "feature"]), + calls[6] + ); + } + shutdown_kw(&app).await; } #[tokio::test] @@ -126,11 +130,14 @@ async fn apply_failure_sets_failure_popup_and_resets_apply_action() { "Patchset Apply Fail", &["`git am` failed", "feature", "apply failed"], ); - let calls = calls.lock().unwrap(); - assert_eq!( - command(&["git", "-C", KERNEL_TREE_PATH, "am", "--abort"]), - calls[6] - ); + { + let calls = calls.lock().unwrap(); + assert_eq!( + command(&["git", "-C", KERNEL_TREE_PATH, "am", "--abort"]), + calls[6] + ); + } + shutdown_kw(&app).await; } #[tokio::test] @@ -172,6 +179,7 @@ async fn apply_success_records_apply_history() { "Patchset Apply Success", &["applied successfully"], ); + shutdown_kw(&app).await; } #[tokio::test] @@ -211,6 +219,7 @@ async fn apply_success_with_history_write_failure_keeps_success_popup() { "inspect or delete that file", ], ); + shutdown_kw(&app).await; } #[tokio::test] @@ -239,6 +248,7 @@ async fn apply_failure_does_not_record_history() { app.consolidate_patchset_actions().await.unwrap(); assert_info_popup_contains(app.state.popup.as_ref(), "Patchset Apply Fail", &[]); + shutdown_kw(&app).await; } #[tokio::test] @@ -271,6 +281,7 @@ async fn reviewed_reply_success_records_persists_and_resets_reply_action() { .clone() .expect("reviewed state should be persisted"); assert_eq!(HashSet::from([0]), saved[&message_id]); + shutdown_kw(&app).await; } #[tokio::test] @@ -300,6 +311,7 @@ async fn reviewed_reply_failure_does_not_record_failed_index() { .clone() .expect("reviewed state should be persisted"); assert!(saved[&message_id].is_empty()); + shutdown_kw(&app).await; } fn app_with_apply_details(fs: MockFileSystemTrait, shell: MockShellTrait) -> App { @@ -313,6 +325,14 @@ fn app_with_apply_details(fs: MockFileSystemTrait, shell: MockShellTrait) -> App ) } +/// Shuts down the KwActor the app was wired with, instead of relying on +/// the test runtime aborting it at drop. +async fn shutdown_kw(app: &App) { + if let Some(kw) = &app.services.kw { + kw.shutdown().await; + } +} + fn app_with_reviewed_reply_details(shell: MockShellTrait, lore_api: LoreApiHandle) -> App { app_with_details( MockFileSystemTrait::new(), diff --git a/src/infrastructure/process/fake.rs b/src/infrastructure/process/fake.rs index 4d6c1af..1cb84eb 100644 --- a/src/infrastructure/process/fake.rs +++ b/src/infrastructure/process/fake.rs @@ -41,6 +41,11 @@ pub struct FakeControl { struct FakeState { raw_status: Option, killed: bool, + force_killed: bool, + /// When set, `kill()` (SIGTERM) is recorded but the process keeps + /// "running": the model for a process group that ignores SIGTERM, so + /// tests can exercise the SIGKILL escalation. + ignores_sigterm: bool, } impl FakeControl { @@ -66,6 +71,10 @@ impl FakeControl { pub fn was_killed(&self) -> bool { self.state.lock().unwrap().killed } + + pub fn was_force_killed(&self) -> bool { + self.state.lock().unwrap().force_killed + } } struct FakeSpawn { @@ -80,6 +89,7 @@ struct FakeSpawn { pub struct FakeProcess { spawns: Mutex>, refuse_spawns: AtomicBool, + ignore_sigterm: AtomicBool, } impl FakeProcess { @@ -94,6 +104,12 @@ impl FakeProcess { self.refuse_spawns.store(refuse, Ordering::Relaxed); } + /// Make subsequently spawned processes ignore `kill()` (SIGTERM): the + /// kill is recorded but they keep "running" until `force_kill()`. + pub fn ignore_sigterm(&self, ignore: bool) { + self.ignore_sigterm.store(ignore, Ordering::Relaxed); + } + pub fn spawned(&self) -> Vec { self.spawns .lock() @@ -135,6 +151,8 @@ impl ProcessTrait for FakeProcess { state: Mutex::new(FakeState { raw_status: None, killed: false, + force_killed: false, + ignores_sigterm: self.ignore_sigterm.load(Ordering::Relaxed), }), notify: Notify::new(), log_path: log_path.to_path_buf(), @@ -177,7 +195,21 @@ impl RunningProcess for FakeRunningProcess { // process is a successful no-op, not a kill. if state.raw_status.is_none() { state.killed = true; - state.raw_status = Some(Signal::SIGTERM as i32); + if !state.ignores_sigterm { + state.raw_status = Some(Signal::SIGTERM as i32); + drop(state); + self.control.notify.notify_one(); + } + } + Ok(()) + } + + fn force_kill(&mut self) -> Result<(), ProcessError> { + let mut state = self.control.state.lock().unwrap(); + // SIGKILL cannot be ignored: even a SIGTERM-stubborn process dies. + if state.raw_status.is_none() { + state.force_killed = true; + state.raw_status = Some(Signal::SIGKILL as i32); drop(state); self.control.notify.notify_one(); } diff --git a/src/infrastructure/process/mod.rs b/src/infrastructure/process/mod.rs index b97a88e..f1e6d3a 100644 --- a/src/infrastructure/process/mod.rs +++ b/src/infrastructure/process/mod.rs @@ -33,7 +33,6 @@ use tokio::process::{Child, Command}; use crate::infrastructure::shell::ShellCommand; -#[allow(dead_code)] pub struct OsProcess; impl ProcessTrait for OsProcess { @@ -88,7 +87,19 @@ impl RunningProcess for OsRunningProcess { } fn kill(&mut self) -> Result<(), ProcessError> { - match killpg(self.pgid, Signal::SIGTERM) { + self.signal_group(Signal::SIGTERM) + } + + fn force_kill(&mut self) -> Result<(), ProcessError> { + self.signal_group(Signal::SIGKILL) + } +} + +impl OsRunningProcess { + /// ESRCH tolerance: signaling an already-gone group is a successful + /// no-op, not an error. + fn signal_group(&self, signal: Signal) -> Result<(), ProcessError> { + match killpg(self.pgid, signal) { Ok(()) | Err(Errno::ESRCH) => Ok(()), Err(errno) => Err(ProcessError::IoError(io::Error::from(errno))), } diff --git a/src/infrastructure/process/trait.rs b/src/infrastructure/process/trait.rs index c9204c6..5ce8070 100644 --- a/src/infrastructure/process/trait.rs +++ b/src/infrastructure/process/trait.rs @@ -12,7 +12,6 @@ pub enum ProcessError { IoError(#[from] io::Error), } -#[allow(dead_code)] #[automock] pub trait ProcessTrait: Send + Sync { /// Spawn `cmd` with `cwd` as its working directory, redirecting stdout and @@ -29,7 +28,6 @@ pub trait ProcessTrait: Send + Sync { // `automock` must stay the outermost attribute: with `async_trait` listed // first, the generated mock's async methods return an unusable type. -#[allow(dead_code)] #[automock] #[async_trait] pub trait RunningProcess: Send { @@ -47,4 +45,8 @@ pub trait RunningProcess: Send { /// to cancel should `tokio::select!` between `wait()` and its cancel /// signal, then call `kill()`. fn kill(&mut self) -> Result<(), ProcessError>; + + /// Send SIGKILL to the whole process group. The escalation rung for a + /// group that ignored `kill()`'s SIGTERM; same ESRCH tolerance. + fn force_kill(&mut self) -> Result<(), ProcessError>; } diff --git a/src/kw/actor.rs b/src/kw/actor.rs index 57cc040..5d07f63 100644 --- a/src/kw/actor.rs +++ b/src/kw/actor.rs @@ -8,7 +8,7 @@ //! [`crate::kw::readiness`] and records applies through the shared //! [`KwHistoryStore`](crate::kw::history::KwHistoryStore). -use std::{ops::ControlFlow, path::PathBuf, process::ExitStatus, sync::Arc}; +use std::{ops::ControlFlow, path::PathBuf, process::ExitStatus, sync::Arc, time::Duration}; use tokio::{ spawn, @@ -35,6 +35,14 @@ use crate::{ pub const DEFAULT_KW_CHANNEL_SIZE: usize = 16; +/// Grace periods for the cancel escalation ladder: SIGTERM the process +/// group, wait, SIGKILL, wait, then give up. Giving up still terminates the +/// job from the actor's point of view — a group that ignores both signals +/// must not wedge the actor into refusing every later Start with +/// JobAlreadyRunning for the rest of the session. +const TERM_GRACE: Duration = Duration::from_secs(3); +const KILL_GRACE: Duration = Duration::from_secs(2); + /// How a job's process ended, as observed by the detached task that owns /// the process handle. enum JobOutcome { @@ -57,6 +65,10 @@ struct JobState { kernel_tree_id: String, branch: String, log_path: PathBuf, + /// HEAD at accept time, so RestorePreviousBranch can offer to switch + /// back after the job. Read once the checkout policy lands. + #[allow(dead_code)] + pre_job_branch: Option, /// `None` once a cancel has been requested; a second `Cancel` is an /// idempotent ack. cancel_tx: Option>, @@ -103,7 +115,6 @@ impl KwActor { } } - #[allow(clippy::too_many_arguments)] pub fn spawn( history: Arc, process: Arc, @@ -126,7 +137,7 @@ impl KwActor { tokio::select! { message = self.rx.recv() => { let Some(message) = message else { break }; - if let ControlFlow::Break(()) = self.handle_message(message) { + if let ControlFlow::Break(()) = self.handle_message(message).await { break; } } @@ -138,7 +149,7 @@ impl KwActor { tracing::info!("kw actor stopped"); } - fn handle_message(&mut self, message: KwMessage) -> ControlFlow<()> { + async fn handle_message(&mut self, message: KwMessage) -> ControlFlow<()> { let message_name = message.name(); tracing::debug!(message = message_name, "kw request received"); @@ -197,15 +208,25 @@ impl KwActor { send_kw_reply(message_name, reply, Err(KwError::NoRecordedBranch)); ControlFlow::Continue(()) } - KwMessage::Shutdown => { + KwMessage::Shutdown { reply } => { // Quit-prompt cooperation: the app asks whether a job is // running before quitting; if it quits anyway, the job's // process group is killed here. The log may capture partial - // output and the tree a partial build. - if let Some(cancel) = self.job.as_mut().and_then(|job| job.cancel_tx.take()) { + // output and the tree a partial build. The reply is held + // until the kill escalation has run its (bounded) course, + // so a caller tearing down the runtime afterwards cannot + // leave orphaned kw processes behind. + if self.job.is_some() { tracing::info!("killing kw job process group during shutdown"); - let _ = cancel.send(()); + self.request_cancel().ok(); + while self.job.is_some() { + match self.job_event_rx.recv().await { + Some(event) => self.handle_job_event(event), + None => break, + } + } } + reply.send(()).ok(); tracing::debug!("kw actor shutting down"); ControlFlow::Break(()) } @@ -218,21 +239,36 @@ impl KwActor { /// /// The argv is the skeleton's minimal `kw build --alert=n`; the real /// argv builder (reserved flags, extra-args merge) and the checkout - /// policy land with the build step. + /// policy land with the build step — as does the readiness-based + /// refusal from the message protocol (refuse Start when readiness + /// fails); until then the only caller is the KwOps screen, which + /// surfaces readiness before offering Start. The `branch` carried by + /// the Running status is the *requested* branch; the checkout policy is + /// what will make the tree actually sit on it. fn start_job(&mut self, kind: KwJobKind, request: StartRequest) -> Result<(), KwStartError> { if self.job.is_some() { return Err(KwStartError::JobAlreadyRunning); } self.fs.create_dir_all(&self.kw_log_dir)?; + // Millisecond suffix: two jobs started within the same second must + // not share a log file — spawn truncates it. let log_path = self.kw_log_dir.join(format!( "build-{}.log", - chrono::Utc::now().format("%Y%m%d-%H%M%S") + chrono::Utc::now().format("%Y%m%d-%H%M%S-%3f") )); let cmd = ShellCommand::new("kw").args(["build", "--alert=n"]); let cwd = PathBuf::from(request.tree.path()); let process = self.process.spawn(&cmd, &cwd, &log_path)?; + // Recorded before the job starts so RestorePreviousBranch can offer + // to switch back; an unprobed HEAD (detached, or not a git repo) + // records nothing rather than a wrong branch. + let pre_job_branch = match self.head_branch(&request.tree) { + branch if branch.is_empty() => None, + branch => Some(branch), + }; + let (cancel_tx, cancel_rx) = oneshot::channel(); spawn(run_job(process, cancel_rx, self.job_event_tx.clone())); @@ -249,6 +285,7 @@ impl KwActor { kernel_tree_id: request.kernel_tree_id.clone(), branch: request.branch.clone(), log_path: log_path.clone(), + pre_job_branch, cancel_tx: Some(cancel_tx), }); self.set_status(KwJobStatus::Running { @@ -287,7 +324,11 @@ impl KwActor { branch = job.branch, "kw job succeeded" ); - KwJobStatus::Succeeded { kind: job.kind } + KwJobStatus::Succeeded { + kind: job.kind, + kernel_tree_id: job.kernel_tree_id, + log_path: job.log_path, + } } JobOutcome::Exited(exit) => { tracing::warn!( @@ -376,7 +417,7 @@ impl KwActor { } /// Owns the spawned process until it ends: waits on it, or — when the -/// cancel signal fires — kills the whole process group and reaps it. The +/// cancel signal fires — runs the kill escalation and reaps it. The /// `wait()` future is dropped before the cancel arm's body runs, releasing /// the mutable borrow so `kill()` can be called. Reports the outcome back /// to the actor over the internal event channel. @@ -391,17 +432,45 @@ async fn run_job( Err(error) => JobOutcome::WaitFailed(error), }, // A dropped sender (actor shutting down) cancels the job too. - _ = &mut cancel_rx => { - if let Err(error) = process.kill() { - tracing::warn!(%error, "failed to kill kw job process group"); - } - let _ = process.wait().await; - JobOutcome::Cancelled - } + _ = &mut cancel_rx => cancel_job(&mut *process).await, }; events.send(JobEvent::Finished(outcome)).await.ok(); } +/// Cancel escalation ladder: SIGTERM the group, wait a grace period, +/// SIGKILL, wait again, then give up. Giving up still reports Cancelled and +/// lets the actor clear its job state — a process group that ignores both +/// signals must not wedge the actor into refusing every later Start with +/// `JobAlreadyRunning` for the rest of the session. +async fn cancel_job(process: &mut dyn RunningProcess) -> JobOutcome { + if let Err(error) = process.kill() { + tracing::warn!(%error, "failed to SIGTERM kw job process group"); + } + match tokio::time::timeout(TERM_GRACE, process.wait()).await { + // A successful reap after the signal means the process had already + // exited before it landed (kill of a dead group is a no-op, and a + // signaled process reaps as signal-terminated, not as success): + // report the real outcome, not Cancelled. + Ok(Ok(status)) if status.success() => JobOutcome::Exited(status), + Ok(_) => JobOutcome::Cancelled, + Err(_) => { + tracing::warn!("kw job ignored SIGTERM; escalating to SIGKILL"); + if let Err(error) = process.force_kill() { + tracing::warn!(%error, "failed to SIGKILL kw job process group"); + } + match tokio::time::timeout(KILL_GRACE, process.wait()).await { + Ok(_) => JobOutcome::Cancelled, + Err(_) => { + tracing::warn!( + "kw job process group could not be reaped after SIGKILL; giving up" + ); + JobOutcome::Cancelled + } + } + } + } +} + fn send_kw_reply( message_name: &'static str, reply: oneshot::Sender>, @@ -513,8 +582,11 @@ mod tests { /// Spawns the actor with the already-configured mocks (mockall /// expectations need `&mut`, so they are set before the mocks move - /// behind `Arc`s). + /// behind `Arc`s). The log dir is a real unique temp dir even though + /// these tests never start a job, so a future test that accidentally + /// does cannot share a fixed path with the job tests. fn spawn_test_actor( + test_name: &str, history: MockKwHistoryStore, shell: MockShellTrait, fs: MockFileSystemTrait, @@ -526,21 +598,31 @@ mod tests { Arc::new(shell), Arc::new(fs), Arc::new(env), - PathBuf::from("/tmp/patch-hub-test-kw-logs"), + tmp_log_dir(test_name), ) } /// Spawns the actor with a real temp log dir and exposes the - /// [`FakeProcess`] so tests drive the "running" process. - fn spawn_job_actor(test_name: &str) -> (KwHandle, Arc, PathBuf) { + /// [`FakeProcess`] so tests drive the "running" process. The shell mock + /// answers the pre-job HEAD probe. + fn spawn_job_actor_with_fs( + test_name: &str, + fs: MockFileSystemTrait, + ) -> (KwHandle, Arc, PathBuf) { let process = Arc::new(FakeProcess::new()); let log_dir = tmp_log_dir(test_name); - let mut fs = MockFileSystemTrait::new(); - fs.expect_create_dir_all().returning(|_| Ok(())); + let mut shell = MockShellTrait::new(); + shell.expect_execute().returning(|_| { + Ok(ShellOutput { + stdout: b"master\n".to_vec(), + stderr: Vec::new(), + success: true, + }) + }); let handle = KwActor::spawn( Arc::new(MockKwHistoryStore::new()), process.clone(), - Arc::new(MockShellTrait::new()), + Arc::new(shell), Arc::new(fs), Arc::new(MockEnvTrait::new()), log_dir.clone(), @@ -548,6 +630,12 @@ mod tests { (handle, process, log_dir) } + fn spawn_job_actor(test_name: &str) -> (KwHandle, Arc, PathBuf) { + let mut fs = MockFileSystemTrait::new(); + fs.expect_create_dir_all().returning(|_| Ok(())); + spawn_job_actor_with_fs(test_name, fs) + } + fn apply_record() -> KwApplyRecord { KwApplyRecord { message_id: "msg-1".to_string(), @@ -569,6 +657,7 @@ mod tests { .times(1) .returning(|_| Ok(())); let handle = spawn_test_actor( + "record-apply", history, MockShellTrait::new(), MockFileSystemTrait::new(), @@ -587,6 +676,7 @@ mod tests { .times(1) .returning(|_| Err(FileSystemError::IoError(io::Error::other("disk full")))); let handle = spawn_test_actor( + "record-apply-error", history, MockShellTrait::new(), MockFileSystemTrait::new(), @@ -602,6 +692,7 @@ mod tests { #[tokio::test] async fn get_status_reports_idle_before_any_job() { let handle = spawn_test_actor( + "idle-status", MockKwHistoryStore::new(), MockShellTrait::new(), MockFileSystemTrait::new(), @@ -617,6 +708,7 @@ mod tests { #[tokio::test] async fn watch_status_receiver_sees_current_snapshot() { let handle = spawn_test_actor( + "watch-idle", MockKwHistoryStore::new(), MockShellTrait::new(), MockFileSystemTrait::new(), @@ -667,7 +759,7 @@ mod tests { success: true, }) }); - let handle = spawn_test_actor(history, shell, fs, env); + let handle = spawn_test_actor("readiness", history, shell, fs, env); let readiness = handle.get_readiness("mainline", &tree).await.unwrap(); @@ -682,11 +774,13 @@ mod tests { /// Waits until the status leaves `Idle`/`Running` and returns the /// terminal status. The receiver may have observed the `Running` - /// transition first, so a single `changed()` is not enough. + /// transition first, so a single `changed()` is not enough. The timeout + /// is generous because the SIGKILL-escalation test waits out the + /// SIGTERM grace period. async fn wait_for_terminal_status( watch: &mut watch::Receiver, ) -> KwJobStatus { - tokio::time::timeout(Duration::from_secs(1), async { + tokio::time::timeout(Duration::from_secs(10), async { loop { let status = watch.borrow().job.clone(); if !matches!(status, KwJobStatus::Idle | KwJobStatus::Running { .. }) { @@ -735,11 +829,15 @@ mod tests { process.last_child().finish(0); let mut watch = handle.watch_status().await.unwrap(); let status = wait_for_terminal_status(&mut watch).await; - assert_eq!( - KwJobStatus::Succeeded { - kind: KwJobKind::Build - }, - status + assert!( + matches!( + status, + KwJobStatus::Succeeded { + kind: KwJobKind::Build, + .. + } + ), + "unexpected status: {status:?}" ); handle.shutdown().await; @@ -821,12 +919,9 @@ mod tests { let (handle, process, log_dir) = spawn_job_actor("shutdown-kill"); handle.start_build(start_request()).await.unwrap(); + // shutdown() returns only after the kill escalation has completed. handle.shutdown().await; - // shutdown() only enqueues the message; the actor being gone - // proves the Shutdown (and its kill) was processed. - let err = handle.get_status().await.unwrap_err(); - assert!(matches!(err, KwError::ActorUnavailable(_))); assert!(process.last_child().was_killed()); std::fs::remove_dir_all(&log_dir).unwrap(); } @@ -852,6 +947,7 @@ mod tests { #[tokio::test] async fn start_deploy_still_refused_until_the_deploy_step() { let handle = spawn_test_actor( + "deploy-refused", MockKwHistoryStore::new(), MockShellTrait::new(), MockFileSystemTrait::new(), @@ -870,6 +966,7 @@ mod tests { #[tokio::test] async fn cancel_and_restore_without_job_are_immediate_errors() { let handle = spawn_test_actor( + "no-job", MockKwHistoryStore::new(), MockShellTrait::new(), MockFileSystemTrait::new(), @@ -892,6 +989,7 @@ mod tests { #[tokio::test] async fn shutdown_stops_actor() { let handle = spawn_test_actor( + "shutdown", MockKwHistoryStore::new(), MockShellTrait::new(), MockFileSystemTrait::new(), @@ -903,4 +1001,71 @@ mod tests { assert!(matches!(err, KwError::ActorUnavailable(_))); } + + #[tokio::test] + async fn log_dir_creation_failure_refuses_start_and_stays_idle() { + let mut fs = MockFileSystemTrait::new(); + fs.expect_create_dir_all().returning(|_| { + Err(FileSystemError::IoError(io::Error::other( + "read-only filesystem", + ))) + }); + let (handle, process, log_dir) = spawn_job_actor_with_fs("log-dir-fail", fs); + + let err = handle.start_build(start_request()).await.unwrap_err(); + + assert!(matches!(err, KwStartError::Fs(_))); + assert_eq!(KwJobStatus::Idle, handle.get_status().await.unwrap().job); + assert!(process.spawned().is_empty()); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn cancel_racing_a_successful_exit_reports_success() { + let (handle, process, log_dir) = spawn_job_actor("cancel-race"); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + process.last_child().finish(0); + // Whether the actor processes the exit or the cancel first is + // timing-dependent, but the terminal status must be Succeeded + // either way: the process exited before any signal landed. + let _ = handle.cancel().await; + + let status = wait_for_terminal_status(&mut watch).await; + assert!( + matches!(status, KwJobStatus::Succeeded { .. }), + "expected Succeeded, got {status:?}" + ); + // Killing an already-finished process is a no-op, not a kill. + assert!(!process.last_child().was_killed()); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn cancel_escalates_to_sigkill_when_sigterm_is_ignored() { + let (handle, process, log_dir) = spawn_job_actor("sigkill-escalation"); + process.ignore_sigterm(true); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + handle.cancel().await.unwrap(); + + // Waits out the SIGTERM grace period before the escalation. + let status = wait_for_terminal_status(&mut watch).await; + assert!( + matches!(status, KwJobStatus::Cancelled { .. }), + "expected Cancelled, got {status:?}" + ); + let child = process.last_child(); + assert!(child.was_killed()); + assert!(child.was_force_killed()); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } } diff --git a/src/kw/errors.rs b/src/kw/errors.rs index aa8c16f..30179b4 100644 --- a/src/kw/errors.rs +++ b/src/kw/errors.rs @@ -3,8 +3,10 @@ use thiserror::Error; +#[cfg(unix)] +use crate::infrastructure::process::ProcessError; use crate::{ - infrastructure::{file_system::FileSystemError, process::ProcessError, shell::ShellError}, + infrastructure::{file_system::FileSystemError, shell::ShellError}, kw::readiness::KwReadinessError, }; @@ -35,6 +37,8 @@ pub enum KwStartError { JobAlreadyRunning, #[error("kw jobs are not supported yet")] NotImplemented, + // Spawning a process is unix-only (ProcessTrait is cfg(unix)). + #[cfg(unix)] #[error("failed to spawn the kw process: {0}")] Spawn(#[from] ProcessError), #[error("filesystem error: {0}")] diff --git a/src/kw/handle.rs b/src/kw/handle.rs index 8e20c77..588d9c9 100644 --- a/src/kw/handle.rs +++ b/src/kw/handle.rs @@ -79,9 +79,14 @@ impl KwHandle { } /// Signals the actor to stop processing messages and exit its run loop. - /// A running job is killed before the actor stops. + /// A running job's process group is killed first; this returns only + /// after the kill escalation has completed, so teardown can drop the + /// runtime without orphaning kw's child processes. pub async fn shutdown(&self) { - self.tx.send(KwMessage::Shutdown).await.ok(); + let (reply, rx) = oneshot::channel(); + if self.tx.send(KwMessage::Shutdown { reply }).await.is_ok() { + rx.await.ok(); + } } async fn request_result( diff --git a/src/kw/messages.rs b/src/kw/messages.rs index ca99c91..8f31d2b 100644 --- a/src/kw/messages.rs +++ b/src/kw/messages.rs @@ -66,7 +66,11 @@ pub enum KwMessage { RestorePreviousBranch { reply: oneshot::Sender>, }, - Shutdown, + /// Unlike the other actors' bare `Shutdown`, this one replies — the + /// `TerminalMessage::Shutdown` convention. Teardown must know the + /// running job's process group was actually killed before the runtime + /// is dropped; a fire-and-forget message leaves that to scheduler luck. + Shutdown { reply: oneshot::Sender<()> }, } impl KwMessage { @@ -81,7 +85,7 @@ impl KwMessage { KwMessage::WatchStatus { .. } => "WatchStatus", KwMessage::GetReadiness { .. } => "GetReadiness", KwMessage::RestorePreviousBranch { .. } => "RestorePreviousBranch", - KwMessage::Shutdown => "Shutdown", + KwMessage::Shutdown { .. } => "Shutdown", } } } diff --git a/src/kw/status.rs b/src/kw/status.rs index f2dbde2..a8a1077 100644 --- a/src/kw/status.rs +++ b/src/kw/status.rs @@ -40,6 +40,8 @@ pub enum KwJobStatus { }, Succeeded { kind: KwJobKind, + kernel_tree_id: String, + log_path: PathBuf, }, Failed { kind: KwJobKind, diff --git a/src/main.rs b/src/main.rs index 43a9fe3..32f6777 100644 --- a/src/main.rs +++ b/src/main.rs @@ -40,7 +40,7 @@ use lore::{ }, }; use render::{actor::RenderActor, ShellRenderService}; -use std::{ops::ControlFlow, sync::Arc}; +use std::{ops::ControlFlow, path::Path, sync::Arc}; use terminal::{actor::TerminalActor, session::CrosstermTerminalSession}; use tokio::sync::mpsc; use tracing::{event, Level}; @@ -115,7 +115,7 @@ async fn main() -> Result<()> { shell_arc.clone(), fs_arc.clone(), Arc::new(OsEnv), - format!("{}/kw_logs", config.cache_dir()).into(), + Path::new(config.cache_dir()).join("kw_logs"), )); #[cfg(not(unix))] let kw_handle = None; From 73d680457580065a11993f1d89dc05301d9088bb Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Thu, 20 Aug 2026 13:52:36 -0300 Subject: [PATCH 5/6] fix(kw): keep real job outcomes after cancel and probe branch before spawn This commit maps a post-cancel reap by whether the process was signal-terminated, so a build that exits on its own keeps its real outcome. The pre-job branch probe moves above spawn so it records the branch from before the job starts, not from after the process is already running. This commit completes the kw integration's step 4. Signed-off-by: lorenzoberts --- .github/workflows/build_and_unit_test.yml | 38 ++++++++ Cargo.toml | 2 +- src/kw/actor.rs | 100 ++++++++++++++++++---- src/kw/handle.rs | 10 +-- src/kw/status.rs | 1 + 5 files changed, 124 insertions(+), 27 deletions(-) diff --git a/.github/workflows/build_and_unit_test.yml b/.github/workflows/build_and_unit_test.yml index 716ffcd..b788c92 100644 --- a/.github/workflows/build_and_unit_test.yml +++ b/.github/workflows/build_and_unit_test.yml @@ -10,9 +10,13 @@ on: - labeled paths: - src/** + - Cargo.toml + - Cargo.lock push: paths: - src/** + - Cargo.toml + - Cargo.lock branches: - master - unstable @@ -50,3 +54,37 @@ jobs: printf '\e[1;33m\tPLEASE, SOLVE THEM LOCALLY W/ `cargo test`\e[0m\n' printf '\e[1;33m\t==========================================\n\e[0m' exit 1 + + # The kw actor and the process infrastructure it uses are unix-only + # (#[cfg(unix)]). This job exists so a cfg leak — non-gated code + # referencing them — fails mechanically instead of in review. + check-non-unix: + runs-on: ubuntu-latest + timeout-minutes: 4 + if: '!github.event.pull_request.draft' + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.sha }} + + - name: Update rustup and install rustc and cargo + shell: bash + run: | + rustup update + rustup install stable + + # rustls/ring needs a C toolchain for the target; windows-gnu builds + # with mingw on Ubuntu, unlike the MSVC target. + - name: Install the mingw cross toolchain + shell: bash + run: | + sudo apt-get update + sudo apt-get install -y gcc-mingw-w64-x86-64 + + - name: Check non-unix compilation + shell: bash + env: + CARGO_TARGET_X86_64_PC_WINDOWS_GNU_LINKER: x86_64-w64-mingw32-gcc + run: | + rustup target add x86_64-pc-windows-gnu + cargo check --target x86_64-pc-windows-gnu --verbose diff --git a/Cargo.toml b/Cargo.toml index d9ffd02..008a705 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,7 @@ chrono = "0.4.41" ansi-to-tui = "7.0.0" which = "8.0.0" ureq = { version = "3.0.12", features = ["rustls"] } -tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync", "process", "time"] } +tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync", "process", "time", "test-util"] } async-trait = "0.1" nix = { version = "0.31", features = ["signal"] } base64 = "0.22" diff --git a/src/kw/actor.rs b/src/kw/actor.rs index 5d07f63..c02f60c 100644 --- a/src/kw/actor.rs +++ b/src/kw/actor.rs @@ -7,6 +7,13 @@ //! kernel build. The actor composes the readiness probes from //! [`crate::kw::readiness`] and records applies through the shared //! [`KwHistoryStore`](crate::kw::history::KwHistoryStore). +//! +//! Tracked note (same class as the integration plan's §2.1i): the actor +//! runs quick blocking calls inline in its async task — `create_dir_all`, +//! the HEAD-probe `git` call, and the history store's atomic write — per +//! the ConfigActor precedent. They are all milliseconds-scale; if a real +//! stall ever shows up while a long job runs, they should move behind +//! `spawn_blocking` like the other actors' heavy work. use std::{ops::ControlFlow, path::PathBuf, process::ExitStatus, sync::Arc, time::Duration}; @@ -220,6 +227,9 @@ impl KwActor { tracing::info!("killing kw job process group during shutdown"); self.request_cancel().ok(); while self.job.is_some() { + // `None` is unreachable — the actor holds + // `job_event_tx`, so the channel never closes — + // but break defensively rather than spin. match self.job_event_rx.recv().await { Some(event) => self.handle_job_event(event), None => break, @@ -250,6 +260,17 @@ impl KwActor { return Err(KwStartError::JobAlreadyRunning); } + // Probed before anything touches the tree: once the checkout policy + // lands, `git switch ` goes between this probe and + // the spawn, and the probe must still capture the pre-job HEAD or + // RestorePreviousBranch would "restore" the branch the job switched + // to. An unprobed HEAD (detached, or not a git repo) records + // nothing rather than a wrong branch. + let pre_job_branch = match self.head_branch(&request.tree) { + branch if branch.is_empty() => None, + branch => Some(branch), + }; + self.fs.create_dir_all(&self.kw_log_dir)?; // Millisecond suffix: two jobs started within the same second must // not share a log file — spawn truncates it. @@ -261,14 +282,6 @@ impl KwActor { let cwd = PathBuf::from(request.tree.path()); let process = self.process.spawn(&cmd, &cwd, &log_path)?; - // Recorded before the job starts so RestorePreviousBranch can offer - // to switch back; an unprobed HEAD (detached, or not a git repo) - // records nothing rather than a wrong branch. - let pre_job_branch = match self.head_branch(&request.tree) { - branch if branch.is_empty() => None, - branch => Some(branch), - }; - let (cancel_tx, cancel_rx) = oneshot::channel(); spawn(run_job(process, cancel_rx, self.job_event_tx.clone())); @@ -327,6 +340,7 @@ impl KwActor { KwJobStatus::Succeeded { kind: job.kind, kernel_tree_id: job.kernel_tree_id, + branch: job.branch, log_path: job.log_path, } } @@ -447,19 +461,14 @@ async fn cancel_job(process: &mut dyn RunningProcess) -> JobOutcome { tracing::warn!(%error, "failed to SIGTERM kw job process group"); } match tokio::time::timeout(TERM_GRACE, process.wait()).await { - // A successful reap after the signal means the process had already - // exited before it landed (kill of a dead group is a no-op, and a - // signaled process reaps as signal-terminated, not as success): - // report the real outcome, not Cancelled. - Ok(Ok(status)) if status.success() => JobOutcome::Exited(status), - Ok(_) => JobOutcome::Cancelled, + Ok(outcome) => outcome_after_cancel(outcome), Err(_) => { tracing::warn!("kw job ignored SIGTERM; escalating to SIGKILL"); if let Err(error) = process.force_kill() { tracing::warn!(%error, "failed to SIGKILL kw job process group"); } match tokio::time::timeout(KILL_GRACE, process.wait()).await { - Ok(_) => JobOutcome::Cancelled, + Ok(outcome) => outcome_after_cancel(outcome), Err(_) => { tracing::warn!( "kw job process group could not be reaped after SIGKILL; giving up" @@ -471,6 +480,30 @@ async fn cancel_job(process: &mut dyn RunningProcess) -> JobOutcome { } } +/// Maps a reap result observed after a cancel request. A signal-terminated +/// process means our SIGTERM/SIGKILL landed — the job was really cancelled. +/// A plain exit means the process finished on its own before the signal: +/// report the real outcome, because a cancel must not mask a failure the +/// build history (and deploy-alone readiness) needs to see. +/// +/// Known, accepted edges: a process that *traps* our SIGTERM and exits 0 +/// counts as success (kw is bash, so this is possible in principle), and an +/// external signal racing a cancel (e.g. the OOM killer) reads as +/// Cancelled. Both are indistinguishable from the honest cases without +/// comparing who signaled first, and both favor showing the user real +/// output over inventing failures. +fn outcome_after_cancel(result: Result) -> JobOutcome { + use std::os::unix::process::ExitStatusExt; + + match result { + Ok(status) => match status.signal() { + Some(_) => JobOutcome::Cancelled, + None => JobOutcome::Exited(status), + }, + Err(error) => JobOutcome::WaitFailed(error), + } +} + fn send_kw_reply( message_name: &'static str, reply: oneshot::Sender>, @@ -775,8 +808,8 @@ mod tests { /// Waits until the status leaves `Idle`/`Running` and returns the /// terminal status. The receiver may have observed the `Running` /// transition first, so a single `changed()` is not enough. The timeout - /// is generous because the SIGKILL-escalation test waits out the - /// SIGTERM grace period. + /// backstops against a wedged actor; tests that exercise the grace + /// periods run with paused time instead of waiting them out. async fn wait_for_terminal_status( watch: &mut watch::Receiver, ) -> KwJobStatus { @@ -1047,6 +1080,38 @@ mod tests { } #[tokio::test] + async fn cancel_racing_a_failed_exit_reports_failure() { + let (handle, process, log_dir) = spawn_job_actor("cancel-race-fail"); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + process.last_child().finish(2); + // Whether the actor processes the exit or the cancel first is + // timing-dependent, but a plain exit (not signal-terminated) means + // the process failed on its own before the signal landed: the + // terminal status must be Failed either way, so the build history + // records a failure rather than a cancel. + let _ = handle.cancel().await; + + let status = wait_for_terminal_status(&mut watch).await; + assert!( + matches!( + status, + KwJobStatus::Failed { + exit_code: Some(2), + .. + } + ), + "expected Failed(2), got {status:?}" + ); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + // Paused time: the runtime auto-advances through the grace-period + // timers instead of burning wall-clock seconds on them. + #[tokio::test(start_paused = true)] async fn cancel_escalates_to_sigkill_when_sigterm_is_ignored() { let (handle, process, log_dir) = spawn_job_actor("sigkill-escalation"); process.ignore_sigterm(true); @@ -1055,7 +1120,6 @@ mod tests { handle.start_build(start_request()).await.unwrap(); handle.cancel().await.unwrap(); - // Waits out the SIGTERM grace period before the escalation. let status = wait_for_terminal_status(&mut watch).await; assert!( matches!(status, KwJobStatus::Cancelled { .. }), diff --git a/src/kw/handle.rs b/src/kw/handle.rs index 588d9c9..f1611e1 100644 --- a/src/kw/handle.rs +++ b/src/kw/handle.rs @@ -92,20 +92,14 @@ impl KwHandle { async fn request_result( &self, build_message: impl FnOnce(oneshot::Sender>) -> KwMessage, - ) -> Result - where - T: Send + 'static, - { + ) -> Result { self.request(build_message).await? } async fn request( &self, build_message: impl FnOnce(oneshot::Sender) -> KwMessage, - ) -> Result - where - T: Send + 'static, - { + ) -> Result { let (reply, rx) = oneshot::channel(); self.tx .send(build_message(reply)) diff --git a/src/kw/status.rs b/src/kw/status.rs index a8a1077..059fb92 100644 --- a/src/kw/status.rs +++ b/src/kw/status.rs @@ -41,6 +41,7 @@ pub enum KwJobStatus { Succeeded { kind: KwJobKind, kernel_tree_id: String, + branch: String, log_path: PathBuf, }, Failed { From 39b9c3f7aaae056f16616decd8a73aa2d3ec7a2b Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Mon, 24 Aug 2026 14:23:32 -0300 Subject: [PATCH 6/6] docs(kw): describe the actor skeleton without later steps This commit drops plan citations and comments about KwOps, checkout, and deploy that this skeleton does not implement. StartDeploy still replies NotImplemented; the comment now says that. This commit is part of the kw integration's step 4. Signed-off-by: lorenzoberts Co-authored-by: Cursor --- src/kw/actor.rs | 34 +++++++++------------------------- src/kw/handle.rs | 3 --- src/kw/messages.rs | 4 +--- src/kw/status.rs | 3 --- 4 files changed, 10 insertions(+), 34 deletions(-) diff --git a/src/kw/actor.rs b/src/kw/actor.rs index c02f60c..3907f70 100644 --- a/src/kw/actor.rs +++ b/src/kw/actor.rs @@ -8,12 +8,8 @@ //! [`crate::kw::readiness`] and records applies through the shared //! [`KwHistoryStore`](crate::kw::history::KwHistoryStore). //! -//! Tracked note (same class as the integration plan's §2.1i): the actor -//! runs quick blocking calls inline in its async task — `create_dir_all`, -//! the HEAD-probe `git` call, and the history store's atomic write — per -//! the ConfigActor precedent. They are all milliseconds-scale; if a real -//! stall ever shows up while a long job runs, they should move behind -//! `spawn_blocking` like the other actors' heavy work. +//! `create_dir_all`, the HEAD-probe `git` call, and history writes run +//! inline on the actor task. use std::{ops::ControlFlow, path::PathBuf, process::ExitStatus, sync::Arc, time::Duration}; @@ -177,9 +173,7 @@ impl KwActor { ); ControlFlow::Continue(()) } - // Deploy acceptance lands with the deploy step; the - // immediate-reply contract already holds, so callers never - // learn to depend on a blocking reply. + // Not implemented: reply immediately with NotImplemented. KwMessage::StartDeploy { reply, .. } | KwMessage::StartBuildThenDeploy { reply, .. } => { send_start_reply(message_name, reply, Err(KwStartError::NotImplemented)); @@ -247,25 +241,15 @@ impl KwActor { /// caller right after this returns: the job itself keeps running in a /// detached task and is observed via the status snapshot. /// - /// The argv is the skeleton's minimal `kw build --alert=n`; the real - /// argv builder (reserved flags, extra-args merge) and the checkout - /// policy land with the build step — as does the readiness-based - /// refusal from the message protocol (refuse Start when readiness - /// fails); until then the only caller is the KwOps screen, which - /// surfaces readiness before offering Start. The `branch` carried by - /// the Running status is the *requested* branch; the checkout policy is - /// what will make the tree actually sit on it. + /// The argv is `kw build --alert=n`. The `branch` on the Running status + /// is the requested branch; this skeleton does not switch the tree. fn start_job(&mut self, kind: KwJobKind, request: StartRequest) -> Result<(), KwStartError> { if self.job.is_some() { return Err(KwStartError::JobAlreadyRunning); } - // Probed before anything touches the tree: once the checkout policy - // lands, `git switch ` goes between this probe and - // the spawn, and the probe must still capture the pre-job HEAD or - // RestorePreviousBranch would "restore" the branch the job switched - // to. An unprobed HEAD (detached, or not a git repo) records - // nothing rather than a wrong branch. + // Recorded at accept so RestorePreviousBranch can switch back. + // An unprobed HEAD (detached, or not a git repo) records nothing. let pre_job_branch = match self.head_branch(&request.tree) { branch if branch.is_empty() => None, branch => Some(branch), @@ -830,8 +814,8 @@ mod tests { async fn start_build_replies_immediately_and_runs_in_background() { let (handle, process, log_dir) = spawn_job_actor("start-immediate"); - // The §3.1 reply contract: start_build resolves while the spawned - // process is still running (no finish() was ever signaled). + // start_build resolves while the spawned process is still running + // (no finish() was ever signaled). let result = tokio::time::timeout(Duration::from_secs(1), handle.start_build(start_request())) .await diff --git a/src/kw/handle.rs b/src/kw/handle.rs index f1611e1..9530f5e 100644 --- a/src/kw/handle.rs +++ b/src/kw/handle.rs @@ -16,9 +16,6 @@ pub struct KwHandle { tx: mpsc::Sender, } -// Only record_apply/shutdown have a production caller until the KwOps -// screen lands; kept per the CachePolicy precedent -// (src/lore/application/cache.rs). #[allow(dead_code)] impl KwHandle { pub fn new(tx: mpsc::Sender) -> Self { diff --git a/src/kw/messages.rs b/src/kw/messages.rs index 8f31d2b..067b6da 100644 --- a/src/kw/messages.rs +++ b/src/kw/messages.rs @@ -20,8 +20,7 @@ use crate::{ pub struct StartRequest { pub kernel_tree_id: String, pub tree: KernelTree, - /// Branch the job must run on; the checkout policy that gets the tree - /// onto it lands with the build step. + /// Branch the job must run on. pub branch: String, } @@ -35,7 +34,6 @@ pub enum KwMessage { reply: oneshot::Sender>, }, StartDeploy { - // Read once deploy execution lands (CachePolicy precedent). #[allow(dead_code)] request: StartRequest, reply: oneshot::Sender>, diff --git a/src/kw/status.rs b/src/kw/status.rs index 059fb92..3f638aa 100644 --- a/src/kw/status.rs +++ b/src/kw/status.rs @@ -11,8 +11,6 @@ use std::path::PathBuf; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum KwJobKind { Build, - // Constructed once deploy execution lands; kept per the CachePolicy - // precedent (src/lore/application/cache.rs). #[allow(dead_code)] Deploy, #[allow(dead_code)] @@ -23,7 +21,6 @@ pub enum KwJobKind { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum KwPhase { Building, - // Constructed once deploy execution lands (CachePolicy precedent). #[allow(dead_code)] Deploying, }