refactor: reorganize imports and add lint allows

for testability

- Add `ManagedChild` trait abstracting process lifecycle (real
  `TokioChild` + `MockChild` for tests)
- Use Unix process groups (`process_group(0)` + `libc::kill(-pgid,
  SIGKILL)`) for tree-wide termination
- Remove cgroup-related code (`CgroupManager`, `ResourceUsage`,
  `cgroup_path` field)
- Update imports across test files to use `workshop-engine` crate
  directly

workshop-baker/Cargo.lock

@@ -5302,6 +5302,7 @@ dependencies = [
  "chrono",
  "duration-str",
  "lazy_static",
+ "libc",
  "log",
  "os_info",
  "serde",

workshop-baker/Cargo.toml

@@ -15,10 +15,6 @@ path = "src/main.rs"
 default = []
 integration-tests = []
 
-[[test]]
-name = "engine_integration"
-path = "tests/engine_integration.rs"
-
 [dependencies]
 workshop-engine = { path = "../workshop-engine" }
 anyhow = "1.0.100"
@@ -62,6 +58,14 @@ globset = "0.4.18"
 [dev-dependencies]
 criterion = "0.5"
 
+[lints.rust]
+dead_code = "allow"
+unreachable_code = "allow"
+
+[lints.clippy]
+inherent_to_string = "allow"
+non_canonical_partial_ord_impl = "allow"
+
 [[bench]]
 name = "parser_bench"
 harness = false

workshop-baker/src/bake.rs

@@ -1,26 +1,26 @@
-use workshop_engine::{Engine, EventSender, ExecutionContext};
-use crate::bake::parser::Function;
 use crate::bake::constant::{DEFAULT_WORKSPACE, TRIVIAL_SCRIPT_NAME};
 use crate::bake::error::BakeError;
+use crate::bake::parser::Function;
 use crate::cli::Cli;
+use crate::prebake;
 use crate::prebake::PrebakeConfig;
 use crate::prebake::security::get_drop_after;
 use crate::prebake::stage::PrebakeStage;
-use crate::prebake;
 use std::path::{Path, PathBuf};
+use workshop_engine::{Engine, EventSender, ExecutionContext};
 
 mod builder;
 pub mod constant;
 mod decorator;
-pub mod parser;
 pub mod error;
+pub mod parser;
 mod schedule;
 
 pub async fn bake(
     script_path: &Path,
     bake_base_path: &Path,
     prebake_path: &Path,
-    cli: &Cli,
+    _cli: &Cli,
     ctx: &mut ExecutionContext,
     event_tx: EventSender,
 ) -> Result<(), BakeError> {
@@ -43,10 +43,10 @@ pub async fn bake(
     let use_template = true;
 
     let engine = Engine::new();
-    if let Some(env) = &prebake.envvars {
-        if let Some(prebake_env) = &env.bake {
-            ctx.env_vars.extend(prebake_env.clone());
-        }
+    if let Some(env) = &prebake.envvars
+        && let Some(prebake_env) = &env.bake
+    {
+        ctx.env_vars.extend(prebake_env.clone());
     }
 
     if !has_pipeline || trivial {
@@ -57,7 +57,7 @@ pub async fn bake(
         };
         let script =
             builder::build_script(&function, bake_base_path, &workspace, None, use_template)?;
-        let result = engine.execute_script(&script, &ctx, &event_tx).await;
+        let result = engine.execute_script(&script, ctx, &event_tx).await;
         result?;
     } else {
         let sorted_functions = schedule::sort_function(functions.pipeline_functions)?;
@@ -72,7 +72,7 @@ pub async fn bake(
                 None,
                 use_template,
             )?;
-            let result = engine.execute_script(&script, &ctx, &event_tx).await;
+            let result = engine.execute_script(&script, ctx, &event_tx).await;
             result?;
         }
     }
@@ -96,10 +96,10 @@ fn get_workspace(prebake_config: &PrebakeConfig) -> PathBuf {
 }
 
 fn privileged(prebake_config: &PrebakeConfig) -> bool {
-    if let Some(bootstrap) = &prebake_config.bootstrap {
-        if bootstrap.user == "root" {
-            return true;
-        }
+    if let Some(bootstrap) = &prebake_config.bootstrap
+        && bootstrap.user == "root"
+    {
+        return true;
     }
     let drop_after = get_drop_after(&prebake_config.security).unwrap_or_default();
     PrebakeStage::Never == drop_after

workshop-baker/src/bake/builder.rs

@@ -1,7 +1,7 @@
-use crate::bake::decorator::Decorator;
-use crate::bake::parser::Function;
 use crate::bake::constant::{PIPELINE_SKIP_ERRORCODE, TRIVIAL_SCRIPT_NAME};
+use crate::bake::decorator::Decorator;
 use crate::bake::error::BakeError;
+use crate::bake::parser::Function;
 use minijinja::{Environment, context};
 use std::fs::Permissions;
 use std::os::unix::fs::PermissionsExt;

workshop-baker/src/bake/error.rs

@@ -2,10 +2,7 @@ use thiserror::Error;
 
 use workshop_engine::{
     ExecutionError,
-    error::{
-        EXITCODE_IO_ERROR, EXITCODE_PARSE_ERROR,
-        HasExitCode,
-    },
+    error::{EXITCODE_IO_ERROR, EXITCODE_PARSE_ERROR, HasExitCode},
 };
 
 #[derive(Error, Debug)]

workshop-baker/src/finalize.rs

@@ -9,12 +9,12 @@ pub mod types;
 
 use crate::cli::Cli;
 use crate::finalize::constant::FINALIZE_STANDALONE_PLUGIN_PATH;
-use workshop_engine::EventSender;
 use crate::finalize::plugin::fetch::{self, FetchArgument};
-use std::path::PathBuf;
 pub use config::*;
 pub use error::FinalizeError;
 use std::path::Path;
+use std::path::PathBuf;
+use workshop_engine::EventSender;
 
 pub fn parse(path: &Path) -> Result<FinalizeConfig, FinalizeError> {
     let content = std::fs::read_to_string(path).map_err(FinalizeError::IoError)?;
@@ -24,7 +24,7 @@ pub fn parse(path: &Path) -> Result<FinalizeConfig, FinalizeError> {
 
 pub async fn finalize(
     finalize_path: &Path,
-    cli: &Cli,
+    _cli: &Cli,
     ctx: &mut workshop_engine::ExecutionContext,
     event_tx: EventSender,
 ) -> Result<(), FinalizeError> {
@@ -55,7 +55,7 @@ pub async fn finalize(
                 checksum: config.checksum.clone(),
                 shallow: config.shallow,
             };
-            fetch::fetch_plugin(&source, &name, &plugin_path, fetch_argument).await?;
+            fetch::fetch_plugin(source, name, &plugin_path, fetch_argument).await?;
             config.runtime.fspath = Some(plugin_path.join(name));
         }
     }
@@ -75,7 +75,7 @@ pub async fn finalize(
 
     // Determine if we should continue with artifacts/latehook
     // If prior stages failed, we still send notification but skip artifacts
-    let prior_failed = earlyhook_result.is_err();
+    let _prior_failed = earlyhook_result.is_err();
 
     // Prepare artifacts
     // TODO: artifact stage implementation

workshop-baker/src/finalize/config.rs

@@ -3,13 +3,13 @@
 //! This module defines the configuration schema for the finalize stage,
 //! which handles artifact collection, publishing, and notifications.
 
-use workshop_engine::{CustomCommand, deserialize_duration_ms};
 use crate::finalize::types::compression::CompressionMethod;
 use semver::{Version, VersionReq};
 use serde::{Deserialize, Serialize};
 use serde_yaml::Value;
 use std::collections::HashMap;
 use std::path::PathBuf;
+use workshop_engine::{CustomCommand, deserialize_duration_ms};
 
 /// Version requirement for finalize.yml schema compatibility
 pub const VERSION_REQUIREMENT: &str = "^0.0.1";
@@ -253,7 +253,6 @@ impl Default for ArtifactDef {
 
 /// Artifact
 /// TODO: Under heavy refactor
-
 /// Cleanup configuration
 #[derive(Debug, Deserialize, Serialize, Clone, Default)]
 pub struct CleanupConfig {

workshop-baker/src/finalize/plugin.rs

@@ -1,5 +1,5 @@
-use crate::finalize::constant::{FINALIZE_PLUGIN_MANIFEST, FINALIZE_PLUGIN_MANIFEST_EXTENSION};
 use crate::finalize::RawPluginMap;
+use crate::finalize::constant::{FINALIZE_PLUGIN_MANIFEST, FINALIZE_PLUGIN_MANIFEST_EXTENSION};
 use crate::finalize::plugin::metadata::{PluginMetadata, PluginType};
 use crate::{EventSender, ExecutionContext, finalize::error::PluginError};
 use async_trait::async_trait;
@@ -134,7 +134,7 @@ fn load_manifest(path: &Path, name: &str, suffix: &str) -> Result<String, Plugin
         reason: format!(
             "Failed to read plugin manifest from {}: {}",
             manifest_path.display(),
-            e.to_string()
+            e
         ),
     })
 }
@@ -189,10 +189,10 @@ fn get_entry(plugin_type: PluginType, name: &str) -> Result<Box<dyn AsyncPluginF
                 "Plugin {} has unknown plugin type and cannot be inferred from entrypoint.",
                 name
             );
-            return Err(PluginError::PluginUnavailable {
+            Err(PluginError::PluginUnavailable {
                 name: name.to_string(),
                 reason: "Unknown plugin type and cannot be inferred from entrypoint".to_string(),
-            });
+            })
         }
     }
 }

workshop-baker/src/finalize/plugin/fetch.rs

@@ -1,9 +1,6 @@
 // Code in this module PARTIALLY OR FULLY utilized AI Coding Agent.
 use crate::finalize::fetch;
-use crate::finalize::{
-    PluginConfig,
-    plugin::{PluginError, RawPluginMap},
-};
+use crate::finalize::plugin::{PluginError, RawPluginMap};
 use std::collections::HashSet;
 use std::path::Path;
 
@@ -66,14 +63,14 @@ pub async fn fetch_plugin(
             // TODO: Support client mode
             // NOTE: In standalone mode, we just assume path is well-defined and utilize
             log::info!("Fetching file plugin: {}", url);
-            return Ok(());
+            Ok(())
         }
         URLScheme::Unknown => {
             log::error!("Unknown URL scheme in plugin URL: {}", url);
-            return Err(PluginError::GeneralError(format!(
+            Err(PluginError::GeneralError(format!(
                 "Unknown URL scheme in plugin URL: {}",
                 url
-            )));
+            )))
         }
     }
 }
@@ -104,7 +101,7 @@ pub async fn fetch_plugins(
             checksum: config.checksum.clone(),
             shallow: config.shallow,
         };
-        fetch::fetch_plugin(&source, &name, &plugin_path, fetch_argument).await?;
+        fetch::fetch_plugin(source, name, plugin_path, fetch_argument).await?;
         log::debug!("Storing plugin {} to {}", name, plugin_path.display());
         config.runtime.fspath = Some(plugin_path.join(name));
     }

workshop-baker/src/finalize/plugin/fetch/extract.rs

@@ -1,6 +1,6 @@
 // Code in this module PARTIALLY OR FULLY utilized AI Coding Agent.
 use crate::finalize::plugin::PluginError;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use tokio::task;
 
 /// Compression type for tar archives.
@@ -48,7 +48,7 @@ impl CompressionType {
 /// Removes existing destination directory if present, creates a fresh directory,
 /// then unpacks the archive using the appropriate decompression based on `compression`.
 pub async fn extract_tar(
-    archive: &PathBuf,
+    archive: &Path,
     destdir: &PathBuf,
     compression: CompressionType,
 ) -> Result<(), PluginError> {
@@ -61,7 +61,7 @@ pub async fn extract_tar(
         .await
         .map_err(|e| PluginError::GeneralError(format!("Failed to create directory: {}", e)))?;
 
-    let archive = archive.clone();
+    let archive = archive.to_path_buf();
     let destdir = destdir.clone();
 
     task::spawn_blocking(move || {

workshop-baker/src/finalize/plugin/fetch/http.rs

@@ -69,7 +69,7 @@ fn extract_archive_extension(url: &str) -> &str {
     } else if lower.ends_with(".tar") {
         ".tar"
     } else {
-        ".tar"
+        ".unknown"
     }
 }
 
@@ -86,23 +86,21 @@ pub async fn fetch_http_plugin(
 
     download_to_file(url, &temp_file, name).await?;
 
-    if let Some(checksum) = checksum_str {
-        if !checksum.is_empty() {
-            let bytes = tokio::fs::read(&temp_file).await.map_err(|e| {
-                PluginError::GeneralError(format!("Failed to read downloaded file: {}", e))
-            })?;
+    if let Some(checksum) = checksum_str
+        && !checksum.is_empty()
+    {
+        let bytes = tokio::fs::read(&temp_file).await.map_err(|e| {
+            PluginError::GeneralError(format!("Failed to read downloaded file: {}", e))
+        })?;
 
-            let checksum_type =
-                detect_checksum_type(checksum).map_err(|e| PluginError::GeneralError(e))?;
+        let checksum_type = detect_checksum_type(checksum).map_err(PluginError::GeneralError)?;
 
-            if let Some(ct) = checksum_type {
-                verify_checksum(&bytes, checksum, ct).map_err(|e| PluginError::GeneralError(e))?;
-            }
+        if let Some(ct) = checksum_type {
+            verify_checksum(&bytes, checksum, ct).map_err(PluginError::GeneralError)?;
         }
     }
 
-    let compression =
-        CompressionType::from_path(&temp_file).map_err(|e| PluginError::GeneralError(e))?;
+    let compression = CompressionType::from_path(&temp_file).map_err(PluginError::GeneralError)?;
     let destdir = basedir.join(name);
     extract_tar(&temp_file, &destdir, compression).await?;
 

workshop-baker/src/finalize/plugin/internal.rs

@@ -10,22 +10,22 @@ pub struct InternalPlugin {
 }
 
 pub fn get_internal_plugin() -> Vec<InternalPlugin> {
-    let mut internal_plugins: Vec<InternalPlugin> = Vec::new();
-    internal_plugins.push(InternalPlugin {
-        name: "in-site-notify",
-        entry: Box::new(insitenotify::InsiteNotify),
-    });
-    internal_plugins.push(InternalPlugin {
-        name: "mail",
-        entry: Box::new(mail::Mail),
-    });
-    internal_plugins.push(InternalPlugin {
-        name: "satori",
-        entry: Box::new(satori::Satori),
-    });
-    internal_plugins.push(InternalPlugin {
-        name: "webhook",
-        entry: Box::new(webhook::Webhook),
-    });
-    internal_plugins
+    vec![
+        InternalPlugin {
+            name: "in-site-notify",
+            entry: Box::new(insitenotify::InsiteNotify),
+        },
+        InternalPlugin {
+            name: "mail",
+            entry: Box::new(mail::Mail),
+        },
+        InternalPlugin {
+            name: "satori",
+            entry: Box::new(satori::Satori),
+        },
+        InternalPlugin {
+            name: "webhook",
+            entry: Box::new(webhook::Webhook),
+        },
+    ]
 }

workshop-baker/src/finalize/plugin/internal/mail/config.rs

@@ -1,6 +1,6 @@
 // Code in this module PARTIALLY OR FULLY utilized AI Coding Agent.
-use workshop_engine::deserialize_duration_ms;
 use serde::Deserialize;
+use workshop_engine::deserialize_duration_ms;
 
 /// Email address representation supporting plain string or structured object.
 #[derive(Debug, Deserialize, Clone)]

workshop-baker/src/finalize/plugin/shell.rs

@@ -94,7 +94,7 @@ fn convert_argument(
             envvars.insert(prefix.to_string(), value.to_string());
         }
         serde_yaml::Value::String(value) => {
-            envvars.insert(prefix.to_string(), format!("{}", value));
+            envvars.insert(prefix.to_string(), value.to_string());
         }
         serde_yaml::Value::Number(value) => {
             envvars.insert(prefix.to_string(), value.to_string());
@@ -112,12 +112,12 @@ fn convert_argument(
                 return envvars;
             }
             let mut array = String::new();
-            array.push_str("(");
+            array.push('(');
             let mut has_mapping = false;
             let mut _envvars: HashMap<String, String> = HashMap::new();
             for (index, item) in seq.into_iter().enumerate() {
                 if index > 0 {
-                    array.push_str(" ");
+                    array.push(' ');
                 }
                 let new_prefix = format!("{}_{}", prefix, index);
                 let subenvvars = convert_argument(new_prefix.as_str(), item.clone(), true);
@@ -136,7 +136,7 @@ fn convert_argument(
             if has_mapping {
                 envvars.extend(_envvars);
             } else {
-                array.push_str(")");
+                array.push(')');
                 envvars.insert(prefix.to_string(), array);
             }
         }

workshop-baker/src/finalize/stage/hook.rs

@@ -1,7 +1,7 @@
 use crate::ExecutionContext;
-use workshop_engine::{DeltaExecutionContext, Engine, EventSender, ExecutionError, CustomCommand};
 use crate::finalize::plugin::{PluginMap, call_named_plugin};
 use std::path::PathBuf;
+use workshop_engine::{CustomCommand, DeltaExecutionContext, Engine, EventSender, ExecutionError};
 
 /// Executes post-build finalize hooks, supporting both plugin and shell command hooks.
 ///

workshop-baker/src/main.rs

@@ -7,7 +7,7 @@ use workshop_baker::{
     finalize::finalize,
     prebake::prebake,
 };
-use workshop_engine::{error::HasExitCode, EventSender, ExecutionContext};
+use workshop_engine::{EventSender, ExecutionContext, error::HasExitCode};
 
 #[tokio::main]
 async fn main() {

workshop-baker/src/notify.rs

@@ -1,10 +1,18 @@
-use crate::finalize::constant::FINALIZE_STANDALONE_PLUGIN_PATH;
 use crate::finalize::FinalizeConfig;
 use crate::finalize::NotificationConfig;
+use crate::finalize::NotificationTemplate;
+use crate::finalize::config::TriggerItemRaw;
+use crate::finalize::constant::FINALIZE_STANDALONE_PLUGIN_PATH;
 use crate::finalize::parse;
 use crate::finalize::plugin;
 use crate::finalize::plugin::fetch;
+use crate::notify::types::Matchable;
 use crate::notify::types::Method;
+use crate::notify::types::NOTIFICATION_DEFAULT_FALLIBLE;
+use crate::notify::types::NOTIFICATION_DEFAULT_MAX_LIVES;
+use crate::notify::types::NOTIFICATION_DEFAULT_RETRY_BACKOFF;
+use crate::notify::types::NOTIFICATION_DEFAULT_TIMEOUT_MS;
+use crate::notify::types::NOTIFICATION_NOTREADY_DELAY;
 use crate::notify::types::NOTIFICATION_TEMPORARY_DELAY_FACTOR;
 use crate::notify::types::NOTIFICATION_TEMPORARY_DELAY_TIME;
 use crate::notify::types::NOTIFICATION_TEMPORARY_MAX_TIME;
@@ -15,20 +23,14 @@ use crate::notify::types::NotificationPluginMap;
 use crate::notify::types::NotificationRenderer;
 use crate::notify::types::NotificationRule;
 use crate::notify::types::NotificationRulesMap;
+use crate::notify::types::ParsedTrigger;
 use crate::{EventSender, ExecutionContext};
+use chrono::Utc;
+use globset::{Glob, GlobSetBuilder};
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::path::Path;
 use std::path::PathBuf;
-use crate::notify::types::ParsedTrigger;
-use crate::notify::types::Matchable;
-use globset::{Glob, GlobSetBuilder};
-use crate::finalize::config::TriggerItemRaw;
-use crate::finalize::NotificationTemplate;
-use crate::notify::types::NOTIFICATION_DEFAULT_FALLIBLE;
-use crate::notify::types::NOTIFICATION_DEFAULT_MAX_LIVES;
-use crate::notify::types::NOTIFICATION_DEFAULT_TIMEOUT_MS;
-use crate::notify::types::NOTIFICATION_DEFAULT_RETRY_BACKOFF;
 
 mod error;
 pub mod types;
@@ -49,8 +51,8 @@ fn extract_plugin(finalize: &FinalizeConfig) -> HashSet<String> {
     finalize
         .notification
         .groups
-        .iter()
-        .flat_map(|(_, methods)| methods.keys())
+        .values()
+        .flat_map(|methods| methods.keys())
         .cloned()
         .collect()
 }
@@ -58,7 +60,7 @@ fn extract_plugin(finalize: &FinalizeConfig) -> HashSet<String> {
 pub async fn notify(
     finalize_path: &Path,
     ctx: &mut ExecutionContext,
-    event_tx: EventSender,
+    _event_tx: EventSender,
     notifyevent_rx: &mut NotificationEventReceiver,
     notifyevent_tx: NotificationEventSender,
 ) -> anyhow::Result<()> {
@@ -84,48 +86,65 @@ pub async fn notify(
 
     let notification_rules = parse_rules(&finalize.notification);
 
-    // loop {
-    //     tokio::select! {
-    //         Some(events) = notifyevent_rx.recv() => {
-    //             match notify_handler(&notify_plugins, &finalize.notification, &events){
-    //                 Ok(_) => {}, // Notification succeeded / failed with fallible
-    //                 Err(e) => {
-    //                     log::debug!("Failed to send notification: {:?}", e);
-    //                     match e {
-    //                         NotifyError::PluginNotAvailable(_) => {
-    //                             // Should keep lives, add static delay and give less priority
-    //                             for mut event in events {
-    //                                 event.not_before = Utc::now() + NOTIFY_NOTREADY_DELAY;
-    //                                 event.effective_priority = event.effective_priority.saturating_add(1);
-    //                                 notifyevent_tx.send(event).await?;
-    //                             }
-    //                         },
-    //                         NotifyError::TemporaryError(_) => {
-    //                             // Should cost 1 life, add dynamic (exponential) delay
-    //                             for mut event in events {
-    //                                 event.lives -= 1;
-    //                                 event.not_before = Utc::now() + notify_delay(&event);
-    //                                 notifyevent_tx.send(event).await?;
-    //                             }
-    //                         },
-    //                         NotifyError::PermanentError(_) => {
-    //                             // Unexpected!
-    //                             // Should be dropped
-    //                             log::error!("None of the configured notification plugins works.");
-    //                             log::error!("Message will be dropped.");
-    //                             // TODO: Send audit/metric information
-    //                         }
-    //                         error => {
-    //                             // Unexpected!
-    //                             log::error!("Internal error: notify_handler returned unexpected error: {:?}", error);
-    //                         }
-    //                     };
-    //                 }
-    //             };
-    //         }
-    //     }
-    // }
-    todo!();
+    loop {
+        tokio::select! {
+            Some(event) = notifyevent_rx.recv() => {
+                handle_notification_event(
+                    event,
+                    &notify_plugins,
+                    &notification_rules,
+                    &notifyevent_tx,
+                )
+                .await?;
+            }
+        }
+    }
+}
+
+/// Process a single notification event.
+///
+/// Delegates to [`notification_handler`] and handles errors:
+/// - [`NotifyError::PluginNotAvailable`] — reschedule with static delay, lower priority
+/// - [`NotifyError::TemporaryError`] — consume one life, reschedule with exponential backoff
+/// - [`NotifyError::PermanentError`] — drop the event
+/// - other — log and drop
+async fn handle_notification_event(
+    event: NotificationEvent,
+    plugins: &NotificationPluginMap,
+    rules: &NotificationRulesMap,
+    notifyevent_tx: &NotificationEventSender,
+) -> anyhow::Result<()> {
+    match notification_handler(plugins, rules, &event) {
+        Ok(_) => {}
+        Err(e) => {
+            log::debug!("Failed to send notification: {:?}", e);
+            match e {
+                NotifyError::PluginNotAvailable(_) => {
+                    let mut event = event;
+                    event.not_before = Utc::now() + NOTIFICATION_NOTREADY_DELAY;
+                    event.effective_priority = event.effective_priority.saturating_add(1);
+                    notifyevent_tx.send(event).await?;
+                }
+                NotifyError::TemporaryError(_) => {
+                    let mut event = event;
+                    event.lives -= 1;
+                    event.not_before = Utc::now() + notify_delay(&event);
+                    notifyevent_tx.send(event).await?;
+                }
+                NotifyError::PermanentError(_) => {
+                    log::error!("None of the configured notification plugins works.");
+                    log::error!("Message will be dropped.");
+                }
+                error => {
+                    log::error!(
+                        "Internal error: notify_handler returned unexpected error: {:?}",
+                        error
+                    );
+                }
+            }
+        }
+    }
+    Ok(())
 }
 
 fn notification_handler(
@@ -133,7 +152,7 @@ fn notification_handler(
     config: &NotificationRulesMap,
     event: &NotificationEvent,
 ) -> Result<(), NotifyError> {
-    for (priority, ruleset) in config.iter() {
+    for (_priority, ruleset) in config.iter() {
         for (method, rules) in ruleset.iter() {
             if !event.target.contains(method) {
                 log::trace!("target does not contain method name: {}", method);
@@ -173,23 +192,21 @@ fn notification_handler(
 
 fn parse_trigger_item_raw(item: &TriggerItemRaw) -> Vec<ParsedTrigger> {
     match item {
-        TriggerItemRaw::Simple(s) => {
-            match s.as_str() {
-                "on_success" => vec![ParsedTrigger::OnSuccess],
-                "on_failure" => vec![ParsedTrigger::OnFailure],
-                s if s.starts_with("on_finish_of:") => {
-                    let pattern = s.trim_start_matches("on_finish_of:").trim();
-                    let mut builder = GlobSetBuilder::new();
-                    builder.add(Glob::new(pattern).expect("Invalid glob pattern"));
-                    let glob_set = builder.build().expect("Failed to compile glob set");
-                    vec![ParsedTrigger::OnFinishOf(Matchable::Glob(glob_set))]
-                }
-                _ => {
-                    log::warn!("Unknown trigger string: {}", s);
-                    vec![]
-                }
+        TriggerItemRaw::Simple(s) => match s.as_str() {
+            "on_success" => vec![ParsedTrigger::OnSuccess],
+            "on_failure" => vec![ParsedTrigger::OnFailure],
+            s if s.starts_with("on_finish_of:") => {
+                let pattern = s.trim_start_matches("on_finish_of:").trim();
+                let mut builder = GlobSetBuilder::new();
+                builder.add(Glob::new(pattern).expect("Invalid glob pattern"));
+                let glob_set = builder.build().expect("Failed to compile glob set");
+                vec![ParsedTrigger::OnFinishOf(Matchable::Glob(glob_set))]
             }
-        }
+            _ => {
+                log::warn!("Unknown trigger string: {}", s);
+                vec![]
+            }
+        },
         TriggerItemRaw::Map(map) => {
             let mut triggers = Vec::new();
             for (key, patterns) in map {
@@ -219,22 +236,21 @@ fn merge_templates(
     policy_template: &Option<serde_yaml::Value>,
 ) -> NotificationTemplate {
     let mut merged = method_template.clone();
-    if let Some(policy_tmpl) = policy_template {
-        if let Ok(policy_template) =
+    if let Some(policy_tmpl) = policy_template
+        && let Ok(policy_template) =
             serde_yaml::from_value::<NotificationTemplate>(policy_tmpl.clone())
-        {
-            if policy_template.schema.is_some() {
-                merged.schema = policy_template.schema;
-            }
-            if policy_template.on_success.is_some() {
-                merged.on_success = policy_template.on_success;
-            }
-            if policy_template.on_failure.is_some() {
-                merged.on_failure = policy_template.on_failure;
-            }
-            if policy_template.on_finish_of.is_some() {
-                merged.on_finish_of = policy_template.on_finish_of;
-            }
+    {
+        if policy_template.schema.is_some() {
+            merged.schema = policy_template.schema;
+        }
+        if policy_template.on_success.is_some() {
+            merged.on_success = policy_template.on_success;
+        }
+        if policy_template.on_failure.is_some() {
+            merged.on_failure = policy_template.on_failure;
+        }
+        if policy_template.on_finish_of.is_some() {
+            merged.on_finish_of = policy_template.on_finish_of;
         }
     }
     merged
@@ -306,7 +322,8 @@ fn parse_rules(config: &NotificationConfig) -> NotificationRulesMap {
                             .flat_map(parse_trigger_item_raw)
                             .collect()
                     };
-                    let template = merge_templates(&method.template, &Some(policy.template.clone()));
+                    let template =
+                        merge_templates(&method.template, &Some(policy.template.clone()));
                     let method_overrides = extract_overrides(&method.config);
                     let policy_overrides = extract_overrides(&policy.overrides);
                     let fallible = if policy.overrides.get("fallible").is_some() {

workshop-baker/src/notify/types.rs

@@ -14,7 +14,7 @@ use std::time::Duration;
 use uuid::Uuid;
 
 use crate::finalize::NotificationTemplate;
-use crate::finalize::config::Trigger;
+
 /// Final runtime configuration for notification, restored and flattened from yaml
 pub struct NotificationRule {
     pub enabled: bool,

workshop-baker/src/prebake.rs

@@ -34,9 +34,9 @@ use crate::{
     cli::Cli,
     prebake::{error::PrebakeError, security::get_drop_after, stage::PrebakeStage},
 };
-use workshop_engine::EventSender;
 use chrono::Utc;
 use std::path::Path;
+use workshop_engine::EventSender;
 
 pub use config::PrebakeConfig;
 pub use security::prebake_drop_privilege;
@@ -51,7 +51,6 @@ pub use security::prebake_drop_privilege;
 ///
 /// Returns `Ok(PrebakeConfig)` on successful parsing, or `Err(serde_yaml::Error)`
 /// if the YAML is invalid.
-
 pub fn parse(yaml_content: &str) -> Result<PrebakeConfig, serde_yaml::Error> {
     serde_yaml::from_str(yaml_content)
 }
@@ -92,7 +91,7 @@ pub async fn prebake(
     prebake_path: &Path,
     cli: &Cli,
     stage: Option<PrebakeStage>,
-    mut ctx: &mut ExecutionContext,
+    ctx: &mut ExecutionContext,
     event_tx: EventSender,
 ) -> anyhow::Result<()> {
     // Initialize
@@ -125,26 +124,26 @@ pub async fn prebake(
 
     // Bootstrap stage
     if stage <= PrebakeStage::Bootstrap {
-        let stage_start = Utc::now();
+        let _stage_start = Utc::now();
         log::info!("Running PBStage: Bootstrap");
-        prebake_drop_privilege(PrebakeStage::Bootstrap, dropafter, target_user, &mut ctx)?;
+        prebake_drop_privilege(PrebakeStage::Bootstrap, dropafter, target_user, ctx)?;
         ctx.task_id = format!("{}-{}-bootstrap", cli.pipeline, cli.build_id);
         let osinfo = os_info::get();
-        let result = stage::bootstrap::bootstrap(&prebake, &osinfo, &ctx, &event_tx).await;
+        let result = stage::bootstrap::bootstrap(&prebake, &osinfo, ctx, &event_tx).await;
         result?;
     }
 
     // EarlyHook stage
     if stage <= PrebakeStage::EarlyHook {
-        let stage_start = Utc::now();
+        let _stage_start = Utc::now();
         log::info!("Running PBStage: EarlyHook");
-        prebake_drop_privilege(PrebakeStage::EarlyHook, dropafter, target_user, &mut ctx)?;
+        prebake_drop_privilege(PrebakeStage::EarlyHook, dropafter, target_user, ctx)?;
         ctx.task_id = format!("{}-{}-earlyhook", cli.pipeline, cli.build_id);
         let earlyhook = match &prebake.hooks {
             Some(hooks) => hooks.early.clone(),
             None => None,
         };
-        let result = stage::hook::hook(earlyhook, &ctx, event_tx.clone()).await;
+        let result = stage::hook::hook(earlyhook, ctx, event_tx.clone()).await;
         result?;
     }
 
@@ -160,7 +159,7 @@ pub async fn prebake(
 
         let upm_sys_deps = if let Some(user) = &dependencies.user {
             if stage <= PrebakeStage::DepsSystem {
-                stage::depsuser::collect_upm_sysdeps(user, &ctx, &endpoint).await?
+                stage::depsuser::collect_upm_sysdeps(user, ctx, &endpoint).await?
             } else {
                 stage::depsuser::CollectedUPMdeps::default()
             }
@@ -208,38 +207,37 @@ pub async fn prebake(
         }
 
         if !system_config.is_empty() && stage <= PrebakeStage::DepsSystem {
-            let stage_start = Utc::now();
+            let _stage_start = Utc::now();
             log::info!("Running PBStage: DepsSystem");
-            prebake_drop_privilege(PrebakeStage::DepsSystem, dropafter, target_user, &mut ctx)?;
+            prebake_drop_privilege(PrebakeStage::DepsSystem, dropafter, target_user, ctx)?;
             ctx.task_id = format!("{}-{}-depssystem", cli.pipeline, cli.build_id);
-            let result =
-                stage::depssystem::depssystem(&system_config, &ctx, event_tx.clone()).await;
+            let result = stage::depssystem::depssystem(&system_config, ctx, event_tx.clone()).await;
             result?;
         }
 
         if let Some(user) = dependencies.user
             && stage <= PrebakeStage::DepsUser
         {
-            let stage_start = Utc::now();
+            let _stage_start = Utc::now();
             log::info!("Running PBStage: DepsUser");
-            prebake_drop_privilege(PrebakeStage::DepsUser, dropafter, target_user, &mut ctx)?;
+            prebake_drop_privilege(PrebakeStage::DepsUser, dropafter, target_user, ctx)?;
             ctx.task_id = format!("{}-{}-depsuser", cli.pipeline, cli.build_id);
-            let result = stage::depsuser::depsuser(&user, &ctx, event_tx.clone(), &endpoint).await;
+            let result = stage::depsuser::depsuser(&user, ctx, event_tx.clone(), &endpoint).await;
             result?;
         }
     }
 
     // LateHook stage
     if stage <= PrebakeStage::LateHook {
-        let stage_start = Utc::now();
+        let _stage_start = Utc::now();
         log::info!("Running PBStage: LateHook");
-        prebake_drop_privilege(PrebakeStage::LateHook, dropafter, target_user, &mut ctx)?;
+        prebake_drop_privilege(PrebakeStage::LateHook, dropafter, target_user, ctx)?;
         ctx.task_id = format!("{}-{}-latehook", cli.pipeline, cli.build_id);
         let latehook = match &prebake.hooks {
             Some(hooks) => hooks.late.clone(),
             None => None,
         };
-        let result = stage::hook::hook(latehook, &ctx, event_tx.clone()).await;
+        let result = stage::hook::hook(latehook, ctx, event_tx.clone()).await;
         result?;
     }
 

workshop-baker/src/prebake/error.rs

@@ -3,10 +3,7 @@ use thiserror::Error;
 
 use workshop_engine::{
     ExecutionError,
-    error::{
-        EXITCODE_IO_ERROR, EXITCODE_PARSE_ERROR, EXITCODE_PRIV_DROP_FAILED,
-        HasExitCode,
-    },
+    error::{EXITCODE_IO_ERROR, EXITCODE_PARSE_ERROR, EXITCODE_PRIV_DROP_FAILED, HasExitCode},
 };
 
 #[derive(Error, Debug)]

workshop-baker/src/prebake/security.rs

@@ -1,6 +1,6 @@
 use crate::ExecutionContext;
-use crate::prebake::error::PrebakeError;
 use crate::prebake::config::Security;
+use crate::prebake::error::PrebakeError;
 use crate::prebake::stage::PrebakeStage;
 use privdrop::PrivDrop;
 use std::ffi::CString;

workshop-baker/src/prebake/stage/bootstrap.rs

@@ -9,15 +9,15 @@
 //! and the target operating system, supporting multiple package managers.
 
 use crate::ExecutionContext;
-use crate::prebake::constant::BOOTSTRAP_SCRIPT_PATH;
-use workshop_engine::{Engine, EventSender, pm};
-use crate::prebake::error::BootstrapError;
 use crate::prebake::config::Bootstrap;
 use crate::prebake::config::PrebakeConfig;
+use crate::prebake::constant::BOOTSTRAP_SCRIPT_PATH;
+use crate::prebake::error::BootstrapError;
 use std::fs::File;
 use std::io::Write;
 use std::path::{Path, PathBuf};
 use which::which;
+use workshop_engine::{Engine, EventSender, pm};
 
 /// Executes the bootstrap stage of the prebake workflow.
 ///
@@ -96,7 +96,7 @@ fn write_bootstrap_content(bootstrap_path: &Path, content: String) -> Result<(),
     let mut file = File::create(bootstrap_path)?;
     file.write_all(content.as_bytes())?;
     std::fs::set_permissions(
-        &bootstrap_path,
+        bootstrap_path,
         std::os::unix::fs::PermissionsExt::from_mode(0o755),
     )?;
     Ok(())

workshop-baker/src/prebake/stage/depssystem.rs

@@ -1,7 +1,7 @@
 use crate::ExecutionContext;
-use workshop_engine::{Engine, EventSender, pm, error::DependencyError};
 use crate::prebake::config::SystemDependency;
 use std::collections::HashMap;
+use workshop_engine::{Engine, EventSender, error::DependencyError, pm};
 
 /// Handles system dependency installation for the target platform.
 ///

workshop-baker/src/prebake/stage/depsuser.rs

@@ -1,8 +1,8 @@
 use serde_yaml::Value;
 
 use crate::ExecutionContext;
-use workshop_engine::{EventSender, upm, error::DependencyError, RepologyEndpoint};
 use std::collections::HashMap;
+use workshop_engine::{EventSender, RepologyEndpoint, error::DependencyError, upm};
 
 #[derive(Default)]
 pub struct CollectedUPMdeps {

workshop-baker/src/prebake/stage/hook.rs

@@ -13,8 +13,8 @@
 //! and timeout configurations.
 
 use crate::ExecutionContext;
-use workshop_engine::{DeltaExecutionContext, Engine, EventSender, ExecutionError, CustomCommand};
 use std::path::PathBuf;
+use workshop_engine::{CustomCommand, DeltaExecutionContext, Engine, EventSender, ExecutionError};
 
 /// Executes a sequence of hook commands.
 ///
@@ -24,7 +24,7 @@ use std::path::PathBuf;
 /// # Arguments
 ///
 /// * `hook` - Optional vector of hook commands to execute. If `None` or empty,
-///            the function returns immediately with `Ok(())`.
+///   the function returns immediately with `Ok(())`.
 /// * `ctx` - Base execution context containing pipeline parameters and settings
 /// * `event_tx` - Event sender for reporting hook execution progress
 ///

workshop-baker/src/types.rs

@@ -1 +1 @@
-pub mod buildstatus;
+pub mod buildstatus;

workshop-baker/tests/finalize_hook_integration.rs

@@ -5,10 +5,10 @@
 
 use std::collections::HashMap;
 use std::time::Duration;
-use workshop_baker::engine::{ExecutionContext, ExecutionEvent};
+use workshop_baker::ExecutionContext;
 use workshop_baker::finalize::plugin::PluginMap;
 use workshop_baker::finalize::stage::hook::hook;
-use workshop_baker::types::command::CustomCommand;
+use workshop_engine::{CustomCommand, ExecutionEvent};
 
 fn create_test_context() -> ExecutionContext {
     ExecutionContext {
@@ -18,7 +18,6 @@ fn create_test_context() -> ExecutionContext {
         working_dir: std::env::temp_dir(),
         env_vars: HashMap::new(),
         timeout: Some(Duration::from_secs(30)),
-        cgroup_path: None,
         shell: "bash".to_string(),
         dry_run: false,
         privileged: false,

workshop-baker/tests/finalize_pipeline_integration.rs

@@ -5,9 +5,10 @@
 
 use std::collections::HashMap;
 use std::time::Duration;
+use workshop_baker::ExecutionContext;
 use workshop_baker::cli::Cli;
-use workshop_baker::engine::{ExecutionContext, ExecutionEvent};
 use workshop_baker::finalize::finalize;
+use workshop_engine::ExecutionEvent;
 
 fn create_test_context() -> ExecutionContext {
     ExecutionContext {
@@ -17,7 +18,6 @@ fn create_test_context() -> ExecutionContext {
         working_dir: std::env::temp_dir(),
         env_vars: HashMap::new(),
         timeout: Some(Duration::from_secs(30)),
-        cgroup_path: None,
         shell: "bash".to_string(),
         dry_run: false,
         privileged: false,

workshop-baker/tests/prebake_config_integration.rs

@@ -1,5 +1,5 @@
 use workshop_baker::prebake::parse;
-use workshop_baker::types::builderconfig::BuilderType;
+use workshop_baker::prebake::types::builderconfig::BuilderType;
 
 #[test]
 fn test_prebake_config_minimal_yaml() {

workshop-baker/tests/types_config_integration.rs

@@ -1,9 +1,8 @@
-use workshop_baker::types::architecture::Architecture;
-use workshop_baker::types::builderconfig::BuilderConfig;
-use workshop_baker::types::cache::{CacheDirectory, CacheMode, CacheStrategyType};
-use workshop_baker::types::command::CustomCommand;
-use workshop_baker::types::compression::CompressionMethod;
-use workshop_baker::types::repology::RepologyEndpoint;
+use workshop_baker::finalize::types::compression::CompressionMethod;
+use workshop_baker::prebake::types::architecture::Architecture;
+use workshop_baker::prebake::types::builderconfig::BuilderConfig;
+use workshop_baker::prebake::types::cache::{CacheDirectory, CacheMode, CacheStrategyType};
+use workshop_engine::{CustomCommand, RepologyEndpoint};
 
 #[test]
 fn test_architecture_from_str() {

workshop-engine/Cargo.lock

@@ -869,6 +869,7 @@ dependencies = [
  "chrono",
  "duration-str",
  "lazy_static",
+ "libc",
  "log",
  "os_info",
  "serde",

workshop-engine/Cargo.toml

@@ -19,3 +19,7 @@ lazy_static = "1.5.0"
 log = "0.4.28"
 os_info = "3.14.0"
 duration-str = "0.21.0"
+libc = "0.2"
+
+[dev-dependencies]
+tokio = { version = "1.48.0", features = ["full"] }

workshop-engine/src/child.rs

@@ -0,0 +1,211 @@
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::AsyncRead;
+
+/// A managed child process, abstracted over the real OS process for testability.
+///
+/// This trait represents a spawned process that we can:
+/// - Read its stdout/stderr
+/// - Wait for it to finish (getting its exit code)
+/// - Kill it
+///
+/// The real implementation wraps `tokio::process::Child`.
+/// Mock implementations are used in unit tests.
+pub trait ManagedChild: Send {
+    /// Returns the OS-assigned process ID, if available.
+    fn pid(&self) -> Option<u32>;
+
+    /// Takes the stdout pipe, if available.
+    fn take_stdout(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>>;
+
+    /// Takes the stderr pipe, if available.
+    fn take_stderr(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>>;
+
+    /// Waits for the process to exit and returns its exit code.
+    fn wait(&mut self) -> impl std::future::Future<Output = io::Result<i32>> + Send;
+
+    /// Kills the process.
+    fn kill(&mut self) -> impl std::future::Future<Output = io::Result<()>> + Send;
+}
+
+// ---------------------------------------------------------------------------
+// Real implementation: wraps tokio::process::Child
+// ---------------------------------------------------------------------------
+
+pub struct TokioChild(pub tokio::process::Child);
+
+impl ManagedChild for TokioChild {
+    fn pid(&self) -> Option<u32> {
+        self.0.id()
+    }
+
+    fn take_stdout(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>> {
+        self.0
+            .stdout
+            .take()
+            .map(|s| Box::new(s) as Box<dyn AsyncRead + Unpin + Send>)
+    }
+
+    fn take_stderr(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>> {
+        self.0
+            .stderr
+            .take()
+            .map(|s| Box::new(s) as Box<dyn AsyncRead + Unpin + Send>)
+    }
+
+    async fn wait(&mut self) -> io::Result<i32> {
+        self.0.wait().await.map(|s| s.code().unwrap_or(-1))
+    }
+
+    async fn kill(&mut self) -> io::Result<()> {
+        // Kill the entire process group. When combined with process_group(0)
+        // in executor.rs, the spawned command is the process group leader,
+        // so PGID = PID. Sending SIGKILL to -PGID terminates the whole tree.
+        #[cfg(unix)]
+        if let Some(pid) = self.0.id() {
+            let pgid = pid as i32;
+            let ret = unsafe { libc::kill(-pgid, libc::SIGKILL) };
+            if ret != 0 {
+                let err = io::Error::last_os_error();
+                // ESRCH = process group already gone, which is fine
+                if err.raw_os_error() != Some(libc::ESRCH) {
+                    log::warn!("kill(-{}, SIGKILL) failed: {}", pgid, err);
+                }
+            }
+        }
+
+        // Fallback: also try tokio's per-process kill
+        let _ = self.0.kill().await;
+        Ok(())
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Mock implementation: for testing Engine orchestration logic
+// ---------------------------------------------------------------------------
+
+/// A mock child that returns pre-configured results without spawning a real process.
+///
+/// Use `MockChildHandle` for advanced scenarios (timeout simulation, etc.).
+pub struct MockChild {
+    /// The exit code that `wait()` returns.
+    pub exit_code: i32,
+    /// Whether `kill()` was called.
+    pub killed: bool,
+    /// Simulated stdout content.
+    pub stdout_data: Vec<u8>,
+    /// Simulated stderr content.
+    pub stderr_data: Vec<u8>,
+    /// Simulated PID.
+    pub pid: u32,
+    /// When true, `wait()` never returns (simulates a hanging process for timeout testing).
+    pub hang_forever: bool,
+}
+
+impl Default for MockChild {
+    fn default() -> Self {
+        Self {
+            exit_code: 0,
+            killed: false,
+            stdout_data: Vec::new(),
+            stderr_data: Vec::new(),
+            pid: 12345,
+            hang_forever: false,
+        }
+    }
+}
+
+impl MockChild {
+    pub fn with_exit_code(code: i32) -> Self {
+        Self {
+            exit_code: code,
+            ..Default::default()
+        }
+    }
+
+    pub fn with_output(stdout: &str, stderr: &str) -> Self {
+        Self {
+            stdout_data: stdout.as_bytes().to_vec(),
+            stderr_data: stderr.as_bytes().to_vec(),
+            ..Default::default()
+        }
+    }
+
+    pub fn hanging() -> Self {
+        Self {
+            hang_forever: true,
+            ..Default::default()
+        }
+    }
+}
+
+impl ManagedChild for MockChild {
+    fn pid(&self) -> Option<u32> {
+        Some(self.pid)
+    }
+
+    fn take_stdout(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>> {
+        if self.stdout_data.is_empty() {
+            None
+        } else {
+            let data = std::mem::take(&mut self.stdout_data);
+            Some(Box::new(VecReader::new(data)))
+        }
+    }
+
+    fn take_stderr(&mut self) -> Option<Box<dyn AsyncRead + Unpin + Send>> {
+        if self.stderr_data.is_empty() {
+            None
+        } else {
+            let data = std::mem::take(&mut self.stderr_data);
+            Some(Box::new(VecReader::new(data)))
+        }
+    }
+
+    async fn wait(&mut self) -> io::Result<i32> {
+        if self.hang_forever {
+            // Park this future indefinitely to simulate a timeout
+            std::future::pending::<()>().await;
+            unreachable!()
+        }
+        Ok(self.exit_code)
+    }
+
+    async fn kill(&mut self) -> io::Result<()> {
+        self.killed = true;
+        Ok(())
+    }
+}
+
+// ---------------------------------------------------------------------------
+// VecReader: wraps Vec<u8> into tokio::io::AsyncRead
+// ---------------------------------------------------------------------------
+
+pub struct VecReader {
+    data: Vec<u8>,
+    pos: usize,
+}
+
+impl VecReader {
+    pub fn new(data: Vec<u8>) -> Self {
+        Self { data, pos: 0 }
+    }
+}
+
+impl AsyncRead for VecReader {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        if self.pos >= self.data.len() {
+            return Poll::Ready(Ok(()));
+        }
+        let available = &self.data[self.pos..];
+        let len = available.len().min(buf.remaining());
+        buf.put_slice(&available[..len]);
+        self.pos += len;
+        Poll::Ready(Ok(()))
+    }
+}

workshop-engine/src/executor.rs

@@ -6,53 +6,16 @@ use tokio::process::Command;
 use tokio::sync::mpsc;
 use tokio::time::timeout;
 
+use crate::child::{ManagedChild, TokioChild};
 use crate::pm::PackageManager;
 use crate::types::{
-    CgroupManager, DeltaExecutionContext, EventSender, ExecutionContext, ExecutionError,
-    ExecutionEvent, ExecutionResult, ResourceLimits, StreamType,
+    DeltaExecutionContext, EventSender, ExecutionContext, ExecutionError, ExecutionEvent,
+    ExecutionResult, StreamType,
 };
 
 impl crate::types::Engine {
     pub fn new() -> Self {
-        Self {
-            cgroup_manager: None,
-        }
-    }
-
-    pub fn with_cgroup(limits: ResourceLimits, build_id: &str) -> Result<Self, String> {
-        let cgroup_manager =
-            CgroupManager::new(build_id).map_err(|e| format!("Failed to create cgroup: {}", e))?;
-
-        if let Some(mem) = limits.memory_bytes {
-            cgroup_manager
-                .set_memory_limit(mem)
-                .map_err(|e| format!("Failed to set memory limit: {}", e))?;
-        }
-
-        if let Some(weight) = limits.cpu_weight {
-            cgroup_manager
-                .set_cpu_weight(weight)
-                .map_err(|e| format!("Failed to set cpu weight: {}", e))?;
-        }
-
-        if let Some(max) = limits.cpu_max_us {
-            cgroup_manager
-                .set_cpu_max(max, limits.cpu_period_us)
-                .map_err(|e| format!("Failed to set cpu max: {}", e))?;
-        }
-
-        Ok(Self {
-            cgroup_manager: Some(cgroup_manager),
-        })
-    }
-
-    fn apply_cgroup(&self, pid: u32) -> Result<(), ExecutionError> {
-        if let Some(ref cgroup) = self.cgroup_manager {
-            cgroup.add_process(pid).map_err(|e| {
-                ExecutionError::IoError(format!("Failed to add process to cgroup: {}", e))
-            })?;
-        }
-        Ok(())
+        Self
     }
 
     pub async fn execute_script(
@@ -129,11 +92,51 @@ impl crate::types::Engine {
             .stderr(Stdio::piped());
     }
 
+    /// Execute a pre-configured command, managing the full lifecycle.
+    ///
+    /// This method:
+    /// 1. Configures the command with context (env vars, working dir, stdio)
+    /// 2. Creates a new process group on Unix for process-tree isolation
+    /// 3. Returns early in dry_run mode
+    /// 4. Spawns the OS process
+    /// 5. Delegates to `execute_child` for wait/event/output management
     pub async fn execute(
         &self,
         command: Command,
         ctx: &ExecutionContext,
         event_tx: &EventSender,
+    ) -> Result<ExecutionResult, ExecutionError> {
+        let mut command = command;
+        self.construct_command(&mut command, ctx);
+
+        if ctx.dry_run {
+            log::info!("[DRY_RUN] {:?}", command);
+            return Ok(ExecutionResult {
+                ..Default::default()
+            });
+        }
+
+        // Create a new process group so that kill() can terminate the
+        // entire process tree (bash + its children) on timeout.
+        #[cfg(unix)]
+        command.process_group(0);
+
+        let child = command.spawn().map_err(|e| {
+            ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e))
+        })?;
+
+        self.execute_child(TokioChild(child), ctx, event_tx).await
+    }
+
+    /// Core execution logic, abstracted over [`ManagedChild`] for testability.
+    ///
+    /// Handles: timeout, event emission, stdout/stderr capture, result construction.
+    /// Does NOT spawn processes itself — that is done by [`execute`].
+    pub async fn execute_child(
+        &self,
+        mut child: impl ManagedChild,
+        ctx: &ExecutionContext,
+        event_tx: &EventSender,
     ) -> Result<ExecutionResult, ExecutionError> {
         let start_time = std::time::Instant::now();
 
@@ -145,25 +148,6 @@ impl crate::types::Engine {
                 })
                 .await;
         }
-        let mut command = command;
-
-        self.construct_command(&mut command, ctx);
-
-        if ctx.dry_run {
-            log::info!("[DRY_RUN] {:?}", command);
-            return Ok(ExecutionResult {
-                ..Default::default()
-            });
-        }
-
-        let mut child = command.spawn().map_err(|e| {
-            ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e))
-        })?;
-
-        let pid = child
-            .id()
-            .ok_or_else(|| ExecutionError::IoError("Failed to get PID".to_string()))?;
-        self.apply_cgroup(pid)?;
 
         let task_id = ctx.task_id.clone();
         let (stdout_tx, mut stdout_rx) = mpsc::channel::<String>(1024);
@@ -171,11 +155,10 @@ impl crate::types::Engine {
         let mut stdout_buf = String::new();
         let mut stderr_buf = String::new();
 
-        let mut child_stdout = child.stdout.take();
-        let mut child_stderr = child.stderr.take();
+        let mut child_stdout = child.take_stdout();
+        let mut child_stderr = child.take_stderr();
 
         let tx_clone = stdout_tx.clone();
-        let _task_id_clone = task_id.clone();
         let stdout_handle = tokio::spawn(async move {
             if let Some(mut stdout) = child_stdout.take() {
                 let mut reader = tokio::io::BufReader::new(&mut stdout);
@@ -215,9 +198,9 @@ impl crate::types::Engine {
             }
         });
 
-        let status = if let Some(timeout_duration) = ctx.timeout {
+        let exit_code = if let Some(timeout_duration) = ctx.timeout {
             match timeout(timeout_duration, child.wait()).await {
-                Ok(Ok(status)) => status,
+                Ok(Ok(code)) => code,
                 Ok(Err(e)) => {
                     return Err(ExecutionError::IoError(format!(
                         "Failed to wait for process: {}",
@@ -289,7 +272,6 @@ impl crate::types::Engine {
             }
         }
 
-        let exit_code = status.code().unwrap_or(-1);
         let duration = start_time.elapsed();
 
         let result = ExecutionResult {
@@ -299,7 +281,6 @@ impl crate::types::Engine {
             duration,
             stdout: stdout_buf,
             stderr: stderr_buf,
-            resource_usage: None,
         };
 
         if let Some(tx) = event_tx {
@@ -342,13 +323,17 @@ impl crate::types::Engine {
             duration: std::time::Duration::default(),
             stdout: String::new(),
             stderr: String::new(),
-            resource_usage: None,
         };
-        for cmd in cmds {
+        for (i, cmd) in cmds.into_iter().enumerate() {
             log::debug!("Installing dependency with command: {:?}", cmd);
             let result = self.execute(cmd, ctx, event_tx).await?;
-            // TODO: Aggregate results properly instead of just taking the last one
-            combined_result.exit_code = result.exit_code;
+            // `execute()` returns Err for non-zero exit codes, so this code only
+            // runs for successful commands (exit_code == 0). We keep the first
+            // command's exit_code as it's semantically the "primary" result;
+            // duration and output are accumulated across all commands.
+            if i == 0 {
+                combined_result.exit_code = result.exit_code;
+            }
             combined_result.success &= result.success;
             combined_result.duration += result.duration;
             combined_result.stdout.push_str(&result.stdout);
@@ -356,15 +341,6 @@ impl crate::types::Engine {
         }
         Ok(combined_result)
     }
-
-    pub async fn cleanup(&self) -> Result<(), String> {
-        if let Some(ref cgroup) = self.cgroup_manager {
-            cgroup
-                .destroy()
-                .map_err(|e| format!("Failed to destroy cgroup: {}", e))?;
-        }
-        Ok(())
-    }
 }
 
 impl Default for crate::types::Engine {
@@ -376,131 +352,19 @@ impl Default for crate::types::Engine {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::collections::VecDeque;
+    use crate::Engine;
+    use crate::child::MockChild;
     use std::path::PathBuf;
-    use std::sync::{Arc, Mutex};
     use std::time::Duration;
 
-    #[derive(Clone, Debug)]
-    struct CapturedCall {
-        program: String,
-        args: Vec<String>,
-        env_vars: std::collections::HashMap<String, String>,
-        working_dir: PathBuf,
-    }
-
-    #[derive(Clone)]
-    struct MockExecutor {
-        calls: Arc<Mutex<VecDeque<CapturedCall>>>,
-    }
-
-    impl MockExecutor {
-        fn new() -> Self {
-            Self {
-                calls: Arc::new(Mutex::new(VecDeque::new())),
-            }
-        }
-
-        fn capture(
-            &self,
-            program: String,
-            args: Vec<String>,
-            env_vars: std::collections::HashMap<String, String>,
-            working_dir: PathBuf,
-        ) {
-            self.calls.lock().unwrap().push_back(CapturedCall {
-                program,
-                args,
-                env_vars,
-                working_dir,
-            });
-        }
-
-        fn pop(&self) -> Option<CapturedCall> {
-            self.calls.lock().unwrap().pop_front()
-        }
-    }
-
-    struct TestExecutor {
-        mock: MockExecutor,
-    }
-
-    impl TestExecutor {
-        fn new() -> Self {
-            Self {
-                mock: MockExecutor::new(),
-            }
-        }
-
-        async fn execute_command(
-            &self,
-            cmd: String,
-            ctx: &ExecutionContext,
-            _event_tx: EventSender,
-        ) -> Result<ExecutionResult, ExecutionError> {
-            let mut env_vars = ctx.env_vars.clone();
-            env_vars.insert("WS_TASKID".to_string(), ctx.task_id.clone());
-            env_vars.insert("WS_PIPELINE".to_string(), ctx.pipeline_name.clone());
-            env_vars.insert("WS_USERNAME".to_string(), ctx.username.clone());
-
-            self.mock.capture(
-                ctx.shell.clone(),
-                vec!["-c".to_string(), cmd.clone()],
-                env_vars,
-                ctx.working_dir.clone(),
-            );
-
-            Ok(ExecutionResult {
-                task_id: ctx.task_id.clone(),
-                exit_code: 0,
-                success: true,
-                duration: Duration::from_millis(10),
-                stdout: String::new(),
-                stderr: String::new(),
-                resource_usage: None,
-            })
-        }
-
-        async fn execute_script(
-            &self,
-            script_path: &PathBuf,
-            ctx: &ExecutionContext,
-            _event_tx: EventSender,
-        ) -> Result<ExecutionResult, ExecutionError> {
-            let abs_path = if script_path.is_absolute() {
-                script_path.clone()
-            } else {
-                ctx.working_dir.join(script_path)
-            };
-
-            self.mock.capture(
-                ctx.shell.clone(),
-                vec!["-c".to_string(), abs_path.to_string_lossy().into_owned()],
-                ctx.env_vars.clone(),
-                ctx.working_dir.clone(),
-            );
-
-            Ok(ExecutionResult {
-                task_id: ctx.task_id.clone(),
-                exit_code: 0,
-                success: true,
-                duration: Duration::from_millis(10),
-                stdout: String::new(),
-                stderr: String::new(),
-                resource_usage: None,
-            })
-        }
-    }
-
-    fn create_test_context() -> ExecutionContext {
+    fn test_ctx() -> ExecutionContext {
         ExecutionContext {
-            task_id: "test-task-1".to_string(),
+            task_id: "test-task".to_string(),
             pipeline_name: "test-pipeline".to_string(),
             username: "testuser".to_string(),
             working_dir: PathBuf::from("/tmp"),
             env_vars: std::collections::HashMap::new(),
-            timeout: Some(Duration::from_secs(30)),
-            cgroup_path: None,
+            timeout: None,
             shell: "bash".to_string(),
             dry_run: false,
             privileged: false,
@@ -508,152 +372,259 @@ mod tests {
         }
     }
 
+    // ------------------------------------------------------------------
+    // TaskStarted event
+    // ------------------------------------------------------------------
+
     #[tokio::test]
-    async fn test_execute_command_captures_args() {
-        let executor = TestExecutor::new();
-        let ctx = create_test_context();
+    async fn send_task_started_event_when_channel_provided() {
+        let engine = Engine::new();
+        let (tx, mut rx) = mpsc::channel(64);
+        let child = MockChild::with_exit_code(0);
 
-        let result = executor
-            .execute_command("echo hello".to_string(), &ctx, None)
-            .await;
+        let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await;
 
-        assert!(result.is_ok());
-
-        let call = executor.mock.pop();
-        assert!(call.is_some());
-
-        let call = call.unwrap();
-        assert_eq!(call.program, "bash");
-        assert!(call.args.contains(&"-c".to_string()));
-        assert!(call.args.contains(&"echo hello".to_string()));
+        assert!(matches!(
+            rx.try_recv(),
+            Ok(ExecutionEvent::TaskStarted { .. })
+        ));
     }
 
     #[tokio::test]
-    async fn test_execute_script_captures_path() {
-        let executor = TestExecutor::new();
-        let ctx = create_test_context();
+    async fn skip_task_started_event_when_channel_is_none() {
+        let engine = Engine::new();
+        let child = MockChild::with_exit_code(0);
 
-        let script_path = PathBuf::from("/tmp/test_script.sh");
-        let result = executor.execute_script(&script_path, &ctx, None).await;
+        let result = engine.execute_child(child, &test_ctx(), &None).await;
 
         assert!(result.is_ok());
+    }
 
-        let call = executor.mock.pop().unwrap();
-        assert!(call.args.contains(&"-c".to_string()));
-        assert!(call.args.iter().any(|a| a.contains("test_script.sh")));
+    // ------------------------------------------------------------------
+    // Exit code → success / failure mapping
+    // ------------------------------------------------------------------
+
+    #[tokio::test]
+    async fn exit_zero_returns_ok() {
+        let engine = Engine::new();
+        let child = MockChild::with_exit_code(0);
+
+        let result = engine.execute_child(child, &test_ctx(), &None).await;
+
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap().exit_code, 0);
     }
 
     #[tokio::test]
-    async fn test_execute_command_with_env_vars() {
-        let executor = TestExecutor::new();
-        let mut ctx = create_test_context();
-        ctx.env_vars
-            .insert("TEST_VAR".to_string(), "test_value".to_string());
+    async fn exit_nonzero_returns_err() {
+        let engine = Engine::new();
+        let child = MockChild::with_exit_code(1);
 
-        executor
-            .execute_command("echo $TEST_VAR".to_string(), &ctx, None)
-            .await
-            .unwrap();
+        let result = engine.execute_child(child, &test_ctx(), &None).await;
 
-        let call = executor.mock.pop().unwrap();
-        assert_eq!(
-            call.env_vars.get("TEST_VAR"),
-            Some(&"test_value".to_string())
+        assert!(result.is_err());
+        assert!(matches!(
+            result.unwrap_err(),
+            ExecutionError::ExecutionFailed(_)
+        ));
+    }
+
+    #[tokio::test]
+    async fn exit_nonzero_sends_task_failed_event() {
+        let engine = Engine::new();
+        let (tx, mut rx) = mpsc::channel(64);
+        let child = MockChild::with_exit_code(42);
+
+        let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await;
+
+        // Skip TaskStarted
+        let _ = rx.recv().await;
+        let event = rx.recv().await;
+        assert!(matches!(event, Some(ExecutionEvent::TaskFailed { .. })));
+        if let Some(ExecutionEvent::TaskFailed { error, .. }) = event {
+            assert!(error.contains("42"), "error should mention exit code 42");
+        }
+    }
+
+    #[tokio::test]
+    async fn exit_zero_sends_task_completed_event() {
+        let engine = Engine::new();
+        let (tx, mut rx) = mpsc::channel(64);
+        let child = MockChild::with_exit_code(0);
+
+        let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await;
+
+        let _ = rx.recv().await; // skip TaskStarted
+        let event = rx.recv().await;
+        assert!(matches!(event, Some(ExecutionEvent::TaskCompleted { .. })));
+    }
+
+    // ------------------------------------------------------------------
+    // Stdout / stderr capture
+    // ------------------------------------------------------------------
+
+    #[tokio::test]
+    async fn captures_stdout() {
+        let engine = Engine::new();
+        let child = MockChild::with_output("hello world\n", "");
+        let ctx = test_ctx();
+
+        let result = engine.execute_child(child, &ctx, &None).await.unwrap();
+
+        assert_eq!(result.stdout, "hello world\n");
+    }
+
+    #[tokio::test]
+    async fn captures_stderr() {
+        let engine = Engine::new();
+        let child = MockChild::with_output("", "error msg\n");
+        let ctx = test_ctx();
+
+        let result = engine.execute_child(child, &ctx, &None).await.unwrap();
+
+        assert_eq!(result.stderr, "error msg\n");
+    }
+
+    #[tokio::test]
+    async fn captures_both_streams() {
+        let engine = Engine::new();
+        let child = MockChild::with_output("out\n", "err\n");
+        let ctx = test_ctx();
+
+        let result = engine.execute_child(child, &ctx, &None).await.unwrap();
+
+        assert_eq!(result.stdout, "out\n");
+        assert_eq!(result.stderr, "err\n");
+    }
+
+    // ------------------------------------------------------------------
+    // OutputChunk events
+    // ------------------------------------------------------------------
+
+    #[tokio::test]
+    async fn sends_output_chunk_events() {
+        let engine = Engine::new();
+        let (tx, mut rx) = mpsc::channel(64);
+        let child = MockChild::with_output("chunk1\nchunk2\n", "");
+        let ctx = test_ctx();
+
+        let _ = engine.execute_child(child, &ctx, &Some(tx)).await;
+
+        let _ = rx.recv().await; // skip TaskStarted
+
+        let mut found_stdout = false;
+        while let Ok(event) = rx.try_recv() {
+            if matches!(
+                &event,
+                ExecutionEvent::OutputChunk {
+                    stream: StreamType::Stdout,
+                    ..
+                }
+            ) {
+                found_stdout = true;
+            }
+        }
+        assert!(
+            found_stdout,
+            "Should have received at least one OutputChunk for stdout"
         );
     }
 
+    // ------------------------------------------------------------------
+    // Timeout handling
+    // ------------------------------------------------------------------
+
     #[tokio::test]
-    async fn test_execute_includes_ws_environment() {
-        let executor = TestExecutor::new();
-        let ctx = create_test_context();
+    async fn timeout_kills_child_and_returns_err() {
+        let engine = Engine::new();
+        let (tx, mut rx) = mpsc::channel(64);
+        let child = MockChild::hanging();
+        let mut ctx = test_ctx();
+        ctx.timeout = Some(Duration::from_millis(10));
 
-        executor
-            .execute_command("ls".to_string(), &ctx, None)
-            .await
-            .unwrap();
+        let result = engine.execute_child(child, &ctx, &Some(tx)).await;
 
-        let call = executor.mock.pop().unwrap();
+        assert!(result.is_err());
+        assert!(matches!(result.unwrap_err(), ExecutionError::Timeout));
 
-        assert!(call.env_vars.contains_key("WS_TASKID"));
-        assert!(call.env_vars.contains_key("WS_PIPELINE"));
-        assert!(call.env_vars.contains_key("WS_USERNAME"));
+        // Should have received TaskFailed for timeout
+        let _ = rx.recv().await; // skip TaskStarted
+        let event = rx.recv().await;
+        assert!(
+            matches!(event, Some(ExecutionEvent::TaskFailed { error, .. }) if error == "Timeout")
+        );
+    }
+
+    // ------------------------------------------------------------------
+    // Result fields
+    // ------------------------------------------------------------------
+
+    #[tokio::test]
+    async fn result_contains_correct_task_id() {
+        let engine = Engine::new();
+        let child = MockChild::with_exit_code(0);
+        let ctx = test_ctx();
+
+        let result = engine.execute_child(child, &ctx, &None).await.unwrap();
+
+        assert_eq!(result.task_id, "test-task");
     }
 
     #[tokio::test]
-    async fn test_execute_command_preserves_working_dir() {
-        let executor = TestExecutor::new();
-        let mut ctx = create_test_context();
-        ctx.working_dir = PathBuf::from("/custom/path");
+    async fn result_contains_duration() {
+        let engine = Engine::new();
+        let child = MockChild::with_exit_code(0);
+        let ctx = test_ctx();
 
-        executor
-            .execute_command("pwd".to_string(), &ctx, None)
-            .await
-            .unwrap();
+        let result = engine.execute_child(child, &ctx, &None).await.unwrap();
 
-        let call = executor.mock.pop().unwrap();
-        assert_eq!(call.working_dir, PathBuf::from("/custom/path"));
+        assert!(result.duration.as_nanos() > 0);
     }
 
+    // ------------------------------------------------------------------
+    // dry_run does not execute_child
+    // ------------------------------------------------------------------
+
     #[tokio::test]
-    async fn test_local_executor_new_creates_default() {
-        let executor = super::super::types::Engine::new();
-        assert!(executor.cgroup_manager.is_none());
+    async fn dry_run_returns_ok_without_spawning() {
+        // This tests the `execute()` path: with dry_run=true it should
+        // return before reaching spawn / execute_child.
+        let engine = Engine::new();
+        let mut ctx = test_ctx();
+        ctx.dry_run = true;
+
+        let cmd = Command::new("true");
+        let result = engine.execute(cmd, &ctx, &None).await;
+
+        assert!(result.is_ok());
+        let res = result.unwrap();
+        assert!(res.success);
+        assert_eq!(res.stdout, "");
+        assert_eq!(res.stderr, "");
     }
 
-    #[tokio::test]
-    async fn test_execution_context_default() {
-        let ctx = ExecutionContext::default();
-
-        assert!(ctx.task_id.is_empty());
-        assert_eq!(ctx.shell, "bash");
-    }
+    // ------------------------------------------------------------------
+    // EventSender=None does not panic
+    // ------------------------------------------------------------------
 
     #[tokio::test]
-    async fn test_execution_result_success() {
-        let result = ExecutionResult {
-            task_id: "test".to_string(),
-            exit_code: 0,
-            success: true,
-            duration: Duration::from_secs(1),
-            stdout: "output".to_string(),
-            stderr: "".to_string(),
-            resource_usage: None,
-        };
+    async fn no_event_channel_does_not_panic() {
+        let engine = Engine::new();
 
-        assert!(result.success);
-        assert_eq!(result.exit_code, 0);
-    }
+        // Various exit codes
+        for code in [0, 1, 127] {
+            let child = MockChild::with_exit_code(code);
+            let _ = engine.execute_child(child, &test_ctx(), &None).await;
+        }
 
-    #[tokio::test]
-    async fn test_execution_result_failure() {
-        let result = ExecutionResult {
-            task_id: "test".to_string(),
-            exit_code: 1,
-            success: false,
-            duration: Duration::from_secs(1),
-            stdout: "".to_string(),
-            stderr: "error".to_string(),
-            resource_usage: None,
-        };
+        // With output
+        let child = MockChild::with_output("data", "err");
+        let _ = engine.execute_child(child, &test_ctx(), &None).await;
 
-        assert!(!result.success);
-        assert_eq!(result.exit_code, 1);
-    }
-
-    #[tokio::test]
-    async fn test_execution_error_variants() {
-        let _err1 = ExecutionError::InvalidCommand("test".to_string());
-        let _err2 = ExecutionError::ScriptNotFound(PathBuf::from("/nonexistent"));
-        let _err3 = ExecutionError::PackageManagerNotFound;
-        let _err4 = ExecutionError::Timeout;
-    }
-
-    #[tokio::test]
-    async fn test_resource_limits_default() {
-        let limits = ResourceLimits::default();
-
-        assert!(limits.memory_bytes.is_none());
-        assert!(limits.cpu_weight.is_none());
-        assert_eq!(limits.cpu_period_us, 0);
+        // Timeout
+        let mut ctx = test_ctx();
+        ctx.timeout = Some(Duration::from_millis(10));
+        let child = MockChild::hanging();
+        let _ = engine.execute_child(child, &ctx, &None).await;
     }
 }

workshop-engine/src/lib.rs

@@ -1,4 +1,4 @@
-pub mod cgroups;
+pub mod child;
 pub mod command;
 pub mod error;
 pub mod executor;
@@ -8,15 +8,21 @@ pub mod time;
 pub mod types;
 pub mod upm;
 
+// Re-export child types for convenience
+pub use child::{ManagedChild, TokioChild};
+
 // Convenience re-exports
-pub use error::{ExecutionError, HasExitCode,
-    EXITCODE_OK, EXITCODE_GENERAL_ERROR, EXITCODE_INVALID_ARGS,
-    EXITCODE_IO_ERROR, EXITCODE_PARSE_ERROR, EXITCODE_EXECUTION_ERROR,
-    EXITCODE_TIMEOUT, EXITCODE_PRIV_DROP_FAILED};
-pub use types::{Engine, EventSender, ExecutionContext, ExecutionEvent, ExecutionResult,
-    StreamType, ResourceLimits, ResourceUsage, DeltaExecutionContext};
 pub use command::CustomCommand;
-pub use repology::RepologyEndpoint;
-pub use time::{parse_duration_to_ms, deserialize_duration_ms, deserialize_option_duration_ms};
+pub use error::{
+    EXITCODE_EXECUTION_ERROR, EXITCODE_GENERAL_ERROR, EXITCODE_INVALID_ARGS, EXITCODE_IO_ERROR,
+    EXITCODE_OK, EXITCODE_PARSE_ERROR, EXITCODE_PRIV_DROP_FAILED, EXITCODE_TIMEOUT, ExecutionError,
+    HasExitCode,
+};
 pub use pm::PackageManager;
+pub use repology::RepologyEndpoint;
+pub use time::{deserialize_duration_ms, deserialize_option_duration_ms, parse_duration_to_ms};
+pub use types::{
+    DeltaExecutionContext, Engine, EventSender, ExecutionContext, ExecutionEvent, ExecutionResult,
+    StreamType,
+};
 pub use upm::UserPackageManager;

workshop-engine/src/types.rs

@@ -1,20 +1,11 @@
-use serde::{Deserialize, Serialize};
+use serde::Serialize;
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::time::Duration;
 use tokio::sync::mpsc;
 
-pub use crate::cgroups::CgroupManager;
 pub use crate::error::{EXITCODE_EXECUTION_ERROR, EXITCODE_IO_ERROR, ExecutionError};
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct ResourceUsage {
-    pub cpu_time_secs: f64,
-    pub max_memory_kb: u64,
-    pub io_read_kb: u64,
-    pub io_write_kb: u64,
-}
-
 #[derive(Debug, Clone)]
 pub struct ExecutionContext {
     pub task_id: String,
@@ -23,7 +14,6 @@ pub struct ExecutionContext {
     pub working_dir: PathBuf,
     pub env_vars: HashMap<String, String>,
     pub timeout: Option<Duration>,
-    pub cgroup_path: Option<PathBuf>,
     pub shell: String,
     pub dry_run: bool,
     pub privileged: bool,
@@ -46,7 +36,6 @@ impl Default for ExecutionContext {
             working_dir: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
             env_vars: HashMap::new(),
             timeout: None,
-            cgroup_path: None,
             shell: String::from("bash"),
             dry_run: false,
             privileged: false,
@@ -55,14 +44,6 @@ impl Default for ExecutionContext {
     }
 }
 
-#[derive(Debug, Clone, Default)]
-pub struct ResourceLimits {
-    pub memory_bytes: Option<u64>,
-    pub cpu_weight: Option<u64>,
-    pub cpu_max_us: Option<u64>,
-    pub cpu_period_us: u64,
-}
-
 #[derive(Debug, Clone, Serialize)]
 pub struct ExecutionResult {
     pub task_id: String,
@@ -71,7 +52,6 @@ pub struct ExecutionResult {
     pub duration: Duration,
     pub stdout: String,
     pub stderr: String,
-    pub resource_usage: Option<ResourceUsage>,
 }
 
 impl Default for ExecutionResult {
@@ -83,7 +63,6 @@ impl Default for ExecutionResult {
             duration: Duration::default(),
             stdout: String::new(),
             stderr: String::new(),
-            resource_usage: None,
         }
     }
 }
@@ -117,6 +96,4 @@ pub enum ExecutionEvent {
 
 pub type EventSender = Option<mpsc::Sender<ExecutionEvent>>;
 
-pub struct Engine {
-    pub cgroup_manager: Option<CgroupManager>,
-}
+pub struct Engine;

workshop-engine/src/upm.rs

@@ -3,8 +3,8 @@ use async_trait::async_trait;
 use serde_yaml::Value;
 mod custom;
 
-use crate::types::ExecutionContext;
 use crate::error::DependencyError;
+use crate::types::ExecutionContext;
 
 /// System dependencies required by a user package manager.
 pub struct UPMSysDeps {

workshop-engine/src/upm/custom.rs

@@ -2,11 +2,11 @@ use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 
-use crate::types::Engine;
-use crate::types::DeltaExecutionContext;
-use crate::upm::UPMSysDeps;
-use crate::error::DependencyError;
 use crate::command::CustomCommand;
+use crate::error::DependencyError;
+use crate::types::DeltaExecutionContext;
+use crate::types::Engine;
+use crate::upm::UPMSysDeps;
 
 use super::UserPackageManager;
 

workshop-engine/tests/engine_integration.rs

@@ -1,7 +1,7 @@
-//! Integration tests for workshop-executor engine
+//! Integration tests for workshop-engine
 //!
 //! These tests execute real commands and verify results.
-//! Run with: cargo test --test engine_integration
+//! Run with: cargo test --test engine_integration -p workshop-engine
 //!
 //! Note: These are TRUE integration tests - they run in a separate test binary
 //! and execute actual commands on the system.
@@ -9,7 +9,7 @@
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::time::Duration;
-use workshop_baker::engine::{Engine, ExecutionContext};
+use workshop_engine::{Engine, ExecutionContext};
 
 fn create_executor() -> Engine {
     Engine::new()
@@ -23,7 +23,6 @@ fn create_context() -> ExecutionContext {
         working_dir: std::env::temp_dir(),
         env_vars: HashMap::new(),
         timeout: Some(Duration::from_secs(30)),
-        cgroup_path: None,
         shell: "bash".to_string(),
         dry_run: false,
         privileged: false,
@@ -173,7 +172,6 @@ async fn test_execute_respects_working_directory() {
         working_dir: PathBuf::from("/tmp"),
         env_vars: HashMap::new(),
         timeout: Some(Duration::from_secs(30)),
-        cgroup_path: None,
         shell: "bash".to_string(),
         standalone: false,
         ..Default::default()
@@ -204,3 +202,69 @@ async fn test_execute_captures_stderr() {
     assert!(result.success);
     assert!(result.stderr.contains("error"));
 }
+
+/// Test: with `process_group(0)`, kill() propagates to the entire process tree.
+///
+/// We spawn a bash that creates a background sleep. With process_group(0),
+/// bash is the process group leader. When we kill(-PGID), the sleep should die too.
+///
+/// This test uses tokio::process::Command + process_group(0) directly
+/// (bypassing Engine) to verify the kernel-level behavior.
+#[tokio::test]
+async fn test_process_group_kill_kills_subprocesses() {
+    use std::os::unix::process::CommandExt;
+    use std::process::Stdio;
+    use tokio::io::AsyncBufReadExt;
+    use tokio::process::Command;
+
+    let mut cmd = Command::new("bash");
+    cmd.args(["-c", "sleep 999 & echo $!"]);
+    cmd.process_group(0);
+    cmd.stdin(Stdio::null());
+    cmd.stdout(Stdio::piped());
+    cmd.stderr(Stdio::piped());
+
+    let mut child = cmd.spawn().expect("spawn failed");
+    let bash_pid = child.id().expect("no pid");
+    eprintln!("[test] bash PID: {}", bash_pid);
+
+    // Read the background sleep's PID from stdout
+    let stdout = child.stdout.take().expect("stdout pipe");
+    let mut reader = tokio::io::BufReader::new(stdout);
+    let mut pid_line = String::new();
+    reader
+        .read_line(&mut pid_line)
+        .await
+        .expect("read pid line");
+    let sleep_pid: u32 = pid_line.trim().parse().expect("valid sleep PID");
+    eprintln!("[test] sleep PID: {}", sleep_pid);
+
+    // Verify sleep is alive before kill
+    let alive_before = std::path::Path::new(&format!("/proc/{}", sleep_pid)).exists();
+    eprintln!("[test] sleep alive before kill: {}", alive_before);
+
+    // Kill by sending SIGKILL to the NEGATIVE PGID (entire process group)
+    unsafe {
+        libc::kill(-(bash_pid as i32), libc::SIGKILL);
+    }
+    child.wait().await.expect("wait failed");
+    eprintln!("[test] kill+wait done");
+
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+
+    // Check if sleep survived — it SHOULD be dead now
+    let alive_after = std::path::Path::new(&format!("/proc/{}", sleep_pid)).exists();
+    eprintln!("[test] sleep alive after kill: {}", alive_after);
+
+    // Cleanup just in case
+    let _ = std::process::Command::new("kill")
+        .arg("-9")
+        .arg(sleep_pid.to_string())
+        .spawn();
+
+    assert!(alive_before, "Test setup: sleep was never created");
+    assert!(
+        !alive_after,
+        "sleep survived process-group kill! process_group(0) fix did not work."
+    );
+}