diff --git a/benches/basic/creation.rs b/benches/basic/creation.rs
index 496b5fd..97d4aa1 100644
--- a/benches/basic/creation.rs
+++ b/benches/basic/creation.rs
@@ -21,7 +21,7 @@ pub fn bench_dag_creation(c: &mut Criterion) {
 
     c.bench_function("add_100_tasks", |b| {
         b.iter(|| {
-            let dag = DagRunner::new();
+            let mut dag = DagRunner::new();
             for i in 0..100 {
                 black_box(dag.add_task(Value(i)));
             }
diff --git a/benches/basic/scaling.rs b/benches/basic/scaling.rs
index 1be3a1c..2ce8e9c 100644
--- a/benches/basic/scaling.rs
+++ b/benches/basic/scaling.rs
@@ -15,11 +15,12 @@ pub fn bench_dag_scaling(c: &mut Criterion) {
         group.bench_with_input(BenchmarkId::from_parameter(size), size, |b, &size| {
             b.iter(|| {
                 rt.block_on(async {
-                    let dag = DagRunner::new();
+                    let mut dag = DagRunner::new();
                     for i in 0..size {
                         dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
                     }
-                    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                    let _output = dag
+                        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                         .await
                         .unwrap();
                 })
diff --git a/benches/comparisons/deep_chain.rs b/benches/comparisons/deep_chain.rs
index 932601d..055fca5 100644
--- a/benches/comparisons/deep_chain.rs
+++ b/benches/comparisons/deep_chain.rs
@@ -3,7 +3,6 @@
 //! Long sequential chain: 100 tasks in sequence
 
 use criterion::Criterion;
-use dagx::TaskHandle;
 
 pub fn bench_deep_chain(c: &mut Criterion) {
     let rt = tokio::runtime::Runtime::new().unwrap();
@@ -16,9 +15,9 @@ pub fn bench_deep_chain(c: &mut Criterion) {
                 use dagx::DagRunner;
                 use dagx_test::task_fn;
 
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
 
-                let first: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 0)).into();
+                let first = dag.add_task(task_fn::<(), _, _>(|_: ()| 0));
                 let mut prev = dag
                     .add_task(task_fn::<i32, _, _>(|&x: &i32| x + 1))
                     .depends_on(first);
@@ -29,10 +28,11 @@ pub fn bench_deep_chain(c: &mut Criterion) {
                         .depends_on(prev);
                 }
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let mut output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
-                dag.get(prev).unwrap()
+                output.get(prev).unwrap()
             })
         });
     });
diff --git a/benches/comparisons/diamond_pattern.rs b/benches/comparisons/diamond_pattern.rs
index 5aae415..2088e6f 100644
--- a/benches/comparisons/diamond_pattern.rs
+++ b/benches/comparisons/diamond_pattern.rs
@@ -3,7 +3,6 @@
 //! Classic diamond: A → B,C → D
 
 use criterion::Criterion;
-use dagx::TaskHandle;
 
 pub fn bench_diamond_pattern(c: &mut Criterion) {
     let rt = tokio::runtime::Runtime::new().unwrap();
@@ -16,9 +15,9 @@ pub fn bench_diamond_pattern(c: &mut Criterion) {
                 use dagx::DagRunner;
                 use dagx_test::task_fn;
 
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
 
-                let a: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)).into();
+                let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
                 let b = dag
                     .add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
                     .depends_on(a);
@@ -29,10 +28,11 @@ pub fn bench_diamond_pattern(c: &mut Criterion) {
                     .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y))
                     .depends_on((&b, &c));
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let mut output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
-                dag.get(d).unwrap()
+                output.get(d).unwrap()
             })
         });
     });
diff --git a/benches/comparisons/etl_pipeline.rs b/benches/comparisons/etl_pipeline.rs
index 1363546..75fe567 100644
--- a/benches/comparisons/etl_pipeline.rs
+++ b/benches/comparisons/etl_pipeline.rs
@@ -15,7 +15,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) {
                 use dagx::DagRunner;
                 use dagx_test::task_fn;
 
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
 
                 // Extract from 3 sources
                 let source1 = dag.add_task(task_fn::<(), _, _>(|_: ()| vec![1, 2, 3, 4, 5]));
@@ -70,10 +70,11 @@ pub fn bench_etl_pipeline(c: &mut Criterion) {
                     }))
                     .depends_on(validate);
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let mut output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
-                dag.get(load).unwrap()
+                output.get(load).unwrap()
             })
         });
     });
diff --git a/benches/comparisons/large_scale.rs b/benches/comparisons/large_scale.rs
index b23a562..16446ed 100644
--- a/benches/comparisons/large_scale.rs
+++ b/benches/comparisons/large_scale.rs
@@ -17,12 +17,13 @@ pub fn bench_large_scale(c: &mut Criterion) {
                 use dagx::DagRunner;
                 use dagx_test::task_fn;
 
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
                 for i in 0..10_000 {
                     dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 2));
                 }
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let _output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
             })
diff --git a/benches/comparisons/linear_pipeline.rs b/benches/comparisons/linear_pipeline.rs
index 6e787c0..4a1f0c0 100644
--- a/benches/comparisons/linear_pipeline.rs
+++ b/benches/comparisons/linear_pipeline.rs
@@ -15,7 +15,7 @@ pub fn bench_linear_pipeline(c: &mut Criterion) {
                 use dagx::DagRunner;
                 use dagx_test::task_fn;
 
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
 
                 let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
                 let b = dag
@@ -31,10 +31,11 @@ pub fn bench_linear_pipeline(c: &mut Criterion) {
                     .add_task(task_fn::<i32, _, _>(|&x: &i32| x * 3))
                     .depends_on(d);
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let mut output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
-                dag.get(e).unwrap()
+                output.get(e).unwrap()
             })
         });
     });
diff --git a/benches/comparisons/wide_fanout.rs b/benches/comparisons/wide_fanout.rs
index fa5149c..60881b0 100644
--- a/benches/comparisons/wide_fanout.rs
+++ b/benches/comparisons/wide_fanout.rs
@@ -3,7 +3,6 @@
 //! One source task feeding 100 dependent tasks
 
 use criterion::Criterion;
-use dagx::TaskHandle;
 
 pub fn bench_wide_fanout(c: &mut Criterion) {
     let rt = tokio::runtime::Runtime::new().unwrap();
@@ -16,8 +15,8 @@ pub fn bench_wide_fanout(c: &mut Criterion) {
                 use dagx::DagRunner;
                 use dagx_test::task_fn;
 
-                let dag = DagRunner::new();
-                let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)).into();
+                let mut dag = DagRunner::new();
+                let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
 
                 // Create 100 tasks that all depend on source
                 for i in 0..100 {
@@ -25,7 +24,8 @@ pub fn bench_wide_fanout(c: &mut Criterion) {
                         .depends_on(source);
                 }
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let _output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
             })
diff --git a/benches/patterns/breakdown.rs b/benches/patterns/breakdown.rs
index 196d857..78e8942 100644
--- a/benches/patterns/breakdown.rs
+++ b/benches/patterns/breakdown.rs
@@ -12,7 +12,7 @@ pub fn bench_10k_breakdown(c: &mut Criterion) {
     // a) Adding 10k tasks (construction only)
     group.bench_function("construction", |b| {
         b.iter(|| {
-            let dag = DagRunner::new();
+            let mut dag = DagRunner::new();
             for i in 0..10_000 {
                 dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
             }
@@ -23,11 +23,12 @@ pub fn bench_10k_breakdown(c: &mut Criterion) {
     group.bench_function("full_execution", |b| {
         b.iter(|| {
             rt.block_on(async {
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
                 for i in 0..10_000 {
                     dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
                 }
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let _output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
             })
@@ -38,7 +39,7 @@ pub fn bench_10k_breakdown(c: &mut Criterion) {
     group.bench_function("execution_only", |b| {
         b.iter_batched(
             || {
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
                 for i in 0..10_000 {
                     dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
                 }
@@ -46,7 +47,8 @@ pub fn bench_10k_breakdown(c: &mut Criterion) {
             },
             |dag| {
                 rt.block_on(async {
-                    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                    let _output = dag
+                        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                         .await
                         .unwrap();
                 })
diff --git a/benches/patterns/diamond.rs b/benches/patterns/diamond.rs
index 75a2263..14f1fda 100644
--- a/benches/patterns/diamond.rs
+++ b/benches/patterns/diamond.rs
@@ -1,7 +1,7 @@
 //! Diamond pattern benchmarks (fan-out then fan-in)
 
 use criterion::Criterion;
-use dagx::{DagRunner, TaskHandle};
+use dagx::DagRunner;
 use dagx_test::task_fn;
 pub fn bench_diamond(c: &mut Criterion) {
     let rt = tokio::runtime::Runtime::new().unwrap();
@@ -9,11 +9,10 @@ pub fn bench_diamond(c: &mut Criterion) {
     c.bench_function("diamond_pattern_50", |b| {
         b.iter(|| {
             rt.block_on(async {
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
 
                 for i in 0..50 {
-                    let source: TaskHandle<_> =
-                        dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into();
+                    let source = dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
 
                     let left = dag
                         .add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
@@ -26,7 +25,8 @@ pub fn bench_diamond(c: &mut Criterion) {
                         .depends_on((&left, &right));
                 }
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let _output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
             })
diff --git a/benches/patterns/fanout.rs b/benches/patterns/fanout.rs
index 0b49035..e411a9e 100644
--- a/benches/patterns/fanout.rs
+++ b/benches/patterns/fanout.rs
@@ -1,7 +1,7 @@
 //! Fan-out pattern benchmarks (1 → N dependencies)
 
 use criterion::Criterion;
-use dagx::{DagRunner, TaskHandle};
+use dagx::DagRunner;
 use dagx_test::task_fn;
 pub fn bench_fanout(c: &mut Criterion) {
     let rt = tokio::runtime::Runtime::new().unwrap();
@@ -10,15 +10,16 @@ pub fn bench_fanout(c: &mut Criterion) {
     c.bench_function("fanout_1_to_100_i32", |b| {
         b.iter(|| {
             rt.block_on(async {
-                let dag = DagRunner::new();
-                let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)).into();
+                let mut dag = DagRunner::new();
+                let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
 
                 for i in 0..100 {
                     dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
                         .depends_on(source);
                 }
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let _output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
             })
@@ -29,13 +30,11 @@ pub fn bench_fanout(c: &mut Criterion) {
     c.bench_function("fanout_1_to_100_vec", |b| {
         b.iter(|| {
             rt.block_on(async {
-                let dag = DagRunner::new();
-                let source: TaskHandle<_> = dag
-                    .add_task(task_fn::<(), _, _>(|_: ()| {
-                        // Create 1000 strings - framework wraps in Arc automatically
-                        (0..1000).map(|i| format!("Item {}", i)).collect::<Vec<_>>()
-                    }))
-                    .into();
+                let mut dag = DagRunner::new();
+                let source = dag.add_task(task_fn::<(), _, _>(|_: ()| {
+                    // Create 1000 strings - framework wraps in Arc automatically
+                    (0..1000).map(|i| format!("Item {}", i)).collect::<Vec<_>>()
+                }));
 
                 for i in 0..100 {
                     dag.add_task(task_fn::<Vec<_>, _, _>(move |data: &Vec<String>| {
@@ -45,7 +44,8 @@ pub fn bench_fanout(c: &mut Criterion) {
                     .depends_on(source);
                 }
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let _output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
             })
diff --git a/benches/realistic/etl.rs b/benches/realistic/etl.rs
index 2be6428..13381c4 100644
--- a/benches/realistic/etl.rs
+++ b/benches/realistic/etl.rs
@@ -1,7 +1,7 @@
 //! ETL (Extract, Transform, Load) pipeline benchmark
 
 use criterion::{BenchmarkId, Criterion};
-use dagx::{DagRunner, TaskHandle};
+use dagx::DagRunner;
 use dagx_test::task_fn;
 pub fn bench_etl_pipeline(c: &mut Criterion) {
     let rt = tokio::runtime::Runtime::new().unwrap();
@@ -17,23 +17,21 @@ pub fn bench_etl_pipeline(c: &mut Criterion) {
             |b, &count| {
                 b.iter(|| {
                     rt.block_on(async {
-                        let dag = DagRunner::new();
+                        let mut dag = DagRunner::new();
 
                         // Extract: Load JSON-like data (framework wraps in Arc automatically)
-                        let extract: TaskHandle<_> = dag
-                            .add_task(task_fn::<(), _, _>(move |_: ()| {
-                                (0..count)
-                                    .map(|i| {
-                                        format!(
-                                            "{{\"id\":{},\"name\":\"User{}\",\"score\":{}}}",
-                                            i,
-                                            i,
-                                            i * 10
-                                        )
-                                    })
-                                    .collect::<Vec<_>>()
-                            }))
-                            .into();
+                        let extract = dag.add_task(task_fn::<(), _, _>(move |_: ()| {
+                            (0..count)
+                                .map(|i| {
+                                    format!(
+                                        "{{\"id\":{},\"name\":\"User{}\",\"score\":{}}}",
+                                        i,
+                                        i,
+                                        i * 10
+                                    )
+                                })
+                                .collect::<Vec<_>>()
+                        }));
 
                         // Transform: Three parallel transformations
 
@@ -92,7 +90,8 @@ pub fn bench_etl_pipeline(c: &mut Criterion) {
                         ))
                         .depends_on((&validate, &scores, &names));
 
-                        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                        let _output = dag
+                            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                             .await
                             .unwrap();
                     })
diff --git a/benches/realistic/memory_efficiency.rs b/benches/realistic/memory_efficiency.rs
index 61ae2c8..3ea0e35 100644
--- a/benches/realistic/memory_efficiency.rs
+++ b/benches/realistic/memory_efficiency.rs
@@ -1,7 +1,7 @@
 //! Memory efficiency benchmark with large data structures
 
 use criterion::{BenchmarkId, Criterion};
-use dagx::{DagRunner, TaskHandle};
+use dagx::DagRunner;
 use dagx_test::task_fn;
 use std::hint::black_box;
 
@@ -10,21 +10,19 @@ pub fn bench_memory_efficiency(c: &mut Criterion) {
     let mut group = c.benchmark_group("memory_efficiency");
 
     // Test with different data sizes (in MB)
-    for mb_size in [1, 5, 10].iter() {
+    for mb_size in [1, 10].iter() {
         group.bench_with_input(
             BenchmarkId::new("large_data_20_consumers", format!("{}MB", mb_size)),
             mb_size,
             |b, &mb| {
                 b.iter(|| {
                     rt.block_on(async {
-                        let dag = DagRunner::new();
+                        let mut dag = DagRunner::new();
 
-                        let source: TaskHandle<_> = dag
-                            .add_task(task_fn::<(), _, _>(move |_: ()| {
-                                // Create MB of data - framework wraps in Arc automatically
-                                vec![0u8; mb * 1024 * 1024]
-                            }))
-                            .into();
+                        let source = dag.add_task(task_fn::<(), _, _>(move |_: ()| {
+                            // Create MB of data - framework wraps in Arc automatically
+                            vec![0u8; mb * 1024 * 1024]
+                        }));
 
                         // 20 consumers share the data via automatic Arc wrapping
                         for i in 0..20 {
@@ -41,7 +39,8 @@ pub fn bench_memory_efficiency(c: &mut Criterion) {
                             .depends_on(source);
                         }
 
-                        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                        let _output = dag
+                            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                             .await
                             .unwrap();
                     })
diff --git a/benches/realistic/mixed.rs b/benches/realistic/mixed.rs
index 5f34d8d..e9d2d08 100644
--- a/benches/realistic/mixed.rs
+++ b/benches/realistic/mixed.rs
@@ -1,7 +1,7 @@
 //! Mixed realistic patterns - fan-out, fan-in, and processing
 
 use criterion::Criterion;
-use dagx::{DagRunner, TaskHandle};
+use dagx::DagRunner;
 use dagx_test::task_fn;
 pub fn bench_mixed_patterns(c: &mut Criterion) {
     let rt = tokio::runtime::Runtime::new().unwrap();
@@ -9,16 +9,14 @@ pub fn bench_mixed_patterns(c: &mut Criterion) {
     c.bench_function("mixed_patterns_realistic", |b| {
         b.iter(|| {
             rt.block_on(async {
-                let dag = DagRunner::new();
+                let mut dag = DagRunner::new();
 
                 // Stage 1: Load data (framework wraps in Arc automatically)
-                let data: TaskHandle<_> = dag
-                    .add_task(task_fn::<(), _, _>(|_: ()| {
-                        (0..1000)
-                            .map(|i| format!("Record-{:04}", i))
-                            .collect::<Vec<_>>()
-                    }))
-                    .into();
+                let data = dag.add_task(task_fn::<(), _, _>(|_: ()| {
+                    (0..1000)
+                        .map(|i| format!("Record-{:04}", i))
+                        .collect::<Vec<_>>()
+                }));
 
                 // Stage 2: Fan-out for parallel analysis
                 let analysis1 = dag
@@ -54,7 +52,8 @@ pub fn bench_mixed_patterns(c: &mut Criterion) {
                     .depends_on(summary);
                 }
 
-                dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+                let _output = dag
+                    .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
                     .await
                     .unwrap();
             })
diff --git a/dagx-macros/src/lib.rs b/dagx-macros/src/lib.rs
index 5dc1d23..2abcc6c 100644
--- a/dagx-macros/src/lib.rs
+++ b/dagx-macros/src/lib.rs
@@ -152,8 +152,7 @@ use syn::{parse_macro_input, FnArg, Ident, ImplItem, ItemImpl, Pat, PatType, Ret
 /// }
 ///
 /// // Generated:
-/// impl Task for Add {
-///     type Input = (i32, i32);
+/// impl Task<(i32, i32)> for Add {
 ///     type Output = i32;
 ///
 ///     async fn run(&mut self, input: Self::Input) -> Self::Output {
@@ -183,8 +182,7 @@ use syn::{parse_macro_input, FnArg, Ident, ImplItem, ItemImpl, Pat, PatType, Ret
 /// }
 ///
 /// // Generated:
-/// impl Task for Counter {
-///     type Input = i32;
+/// impl Task<(i32)> for Counter {
 ///     type Output = i32;
 ///
 ///     async fn run(&mut self, input: Self::Input) -> Self::Output {
diff --git a/dagx-test/src/lib.rs b/dagx-test/src/lib.rs
index 889973b..ab72a8a 100644
--- a/dagx-test/src/lib.rs
+++ b/dagx-test/src/lib.rs
@@ -5,6 +5,8 @@
 
 #![cfg(not(tarpaulin_include))]
 
+use std::marker::PhantomData;
+
 use crate::extract::ExtractInput;
 
 use dagx::{Task, TaskInput};
@@ -18,7 +20,7 @@ where
     O: Send,
 {
     f: F,
-    _phantom: std::marker::PhantomData<fn(I) -> O>,
+    _phantom: PhantomData<fn(I) -> O>,
 }
 
 impl<I, O, F> Task<I::Input> for TaskFn<I, O, F>
@@ -45,6 +47,6 @@ where
 {
     TaskFn {
         f,
-        _phantom: std::marker::PhantomData,
+        _phantom: PhantomData,
     }
 }
diff --git a/examples/01_basic.rs b/examples/01_basic.rs
index 5bd065c..be8ac2e 100644
--- a/examples/01_basic.rs
+++ b/examples/01_basic.rs
@@ -77,13 +77,13 @@ impl Add {
 #[tokio::main]
 async fn main() {
     // Step 3: Create a new DAG
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Step 4: Add source tasks (tasks with no dependencies)
     //
     // add_task() returns a TaskHandle<T> where T is the task's output type
-    let x = dag.add_task(Value::new(2)).into(); // TaskHandle<i32>
-    let y = dag.add_task(Value::new(3)).into(); // TaskHandle<i32>
+    let x = dag.add_task(Value::new(2)); // TaskHandle<i32>
+    let y = dag.add_task(Value::new(3)); // TaskHandle<i32>
 
     // Step 5: Add a task with dependencies
     //
@@ -105,7 +105,8 @@ async fn main() {
     // 1. Tasks with no dependencies run first (x and y in parallel)
     // 2. Once x and y complete, sum runs (it depends on both)
     println!("Running DAG...\n");
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
@@ -113,9 +114,9 @@ async fn main() {
     //
     // Use .get() with a TaskHandle to retrieve the output.
     // Results are not cloned, so you can only retrieve them once.
-    println!("\nResult: {}", dag.get(sum).unwrap());
+    println!("\nResult: {}", output.get(sum).unwrap());
 
     // Note: You can also retrieve intermediate results
-    assert_eq!(dag.get(x).unwrap(), 2);
-    assert_eq!(dag.get(y).unwrap(), 3);
+    assert_eq!(output.get(x).unwrap(), 2);
+    assert_eq!(output.get(y).unwrap(), 3);
 }
diff --git a/examples/02_fan_out.rs b/examples/02_fan_out.rs
index f387a79..25513cb 100644
--- a/examples/02_fan_out.rs
+++ b/examples/02_fan_out.rs
@@ -55,7 +55,7 @@
 //!   Squared: 100
 //! ```
 
-use dagx::{task, DagRunner, TaskHandle};
+use dagx::{task, DagRunner};
 
 // Source task with a value field
 struct Source {
@@ -127,10 +127,10 @@ impl Power {
 
 #[tokio::main]
 async fn main() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Single source task - constructed with ::new()
-    let source: TaskHandle<_> = dag.add_task(Source::new(10)).into();
+    let source = dag.add_task(Source::new(10));
 
     // Multiple downstream tasks consuming the same value
     // Each constructed differently to show flexibility
@@ -140,14 +140,15 @@ async fn main() {
 
     // Run the DAG
     println!("Running fan-out DAG...\n");
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
     // Print all results
     println!("\nResults:");
-    println!("  Source: {}", dag.get(source).unwrap());
-    println!("  Plus One: {}", dag.get(plus_one).unwrap());
-    println!("  Times Two: {}", dag.get(times_two).unwrap());
-    println!("  Squared: {}", dag.get(squared).unwrap());
+    println!("  Source: {}", output.get(source).unwrap());
+    println!("  Plus One: {}", output.get(plus_one).unwrap());
+    println!("  Times Two: {}", output.get(times_two).unwrap());
+    println!("  Squared: {}", output.get(squared).unwrap());
 }
diff --git a/examples/03_fan_in.rs b/examples/03_fan_in.rs
index 48596c1..05db4f6 100644
--- a/examples/03_fan_in.rs
+++ b/examples/03_fan_in.rs
@@ -118,7 +118,7 @@ impl BuildProfile {
 
 #[tokio::main]
 async fn main() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Multiple independent source tasks
     let name = dag.add_task(FetchName);
@@ -133,10 +133,11 @@ async fn main() {
 
     // Run the DAG
     println!("Running fan-in DAG...\n");
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
     // Print result
-    println!("\n{}", dag.get(profile).unwrap());
+    println!("\n{}", output.get(profile).unwrap());
 }
diff --git a/examples/04_parallel_computation.rs b/examples/04_parallel_computation.rs
index 1c57389..4036740 100644
--- a/examples/04_parallel_computation.rs
+++ b/examples/04_parallel_computation.rs
@@ -115,7 +115,7 @@ async fn main() -> DagResult<()> {
     println!("Running PARALLEL computation (4 workers on separate threads)...");
     let (parallel_time, parallel_result) = {
         let start = Instant::now();
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         // Create 4 workers - dagx will run them in parallel
         let sum1 = dag.add_task(ComputeSum::new(1, 251, WORK_DELAY));
@@ -125,10 +125,11 @@ async fn main() -> DagResult<()> {
 
         let total = dag.add_task(Aggregate).depends_on((sum1, sum2, sum3, sum4));
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await?;
 
-        let result = dag.get(total)?;
+        let result = output.get(total)?;
         let elapsed = start.elapsed();
 
         println!(
diff --git a/examples/circuit_breaker.rs b/examples/circuit_breaker.rs
index c661448..1c65b1e 100644
--- a/examples/circuit_breaker.rs
+++ b/examples/circuit_breaker.rs
@@ -381,7 +381,7 @@ async fn main() {
         // Create shared circuit breaker: 3 failures trigger open, 3 second recovery timeout
         let circuit = Arc::new(CircuitBreaker::new(3, Duration::from_secs(3)));
 
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         // Request 1: Success (circuit closed)
         let call1 = dag.add_task(ProtectedServiceCall::new(
@@ -390,56 +390,60 @@ async fn main() {
             false, // success
         ));
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        println!("Result 1: {:?}\n", dag.get(call1).unwrap());
+        println!("Result 1: {:?}\n", output.get(call1).unwrap());
 
         // Request 2: Failure (circuit closed, count = 1)
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
         let call2 = dag.add_task(ProtectedServiceCall::new(
             "ServiceCall-2",
             circuit.clone(),
             true, // fail
         ));
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        println!("Result 2: {:?}\n", dag.get(call2).unwrap());
+        println!("Result 2: {:?}\n", output.get(call2).unwrap());
 
         // Request 3: Failure (circuit closed, count = 2)
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
         let call3 = dag.add_task(ProtectedServiceCall::new(
             "ServiceCall-3",
             circuit.clone(),
             true, // fail
         ));
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        println!("Result 3: {:?}\n", dag.get(call3).unwrap());
+        println!("Result 3: {:?}\n", output.get(call3).unwrap());
 
         // Request 4: Failure (circuit closed, count = 3, opens!)
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
         let call4 = dag.add_task(ProtectedServiceCall::new(
             "ServiceCall-4",
             circuit.clone(),
             true, // fail
         ));
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        println!("Result 4: {:?}\n", dag.get(call4).unwrap());
+        println!("Result 4: {:?}\n", output.get(call4).unwrap());
 
         // Request 5: Circuit is now open, fails fast with fallback
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
         let call5 = dag.add_task(ProtectedServiceCall::new(
             "ServiceCall-5",
             circuit.clone(),
@@ -450,11 +454,12 @@ async fn main() {
             .add_task(FallbackService::new("ServiceCall-5"))
             .depends_on(call5);
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        println!("Result 5: {:?}\n", dag.get(fallback5).unwrap());
+        println!("Result 5: {:?}\n", output.get(fallback5).unwrap());
 
         // Example 2: Circuit recovery after timeout
         println!("Example 2: Circuit recovery after timeout\n");
@@ -462,32 +467,34 @@ async fn main() {
         sleep(Duration::from_secs(3)).await;
 
         // Request 6: Circuit should be half-open, test succeeds, closes circuit
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
         let call6 = dag.add_task(ProtectedServiceCall::new(
             "ServiceCall-6",
             circuit.clone(),
             false, // success
         ));
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        println!("Result 6: {:?}\n", dag.get(call6).unwrap());
+        println!("Result 6: {:?}\n", output.get(call6).unwrap());
 
         // Request 7: Circuit is now closed again, normal operation
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
         let call7 = dag.add_task(ProtectedServiceCall::new(
             "ServiceCall-7",
             circuit.clone(),
             false, // success
         ));
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        println!("Result 7: {:?}\n", dag.get(call7).unwrap());
+        println!("Result 7: {:?}\n", output.get(call7).unwrap());
     }
 
     println!("=== Circuit Breaker Example Complete ===");
diff --git a/examples/data_pipeline.rs b/examples/data_pipeline.rs
index 67cc0ce..e012fe1 100644
--- a/examples/data_pipeline.rs
+++ b/examples/data_pipeline.rs
@@ -178,7 +178,7 @@ impl GenerateReport {
 
 #[tokio::main]
 async fn main() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     println!("Building data pipeline DAG...\n");
 
@@ -220,10 +220,11 @@ async fn main() {
 
     // Execute the pipeline
     println!("Executing pipeline...\n");
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
     // Print final report
-    println!("{}", dag.get(report).unwrap());
+    println!("{}", output.get(report).unwrap());
 }
diff --git a/examples/error_handling.rs b/examples/error_handling.rs
index c4dde6b..67cea84 100644
--- a/examples/error_handling.rs
+++ b/examples/error_handling.rs
@@ -209,7 +209,7 @@ async fn main() {
     // Example 1: Successful path
     println!("Example 1: Valid input");
     {
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         let input = dag.add_task(StringSource("42"));
 
@@ -222,11 +222,12 @@ async fn main() {
         let processed = dag.add_task(ProcessData).depends_on(validated);
         let logged = dag.add_task(LogError).depends_on(processed);
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        match dag.get(logged).unwrap().as_ref() {
+        match output.get(logged).unwrap().as_ref() {
             Ok(result) => println!("Final result: {}\n", result),
             Err(e) => eprintln!("Final error: {}\n", e),
         }
@@ -235,7 +236,7 @@ async fn main() {
     // Example 2: Parse error
     println!("Example 2: Invalid input (parse error)");
     {
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         let input = dag.add_task(StringSource("not a number"));
 
@@ -248,11 +249,12 @@ async fn main() {
         let processed = dag.add_task(ProcessData).depends_on(validated);
         let logged = dag.add_task(LogError).depends_on(processed);
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        match dag.get(logged).unwrap().as_ref() {
+        match output.get(logged).unwrap().as_ref() {
             Ok(result) => println!("Final result: {}\n", result),
             Err(e) => eprintln!("Final error: {}\n", e),
         }
@@ -261,7 +263,7 @@ async fn main() {
     // Example 3: Validation error
     println!("Example 3: Out of range");
     {
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         let input = dag.add_task(StringSource("150"));
 
@@ -274,11 +276,12 @@ async fn main() {
         let processed = dag.add_task(ProcessData).depends_on(validated);
         let logged = dag.add_task(LogError).depends_on(processed);
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        match dag.get(logged).unwrap().as_ref() {
+        match output.get(logged).unwrap().as_ref() {
             Ok(result) => println!("Final result: {}\n", result),
             Err(e) => eprintln!("Final error: {}\n", e),
         }
@@ -287,7 +290,7 @@ async fn main() {
     // Example 4: Error recovery with fallback
     println!("Example 4: Error recovery with fallback");
     {
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         let input = dag.add_task(StringSource("invalid"));
 
@@ -297,11 +300,12 @@ async fn main() {
             .add_task(WithFallback { fallback: 0 })
             .depends_on(parsed);
 
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+        let mut output = dag
+            .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
             .await
             .unwrap();
 
-        let result = dag.get(with_fallback).unwrap();
+        let result = output.get(with_fallback).unwrap();
         println!("Recovered with fallback: {}\n", result);
     }
 
diff --git a/scripts/release_check.sh b/scripts/release_check.sh
index bc5725f..e90afbc 100755
--- a/scripts/release_check.sh
+++ b/scripts/release_check.sh
@@ -157,14 +157,14 @@ section "TESTING"
 
 echo "→ Running library unit tests..."
 cargo test --lib || check_fail "Library unit tests failed"
-check_pass "Library unit tests pass (70 tests)"
+check_pass "Library unit tests pass"
 
 echo ""
 echo "→ Running integration tests with all feature combinations..."
 cargo test --test '*' --no-default-features || check_fail "Integration tests failed (no features)"
 cargo test --test '*' --all-features || check_fail "Integration tests failed (all features)"
 cargo test --test '*' || check_fail "Integration tests failed (default features)"
-check_pass "Integration tests pass across all feature combinations (280+ tests total)"
+check_pass "Integration tests pass across all feature combinations"
 
 echo ""
 echo "→ Running doc tests with all features..."
diff --git a/src/error/tests.rs b/src/error/tests.rs
index 4e3d87a..3e9d680 100644
--- a/src/error/tests.rs
+++ b/src/error/tests.rs
@@ -51,9 +51,7 @@ fn test_dag_error_display_result_not_found() {
 
     assert!(display.contains("Result not found"));
     assert!(display.contains("task #7"));
-    assert!(display.contains(
-        "Calllet mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })"
-    ));
+    assert!(display.contains("Call dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })"));
 }
 
 #[test]
diff --git a/src/lib.rs b/src/lib.rs
index 06294d4..d7b9a03 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -541,11 +541,12 @@
 //! # let mut dag = DagRunner::new();
 //! # let node = dag.add_task(Value(42));
 //! // Simple approach with .unwrap()
-//!let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap();
+//! let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap();
 //! let result = output.get(node).unwrap();
+//! # let mut dag = DagRunner::new();
 //!
 //! // Or handle errors explicitly
-//! matchlet mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await {
+//! match dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await {
 //!     Ok(_) => println!("DAG executed successfully"),
 //!     Err(e) => eprintln!("DAG execution failed: {}", e),
 //! }
diff --git a/tests/boundaries/deep_chains.rs b/tests/boundaries/deep_chains.rs
index a6bb29e..c38a008 100644
--- a/tests/boundaries/deep_chains.rs
+++ b/tests/boundaries/deep_chains.rs
@@ -9,13 +9,13 @@ use std::sync::Arc;
 #[tokio::test]
 async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
     // Test 10 parallel chains of depth 500 each
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let mut chain_ends = Vec::new();
 
     for chain_id in 0..8 {
         let first = dag.add_task(task_fn::<(), _, _>(move |_: ()| chain_id * 1000));
-        let mut current: dagx::TaskHandle<i32> = first.into();
+        let mut current: dagx::TaskHandle<i32> = first;
 
         for _ in 0..500 {
             current = dag
@@ -51,7 +51,8 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
             &chain_ends[7],
         ));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Each chain: start + 500
@@ -59,7 +60,7 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
     // Chain 1: 1000 + 500 = 1500
     // ...
     // Sum = 500*8 + (0+1000+2000+...+7000) = 4000 + 28000 = 32000
-    assert_eq!(dag.get(final_task)?, 32000);
+    assert_eq!(output.get(final_task)?, 32000);
 
     Ok(())
 }
@@ -67,7 +68,7 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
 #[tokio::test]
 async fn test_deep_chain_execution_order() -> DagResult<()> {
     // Verify tasks in a deep chain execute in correct order
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let order = Arc::new(AtomicUsize::new(0));
 
     let first = dag.add_task(task_fn::<(), _, _>({
@@ -78,7 +79,7 @@ async fn test_deep_chain_execution_order() -> DagResult<()> {
             0
         }
     }));
-    let mut current: dagx::TaskHandle<i32> = first.into();
+    let mut current: dagx::TaskHandle<i32> = first;
 
     for expected in 1..100 {
         current = dag
@@ -93,11 +94,12 @@ async fn test_deep_chain_execution_order() -> DagResult<()> {
             .depends_on(current);
     }
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     assert_eq!(order.load(Ordering::SeqCst), 100);
-    assert_eq!(dag.get(current)?, 99);
+    assert_eq!(output.get(current)?, 99);
 
     Ok(())
 }
@@ -105,12 +107,12 @@ async fn test_deep_chain_execution_order() -> DagResult<()> {
 #[tokio::test]
 async fn test_deep_binary_tree() -> DagResult<()> {
     // Test a deep binary tree structure (depth 10 = 1023 nodes)
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
-    fn build_tree(dag: &DagRunner, depth: usize, value: i32) -> dagx::TaskHandle<i32> {
+    fn build_tree(dag: &mut DagRunner, depth: usize, value: i32) -> dagx::TaskHandle<i32> {
         if depth == 0 {
             let task = dag.add_task(task_fn::<(), _, _>(move |_: ()| value));
-            task.into() // Convert TaskBuilder to TaskHandle
+            task // Convert TaskBuilder to TaskHandle
         } else {
             let left = build_tree(dag, depth - 1, value * 2);
             let right = build_tree(dag, depth - 1, value * 2 + 1);
@@ -122,14 +124,15 @@ async fn test_deep_binary_tree() -> DagResult<()> {
         }
     }
 
-    let root = build_tree(&dag, 10, 1);
+    let root = build_tree(&mut dag, 10, 1);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Sum of all values in a complete binary tree with values 1024..2047
     // This is sum from 2^10 to 2^11-1 = 1024 * 1023 / 2 + 1024 * 512 = 1047552
-    let result = dag.get(root)?;
+    let result = output.get(root)?;
     assert!(result > 0);
 
     Ok(())
@@ -138,13 +141,13 @@ async fn test_deep_binary_tree() -> DagResult<()> {
 #[tokio::test]
 async fn test_fibonacci_chain() -> DagResult<()> {
     // Test a Fibonacci-like dependency pattern
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let f0_builder = dag.add_task(task_fn::<(), _, _>(|_: ()| 0i64));
     let f1_builder = dag.add_task(task_fn::<(), _, _>(|_: ()| 1i64));
 
-    let mut prev2: dagx::TaskHandle<i64> = f0_builder.into();
-    let mut prev1: dagx::TaskHandle<i64> = f1_builder.into();
+    let mut prev2: dagx::TaskHandle<i64> = f0_builder;
+    let mut prev1: dagx::TaskHandle<i64> = f1_builder;
 
     for _ in 2..50 {
         let next = dag
@@ -155,11 +158,12 @@ async fn test_fibonacci_chain() -> DagResult<()> {
         prev1 = next;
     }
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // 50th Fibonacci number (0-indexed) = 7778742049
-    assert_eq!(dag.get(prev1)?, 7778742049);
+    assert_eq!(output.get(prev1)?, 7778742049);
 
     Ok(())
 }
@@ -167,13 +171,13 @@ async fn test_fibonacci_chain() -> DagResult<()> {
 #[tokio::test]
 async fn test_zigzag_dependencies() -> DagResult<()> {
     // Test a zigzag pattern: two chains that cross-reference each other
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let a_first = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
     let b_first = dag.add_task(task_fn::<(), _, _>(|_: ()| 2));
 
-    let mut chain_a: dagx::TaskHandle<i32> = a_first.into();
-    let mut chain_b: dagx::TaskHandle<i32> = b_first.into();
+    let mut chain_a: dagx::TaskHandle<i32> = a_first;
+    let mut chain_b: dagx::TaskHandle<i32> = b_first;
 
     for i in 0..100 {
         if i % 2 == 0 {
@@ -194,10 +198,11 @@ async fn test_zigzag_dependencies() -> DagResult<()> {
         .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b))
         .depends_on((&chain_a, &chain_b));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    let result = dag.get(final_task)?;
+    let result = output.get(final_task)?;
     assert!(result > 100);
 
     Ok(())
diff --git a/tests/boundaries/type_limits.rs b/tests/boundaries/type_limits.rs
index 79a3f21..11d1e30 100644
--- a/tests/boundaries/type_limits.rs
+++ b/tests/boundaries/type_limits.rs
@@ -1,6 +1,6 @@
 //! Tests for type system edge cases and limits
 
-use dagx::{DagResult, DagRunner, TaskHandle};
+use dagx::{DagResult, DagRunner};
 use dagx_test::task_fn;
 
 use std::marker::PhantomData;
@@ -16,13 +16,13 @@ async fn test_zero_sized_types_throughout() -> DagResult<()> {
         _phantom: PhantomData<T>,
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Task producing ZST
     let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| Empty));
 
     // Task that also produces ZST (independent since we can't extract custom types)
-    let t2: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)).into();
+    let t2 = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
 
     // Task with PhantomData
     let _t3 = dag.add_task(task_fn::<(), _, _>(|_: ()| PhantomWrapper::<i32> {
@@ -37,12 +37,13 @@ async fn test_zero_sized_types_throughout() -> DagResult<()> {
         }))
         .depends_on(t2);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(t1)?, Empty);
-    assert_eq!(dag.get(t2)?, 42);
-    assert_eq!(dag.get(t4)?, "success");
+    assert_eq!(output.get(t1)?, Empty);
+    assert_eq!(output.get(t2)?, 42);
+    assert_eq!(output.get(t4)?, "success");
 
     Ok(())
 }
@@ -56,7 +57,7 @@ async fn test_large_value_types() -> DagResult<()> {
         more_data: Vec<u64>,
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| {
         let large = LargeStruct {
@@ -73,37 +74,11 @@ async fn test_large_value_types() -> DagResult<()> {
         }))
         .depends_on(t1);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(t2)?, 1042);
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_unit_type_chains() -> DagResult<()> {
-    // Test chains of unit type operations
-    let dag = DagRunner::new();
-
-    let t1: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| {})).into();
-    let t2 = dag
-        .add_task(task_fn::<(), _, _>(|_: ()| {}))
-        .depends_on((t1,));
-    let t3 = dag
-        .add_task(task_fn::<(), _, _>(|_: ()| {}))
-        .depends_on((t2,));
-    let t4 = dag
-        .add_task(task_fn::<(), _, _>(|_: ()| "done"))
-        .depends_on((t3,));
-
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await?;
-
-    dag.get(t1)?; // Verify no error
-    dag.get(t2)?;
-    dag.get(t3)?;
-    assert_eq!(dag.get(t4)?, "done");
+    assert_eq!(output.get(t2)?, 1042);
 
     Ok(())
 }
@@ -113,7 +88,7 @@ async fn test_reference_wrapper_types() -> DagResult<()> {
     // Test types that wrap references
     use std::sync::{Arc, Mutex};
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| {
         let arc = Arc::new(Mutex::new(100));
@@ -126,10 +101,11 @@ async fn test_reference_wrapper_types() -> DagResult<()> {
         .add_task(task_fn::<i32, _, _>(|&val: &i32| val * 2))
         .depends_on(t1);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(t2)?, 200);
+    assert_eq!(output.get(t2)?, 200);
 
     Ok(())
 }
diff --git a/tests/boundaries/zero_tasks.rs b/tests/boundaries/zero_tasks.rs
index 3548a63..b9c2100 100644
--- a/tests/boundaries/zero_tasks.rs
+++ b/tests/boundaries/zero_tasks.rs
@@ -8,17 +8,18 @@ use tokio::task::yield_now;
 #[tokio::test]
 async fn test_dag_with_only_source_tasks() -> DagResult<()> {
     // Test a DAG with multiple source tasks but no dependencies
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let tasks: Vec<_> = (0..100)
         .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
         .collect();
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     for (i, task) in tasks.into_iter().enumerate() {
-        assert_eq!(dag.get(task)?, i);
+        assert_eq!(output.get(task)?, i);
     }
 
     Ok(())
@@ -38,20 +39,21 @@ async fn test_single_self_contained_stateful_task() -> DagResult<()> {
         }
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let task = dag.add_task(Counter { value: 21 });
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(task)?, 42);
+    assert_eq!(output.get(task)?, 42);
     Ok(())
 }
 
 #[tokio::test]
 async fn test_minimal_task_with_minimal_async_work() -> DagResult<()> {
     // Test a task that does the absolute minimum async work
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     struct YieldTask;
 
@@ -65,9 +67,10 @@ async fn test_minimal_task_with_minimal_async_work() -> DagResult<()> {
 
     let task = dag.add_task(YieldTask);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(task)?, 1337);
+    assert_eq!(output.get(task)?, 1337);
     Ok(())
 }
diff --git a/tests/custom_type_test.rs b/tests/custom_type_test.rs
index f7d5374..607c4cc 100644
--- a/tests/custom_type_test.rs
+++ b/tests/custom_type_test.rs
@@ -1,5 +1,5 @@
 // Test that custom types work with the #[task] macro
-use dagx::{task, DagRunner, TaskHandle};
+use dagx::{task, DagRunner};
 
 #[derive(Clone, Debug, PartialEq)]
 struct Person {
@@ -51,33 +51,35 @@ impl PersonToCompany {
 
 #[tokio::test]
 async fn test_custom_type_single_dependency() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let fetch = dag.add_task(FetchPerson);
     let process = dag.add_task(ProcessPerson).depends_on(fetch);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
-    let result = dag.get(process).unwrap();
+    let result = output.get(process).unwrap();
     assert_eq!(result, "Alice is 30 years old");
 }
 
 #[tokio::test]
 async fn test_custom_type_passthrough() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
-    let fetch: TaskHandle<_> = dag.add_task(FetchPerson).into();
+    let fetch = dag.add_task(FetchPerson);
     let transform = dag.add_task(PersonToCompany).depends_on(fetch);
     let process = dag.add_task(ProcessPerson).depends_on(fetch);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
-    let person_result = dag.get(process).unwrap();
-    let company_result = dag.get(transform).unwrap();
+    let person_result = output.get(process).unwrap();
+    let company_result = output.get(transform).unwrap();
 
     assert_eq!(person_result.as_str(), "Alice is 30 years old");
     assert_eq!(company_result.name, "Alice's Company");
@@ -86,20 +88,21 @@ async fn test_custom_type_passthrough() {
 
 #[tokio::test]
 async fn test_custom_type_fan_out() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
-    let fetch: TaskHandle<_> = dag.add_task(FetchPerson).into();
+    let fetch = dag.add_task(FetchPerson);
     let process1 = dag.add_task(ProcessPerson).depends_on(fetch);
     let process2 = dag.add_task(ProcessPerson).depends_on(fetch);
     let process3 = dag.add_task(ProcessPerson).depends_on(fetch);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
-    let result1 = dag.get(process1).unwrap();
-    let result2 = dag.get(process2).unwrap();
-    let result3 = dag.get(process3).unwrap();
+    let result1 = output.get(process1).unwrap();
+    let result2 = output.get(process2).unwrap();
+    let result3 = output.get(process3).unwrap();
 
     assert_eq!(result1, "Alice is 30 years old");
     assert_eq!(result2, "Alice is 30 years old");
diff --git a/tests/cycle_prevention.rs b/tests/cycle_prevention.rs
index 7365115..1c6e6ff 100644
--- a/tests/cycle_prevention.rs
+++ b/tests/cycle_prevention.rs
@@ -3,7 +3,7 @@
 //! These tests show that even if you try to create cycles, the Rust compiler
 //! prevents it through the type-state pattern.
 
-use dagx::{task, DagRunner, TaskHandle};
+use dagx::{task, DagRunner};
 
 /// Demonstrates that TaskBuilder is consumed by depends_on,
 /// preventing you from wiring the same task multiple times
@@ -25,7 +25,7 @@ fn test_builder_consumed_by_depends_on() {
         }
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let source = dag.add_task(Source);
     let consumer_builder = dag.add_task(Consumer);
 
@@ -49,11 +49,11 @@ fn test_handle_is_immutable_reference() {
         }
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let builder = dag.add_task(NoInput);
 
     // Convert to handle (only works because Input = ())
-    let handle: TaskHandle<_> = builder.into();
+    let handle = builder;
 
     // TaskHandle is Copy/Clone - just an ID wrapper
     let _copy1 = handle;
@@ -85,7 +85,7 @@ fn test_ordering_prevents_cycles() {
         }
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let a_builder = dag.add_task(TaskA);
     let b_builder = dag.add_task(TaskB);
 
@@ -107,7 +107,7 @@ fn test_ordering_prevents_cycles() {
     // So we can only create DAGs, never cycles!
 
     // Let's create a valid DAG instead: A→B (B depends on A)
-    let a_handle: TaskHandle<_> = a_builder.into(); // A has no deps (unit input)
+    let a_handle = a_builder; // A has no deps (unit input)
     let _b_handle = b_builder.depends_on(a_handle);
     // Now we could try: a_handle.depends_on(b_handle)?
     // But TaskHandle has no such method!
@@ -132,13 +132,13 @@ fn test_unit_tasks_cannot_create_cycles() {
         }
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let task1 = dag.add_task(Task1);
     let task2 = dag.add_task(Task2);
 
     // Both have unit input, so we can convert to handles
-    let _handle1: TaskHandle<_> = task1.into();
-    let _handle2: TaskHandle<_> = task2.into();
+    let _handle1 = task1;
+    let _handle2 = task2;
 
     // Now we have two handles, but...
     // TaskHandle has no methods to add dependencies!
@@ -176,7 +176,7 @@ fn test_three_way_cycle_impossible() {
         }
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let _a_builder = dag.add_task(A);
     let _b_builder = dag.add_task(B);
     let _c_builder = dag.add_task(C);
@@ -205,7 +205,7 @@ fn test_self_loop_prevented() {
         }
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let _builder = dag.add_task(SelfReferential);
 
     // To create A→A (self-loop):
@@ -256,7 +256,7 @@ async fn test_type_safety_prevents_invalid_graphs() {
         }
     }
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Build a valid DAG: two sources, two transforms, one aggregator
     let s1 = dag.add_task(Source);
@@ -270,11 +270,12 @@ async fn test_type_safety_prevents_invalid_graphs() {
     let result = dag.add_task(Aggregate).depends_on((&t1, &t2));
 
     // Execute and verify
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
-    assert_eq!(dag.get(result).unwrap(), 40); // (10*2) + (10*2)
+    assert_eq!(output.get(result).unwrap(), 40); // (10*2) + (10*2)
 
     // Now if we wanted to add a cycle, we'd need to:
     // - Make one of the Source tasks depend on the Aggregate result
diff --git a/tests/dependencies/resolution.rs b/tests/dependencies/resolution.rs
index 31ac727..4aecc5c 100644
--- a/tests/dependencies/resolution.rs
+++ b/tests/dependencies/resolution.rs
@@ -8,7 +8,7 @@ use std::sync::{Arc, Mutex};
 #[tokio::test]
 async fn test_dependency_resolution_order() -> DagResult<()> {
     // Test that dependencies are resolved in the correct order
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let execution_log = Arc::new(Mutex::new(Vec::new()));
 
@@ -79,10 +79,11 @@ async fn test_dependency_resolution_order() -> DagResult<()> {
         .depends_on(d)
     };
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(e)?, 5);
+    assert_eq!(output.get(e)?, 5);
 
     // Verify execution order
     let log = execution_log.lock().unwrap();
diff --git a/tests/dependencies/tuples.rs b/tests/dependencies/tuples.rs
index aafbd3a..51b6385 100644
--- a/tests/dependencies/tuples.rs
+++ b/tests/dependencies/tuples.rs
@@ -5,7 +5,7 @@ use dagx_test::task_fn;
 
 #[tokio::test]
 async fn test_two_dependencies() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
     let b = dag.add_task(task_fn::<(), _, _>(|_: ()| 20));
@@ -13,16 +13,17 @@ async fn test_two_dependencies() -> DagResult<()> {
         .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y))
         .depends_on((a, b));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(sum)?, 30);
+    assert_eq!(output.get(sum)?, 30);
     Ok(())
 }
 
 #[tokio::test]
 async fn test_three_dependencies() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
     let b = dag.add_task(task_fn::<(), _, _>(|_: ()| 2));
@@ -33,19 +34,20 @@ async fn test_three_dependencies() -> DagResult<()> {
         ))
         .depends_on((a, b, c));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(sum)?, 6);
+    assert_eq!(output.get(sum)?, 6);
     Ok(())
 }
 
 #[tokio::test]
 async fn test_four_dependencies() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let deps: Vec<_> = (1..=4)
-        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
+        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
         .collect();
 
     let sum = dag
@@ -54,19 +56,20 @@ async fn test_four_dependencies() -> DagResult<()> {
         ))
         .depends_on((&deps[0], &deps[1], &deps[2], &deps[3]));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(sum)?, 10); // 1+2+3+4
+    assert_eq!(output.get(sum)?, 10); // 1+2+3+4
     Ok(())
 }
 
 #[tokio::test]
 async fn test_five_dependencies() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let deps: Vec<_> = (1..=5)
-        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
+        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
         .collect();
 
     let sum = dag
@@ -75,19 +78,20 @@ async fn test_five_dependencies() -> DagResult<()> {
         ))
         .depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4]));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(sum)?, 15); // 1+2+3+4+5
+    assert_eq!(output.get(sum)?, 15); // 1+2+3+4+5
     Ok(())
 }
 
 #[tokio::test]
 async fn test_six_dependencies() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let deps: Vec<_> = (1..=6)
-        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
+        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
         .collect();
 
     let sum = dag
@@ -96,19 +100,20 @@ async fn test_six_dependencies() -> DagResult<()> {
         ))
         .depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5]));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(sum)?, 21); // 1+2+3+4+5+6
+    assert_eq!(output.get(sum)?, 21); // 1+2+3+4+5+6
     Ok(())
 }
 
 #[tokio::test]
 async fn test_seven_dependencies() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let deps: Vec<_> = (1..=7)
-        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
+        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
         .collect();
 
     let sum = dag
@@ -121,19 +126,20 @@ async fn test_seven_dependencies() -> DagResult<()> {
             &deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6],
         ));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(sum)?, 28); // 1+2+3+4+5+6+7
+    assert_eq!(output.get(sum)?, 28); // 1+2+3+4+5+6+7
     Ok(())
 }
 
 #[tokio::test]
 async fn test_eight_dependencies() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let deps: Vec<_> = (1..=8)
-        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
+        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
         .collect();
 
     let sum = dag
@@ -146,9 +152,10 @@ async fn test_eight_dependencies() -> DagResult<()> {
             &deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6], &deps[7],
         ));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(sum)?, 36); // Sum of 1..=8
+    assert_eq!(output.get(sum)?, 36); // Sum of 1..=8
     Ok(())
 }
diff --git a/tests/errors/propagation.rs b/tests/errors/propagation.rs
index 44501c9..abd0459 100644
--- a/tests/errors/propagation.rs
+++ b/tests/errors/propagation.rs
@@ -1,17 +1,17 @@
 //! Tests for error propagation through the DAG
 
-use dagx::{task, DagRunner, TaskHandle};
+use dagx::{task, DagError, DagRunner};
 use dagx_test::task_fn;
 
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::time::sleep;
 
 #[tokio::test]
 async fn test_linear_error_propagation() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
-    let t1: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 1)).into();
+    let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
     let t2 = dag
         .add_task(task_fn::<i32, _, _>(|&x: &i32| -> i32 {
             panic!("Error at t2 with input {}", x);
@@ -20,28 +20,19 @@ async fn test_linear_error_propagation() {
     let t3 = dag
         .add_task(task_fn::<i32, _, _>(|&x: &i32| x + 1))
         .depends_on(t2);
-    let t4 = dag
-        .add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
+    dag.add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
         .depends_on(t3);
 
-    let _ = dag
-        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await;
-
-    // t1 succeeds
-    assert_eq!(dag.get(t1).unwrap(), 1);
-
-    // t2 fails
-    assert!(dag.get(t2).is_err());
-
-    // t3 and t4 fail due to dependency
-    assert!(dag.get(t3).is_err());
-    assert!(dag.get(t4).is_err());
+    assert!(matches!(
+        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+            .await,
+        Err(DagError::TaskPanicked { .. })
+    ));
 }
 
 #[tokio::test]
 async fn test_diamond_error_propagation() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     //     source
     //     /    \
@@ -49,7 +40,7 @@ async fn test_diamond_error_propagation() {
     //     \    /
     //      sink
 
-    let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 100)).into();
+    let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 100));
 
     let left = dag
         .add_task(task_fn::<i32, _, _>(|&x: &i32| x / 2))
@@ -61,61 +52,45 @@ async fn test_diamond_error_propagation() {
         }))
         .depends_on(source);
 
-    let sink = dag
-        .add_task(task_fn::<(i32, i32), _, _>(|(l, r): (&i32, &i32)| l + r))
+    dag.add_task(task_fn::<(i32, i32), _, _>(|(l, r): (&i32, &i32)| l + r))
         .depends_on((&left, &right));
 
-    let _ = dag
-        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await;
-
-    assert_eq!(dag.get(source).unwrap(), 100);
-    assert_eq!(dag.get(left).unwrap(), 50);
-    assert!(dag.get(right).is_err());
-    assert!(dag.get(sink).is_err()); // Fails due to right dependency
+    assert!(matches!(
+        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+            .await,
+        Err(DagError::TaskPanicked { .. })
+    ));
 }
 
 #[tokio::test]
 async fn test_error_in_wide_fanout() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
-    let source: TaskHandle<_> = dag
-        .add_task(task_fn::<(), _, _>(|_: ()| -> i32 {
-            panic!("Source fails");
-        }))
-        .into();
+    let source = dag.add_task(task_fn::<(), _, _>(|_: ()| -> i32 {
+        panic!("Source fails");
+    }));
 
     // Create wide fanout from failing source
-    let dependents: Vec<_> = (0..20)
-        .map(|i| {
-            dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
-                .depends_on(source)
-        })
-        .collect();
-
-    let _ = dag
-        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await;
-
-    // Source fails
-    assert!(dag.get(source).is_err());
-
-    // All dependents fail
-    for dependent in &dependents {
-        assert!(dag.get(dependent).is_err());
+    for i in 0..20 {
+        dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
+            .depends_on(source);
     }
+
+    assert!(matches!(
+        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+            .await,
+        Err(DagError::TaskPanicked { .. })
+    ));
 }
 
 #[tokio::test]
 async fn test_selective_error_propagation() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Source that produces a Result
-    let source: TaskHandle<_> = dag
-        .add_task(task_fn::<(), _, _>(|_: ()| {
-            Result::<i32, String>::Err("Source error".to_string())
-        }))
-        .into();
+    let source = dag.add_task(task_fn::<(), _, _>(|_: ()| {
+        Result::<i32, String>::Err("Source error".to_string())
+    }));
 
     // Handler that processes the Result
     let handler = dag
@@ -140,34 +115,35 @@ async fn test_selective_error_propagation() {
         }))
         .depends_on(handler);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
     // Source returns an Err variant (not a panic)
-    assert_eq!(dag.get(source).unwrap(), Err("Source error".to_string()));
+    assert_eq!(output.get(source).unwrap(), Err("Source error".to_string()));
 
     // Handler processes it
-    assert_eq!(dag.get(handler).unwrap(), -1);
+    assert_eq!(output.get(handler).unwrap(), -1);
 
     // Final task handles the error case
-    assert_eq!(dag.get(final_task).unwrap(), "Handled error case");
+    assert_eq!(output.get(final_task).unwrap(), "Handled error case");
 }
 
 #[tokio::test]
 async fn test_error_propagation_timing() {
-    let dag = DagRunner::new();
-    let propagation_order = Arc::new(parking_lot::Mutex::new(Vec::new()));
+    let mut dag = DagRunner::new();
+    let propagation_order = Arc::new(Mutex::new(Vec::new()));
 
     struct T1Task {
-        order: Arc<parking_lot::Mutex<Vec<&'static str>>>,
+        order: Arc<Mutex<Vec<&'static str>>>,
     }
 
     #[task]
     impl T1Task {
         async fn run(&self) -> i32 {
             sleep(Duration::from_millis(10)).await;
-            self.order.lock().push("t1_complete");
+            self.order.lock().unwrap().push("t1_complete");
             1
         }
     }
@@ -177,13 +153,13 @@ async fn test_error_propagation_timing() {
     });
 
     struct T2Task {
-        order: Arc<parking_lot::Mutex<Vec<&'static str>>>,
+        order: Arc<Mutex<Vec<&'static str>>>,
     }
 
     #[task]
     impl T2Task {
         async fn run(&self, x: &i32) -> i32 {
-            self.order.lock().push("t2_start");
+            self.order.lock().unwrap().push("t2_start");
             sleep(Duration::from_millis(10)).await;
             panic!("t2 fails with {x}");
         }
@@ -200,7 +176,7 @@ async fn test_error_propagation_timing() {
             let order = propagation_order.clone();
             move |x: &i32| {
                 let order = order.clone();
-                order.lock().push("t3_should_not_run");
+                order.lock().unwrap().push("t3_should_not_run");
                 x + 1
             }
         }))
@@ -210,7 +186,7 @@ async fn test_error_propagation_timing() {
         .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await;
 
-    let order = propagation_order.lock().clone();
+    let order = propagation_order.lock().unwrap().clone();
 
     // t1 completes, t2 starts and fails, t3 never runs
     assert!(order.contains(&"t1_complete"));
diff --git a/tests/errors/recovery.rs b/tests/errors/recovery.rs
index e1dc5b9..ef96102 100644
--- a/tests/errors/recovery.rs
+++ b/tests/errors/recovery.rs
@@ -4,11 +4,11 @@ use dagx::{DagResult, DagRunner};
 use dagx_test::task_fn;
 
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 #[tokio::test]
 async fn test_retry_pattern_simulation() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let attempt_counter = Arc::new(AtomicUsize::new(0));
 
     // Task that would fail on first attempts (simulated)
@@ -29,18 +29,19 @@ async fn test_retry_pattern_simulation() -> DagResult<()> {
         }
     }));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Task completes with error on first "attempt"
-    assert_eq!(dag.get(task)?, Err("Simulated failure"));
+    assert_eq!(output.get(task)?, Err("Simulated failure"));
 
     Ok(())
 }
 
 #[tokio::test]
 async fn test_circuit_breaker_pattern() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let failure_count = Arc::new(AtomicUsize::new(0));
     let circuit_open = Arc::new(AtomicBool::new(false));
 
@@ -71,12 +72,13 @@ async fn test_circuit_breaker_pattern() -> DagResult<()> {
         })
         .collect();
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // First few tasks fail and open circuit
     for task in tasks.into_iter().take(4) {
-        assert!(dag.get(task)?.is_err());
+        assert!(output.get(task)?.is_err());
     }
 
     Ok(())
@@ -84,8 +86,8 @@ async fn test_circuit_breaker_pattern() -> DagResult<()> {
 
 #[tokio::test]
 async fn test_error_accumulation_pattern() -> DagResult<()> {
-    let dag = DagRunner::new();
-    let errors = Arc::new(parking_lot::Mutex::new(Vec::new()));
+    let mut dag = DagRunner::new();
+    let errors = Arc::new(Mutex::new(Vec::new()));
 
     // Tasks that may produce errors
     let mut tasks: Vec<_> = (0..10)
@@ -94,7 +96,10 @@ async fn test_error_accumulation_pattern() -> DagResult<()> {
             dag.add_task(task_fn::<(), _, _>(move |_: ()| {
                 let errors = errors.clone();
                 if i % 2 == 0 {
-                    errors.lock().push(format!("Error from task {}", i));
+                    errors
+                        .lock()
+                        .unwrap()
+                        .push(format!("Error from task {}", i));
                     return Err(format!("Task {} error", i));
                 }
                 Ok(i)
@@ -102,11 +107,12 @@ async fn test_error_accumulation_pattern() -> DagResult<()> {
         })
         .collect();
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Check accumulated errors
-    let error_list = errors.lock().clone();
+    let error_list = errors.lock().unwrap().clone();
     assert_eq!(error_list.len(), 5); // Even numbered tasks
 
     tasks.truncate(2);
@@ -114,15 +120,15 @@ async fn test_error_accumulation_pattern() -> DagResult<()> {
     let [first, second] = <[_; 2]>::try_from(tasks).map_err(|_e| ()).unwrap();
 
     // Verify specific results
-    assert!(dag.get(first)?.is_err());
-    assert_eq!(dag.get(second)?, Ok(1));
+    assert!(output.get(first)?.is_err());
+    assert_eq!(output.get(second)?, Ok(1));
 
     Ok(())
 }
 
 #[tokio::test]
 async fn test_error_boundary_isolation() -> DagResult<()> {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Group A: contains an error - use Result
     let a1 = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
@@ -143,15 +149,16 @@ async fn test_error_boundary_isolation() -> DagResult<()> {
         .add_task(task_fn::<i32, _, _>(|&x: &i32| x + 5))
         .depends_on(b2);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Group A fails (returns Err)
-    assert_eq!(dag.get(a2)?, Err("Group A error"));
+    assert_eq!(output.get(a2)?, Err("Group A error"));
 
     // Groups B and C succeed
-    assert_eq!(dag.get(b2)?, 40);
-    assert_eq!(dag.get(c1)?, 45);
+    assert_eq!(output.get(b2)?, 40);
+    assert_eq!(output.get(c1)?, 45);
 
     Ok(())
 }
diff --git a/tests/execution/mod.rs b/tests/execution/mod.rs
index 9f29dd3..88bbb12 100644
--- a/tests/execution/mod.rs
+++ b/tests/execution/mod.rs
@@ -6,43 +6,46 @@ use dagx::*;
 #[tokio::test]
 async fn test_empty_dag() {
     let dag = DagRunner::new();
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let _ = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap(); // Should complete without error
 }
 
 #[tokio::test]
 async fn test_single_task() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
-    assert_eq!(dag.get(task).unwrap(), 42);
+    assert_eq!(output.get(task).unwrap(), 42);
 }
 
 #[tokio::test]
 async fn test_wide_parallel() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let tasks: Vec<_> = (0..10)
         .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 2)))
         .collect();
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
     for (i, task) in tasks.into_iter().enumerate() {
-        assert_eq!(dag.get(task).unwrap(), i * 2);
+        assert_eq!(output.get(task).unwrap(), i * 2);
     }
 }
 
 #[tokio::test]
 async fn test_diamond_dependency() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
-    let a: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)).into();
+    let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
     let b = dag
         .add_task(task_fn::<i32, _, _>(|&x: &i32| x + 5))
         .depends_on(a);
@@ -53,13 +56,14 @@ async fn test_diamond_dependency() {
         .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y))
         .depends_on((&b, &c));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
     // Note: a, b, and c are not sinks (they have dependents)
     // Only d is a sink and can be retrieved
-    assert_eq!(dag.get(d).unwrap(), 35); // 15 + 20
+    assert_eq!(output.get(d).unwrap(), 35); // 15 + 20
 }
 
 #[derive(Clone)]
@@ -70,7 +74,7 @@ struct User {
 
 #[tokio::test]
 async fn test_different_output_types() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let string_task = dag.add_task(task_fn::<(), _, _>(|_: ()| "hello".to_string()));
     let vec_task = dag.add_task(task_fn::<(), _, _>(|_: ()| vec![1, 2, 3]));
@@ -79,22 +83,23 @@ async fn test_different_output_types() {
         age: 30,
     }));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
-    assert_eq!(&dag.get(string_task).unwrap(), "hello");
-    assert_eq!(dag.get(vec_task).unwrap(), vec![1, 2, 3]);
-    let user = dag.get(struct_task).unwrap();
+    assert_eq!(&output.get(string_task).unwrap(), "hello");
+    assert_eq!(output.get(vec_task).unwrap(), vec![1, 2, 3]);
+    let user = output.get(struct_task).unwrap();
     assert_eq!(user.name, "Alice");
     assert_eq!(user.age, 30);
 }
 
 #[tokio::test]
 async fn test_multiple_sinks() {
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
-    let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)).into();
+    let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
 
     // Branch 1
     let branch1 = dag
@@ -112,21 +117,22 @@ async fn test_multiple_sinks() {
         .add_task(task_fn::<i32, _, _>(|&x: &i32| x * 3))
         .depends_on(branch2);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap(); // Should wait for both sinks
 
-    assert_eq!(dag.get(sink1).unwrap(), 21); // (10 * 2) + 1
-    assert_eq!(dag.get(sink2).unwrap(), 45); // (10 + 5) * 3
+    assert_eq!(output.get(sink1).unwrap(), 21); // (10 * 2) + 1
+    assert_eq!(output.get(sink2).unwrap(), 45); // (10 + 5) * 3
 }
 
 #[tokio::test]
 async fn test_single_task_inline_path() {
     // Test single-task layer inline execution path (line 367-410 in runner.rs)
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Single task in its own layer
-    let t1: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)).into();
+    let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
     let t2 = dag
         .add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
         .depends_on(t1);
@@ -134,19 +140,20 @@ async fn test_single_task_inline_path() {
         .add_task(task_fn::<i32, _, _>(|&x: &i32| x + 5))
         .depends_on(t2);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
-    assert_eq!(dag.get(t3).unwrap(), 25); // (10 * 2) + 5
+    assert_eq!(output.get(t3).unwrap(), 25); // (10 * 2) + 5
 }
 
 #[tokio::test]
 async fn test_multi_consumer_fanout() {
     // Test fanout with multiple consumers from one producer
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
-    let producer: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)).into();
+    let producer = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
 
     // Multiple consumers
     let c1 = dag
@@ -159,11 +166,12 @@ async fn test_multi_consumer_fanout() {
         .add_task(task_fn::<i32, _, _>(|&x: &i32| x - 5))
         .depends_on(producer);
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
-    assert_eq!(dag.get(c1).unwrap(), 84);
-    assert_eq!(dag.get(c2).unwrap(), 52);
-    assert_eq!(dag.get(c3).unwrap(), 37);
+    assert_eq!(output.get(c1).unwrap(), 84);
+    assert_eq!(output.get(c2).unwrap(), 52);
+    assert_eq!(output.get(c3).unwrap(), 37);
 }
diff --git a/tests/timing/interleaving.rs b/tests/interleaving/mod.rs
similarity index 78%
rename from tests/timing/interleaving.rs
rename to tests/interleaving/mod.rs
index cb430d9..acd84ef 100644
--- a/tests/timing/interleaving.rs
+++ b/tests/interleaving/mod.rs
@@ -4,19 +4,19 @@ use dagx::{task, DagResult, DagRunner};
 use dagx_test::task_fn;
 
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use tokio::sync::Barrier;
 
 #[tokio::test]
 async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
     // Force specific interleaving using barriers
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let barrier = Arc::new(Barrier::new(3));
-    let order = Arc::new(parking_lot::Mutex::new(Vec::new()));
+    let order = Arc::new(Mutex::new(Vec::new()));
 
     struct InterleaveTask {
         barrier: Arc<Barrier>,
-        order: Arc<parking_lot::Mutex<Vec<String>>>,
+        order: Arc<Mutex<Vec<String>>>,
         prefix: &'static str,
         value: i32,
     }
@@ -24,9 +24,15 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
     #[task]
     impl InterleaveTask {
         async fn run(&self) -> i32 {
-            self.order.lock().push(format!("{}_start", self.prefix));
+            self.order
+                .lock()
+                .unwrap()
+                .push(format!("{}_start", self.prefix));
             self.barrier.wait().await;
-            self.order.lock().push(format!("{}_end", self.prefix));
+            self.order
+                .lock()
+                .unwrap()
+                .push(format!("{}_end", self.prefix));
             self.value
         }
     }
@@ -58,13 +64,14 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
         ))
         .depends_on((t1, t2, t3));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(collector)?, 6);
+    assert_eq!(output.get(collector)?, 6);
 
     // Check interleaving pattern
-    let events = order.lock().clone();
+    let events = order.lock().unwrap().clone();
     assert_eq!(events.len(), 6);
 
     // All starts should come before all ends due to barrier
@@ -80,14 +87,13 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
 #[tokio::test]
 async fn test_alternating_execution_pattern() -> DagResult<()> {
     // Create an alternating execution pattern
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let turn = Arc::new(AtomicUsize::new(0));
-    let events: Arc<parking_lot::Mutex<Vec<String>>> =
-        Arc::new(parking_lot::Mutex::new(Vec::new()));
+    let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
 
     struct FirstA {
         turn: Arc<AtomicUsize>,
-        events: Arc<parking_lot::Mutex<Vec<String>>>,
+        events: Arc<Mutex<Vec<String>>>,
     }
 
     #[task]
@@ -96,22 +102,20 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
             while self.turn.load(Ordering::SeqCst) != 0 {
                 tokio::task::yield_now().await;
             }
-            self.events.lock().push("A0".to_string());
+            self.events.lock().unwrap().push("A0".to_string());
             self.turn.store(1, Ordering::SeqCst);
             0
         }
     }
 
-    let first_a = dag.add_task(FirstA {
+    let mut chain_a = dag.add_task(FirstA {
         turn: turn.clone(),
         events: events.clone(),
     });
 
-    let mut chain_a: dagx::TaskHandle<i32> = first_a.into();
-
     struct FirstB {
         turn: Arc<AtomicUsize>,
-        events: Arc<parking_lot::Mutex<Vec<String>>>,
+        events: Arc<Mutex<Vec<String>>>,
     }
 
     #[task]
@@ -120,22 +124,20 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
             while self.turn.load(Ordering::SeqCst) != 1 {
                 tokio::task::yield_now().await;
             }
-            self.events.lock().push("B0".to_string());
+            self.events.lock().unwrap().push("B0".to_string());
             self.turn.store(2, Ordering::SeqCst);
             0
         }
     }
 
-    let first_b = dag.add_task(FirstB {
+    let mut chain_b = dag.add_task(FirstB {
         turn: turn.clone(),
         events: events.clone(),
     });
 
-    let mut chain_b: dagx::TaskHandle<i32> = first_b.into();
-
     struct ChainA {
         turn: Arc<AtomicUsize>,
-        events: Arc<parking_lot::Mutex<Vec<String>>>,
+        events: Arc<Mutex<Vec<String>>>,
         idx: usize,
     }
 
@@ -145,7 +147,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
             while self.turn.load(Ordering::SeqCst) != self.idx * 2 {
                 tokio::task::yield_now().await;
             }
-            self.events.lock().push(format!("A{}", self.idx));
+            self.events.lock().unwrap().push(format!("A{}", self.idx));
             self.turn.store(self.idx * 2 + 1, Ordering::SeqCst);
             x + 1
         }
@@ -153,7 +155,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
 
     struct ChainB {
         turn: Arc<AtomicUsize>,
-        events: Arc<parking_lot::Mutex<Vec<String>>>,
+        events: Arc<Mutex<Vec<String>>>,
         idx: usize,
     }
 
@@ -163,7 +165,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
             while self.turn.load(Ordering::SeqCst) != self.idx * 2 + 1 {
                 tokio::task::yield_now().await;
             }
-            self.events.lock().push(format!("B{}", self.idx));
+            self.events.lock().unwrap().push(format!("B{}", self.idx));
             self.turn.store(self.idx * 2 + 2, Ordering::SeqCst);
             x + 1
         }
@@ -187,11 +189,12 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
             .depends_on(chain_b);
     }
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let _output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Check alternating pattern
-    let event_list = events.lock().clone();
+    let event_list = events.lock().unwrap().clone();
     assert_eq!(
         event_list,
         vec!["A0", "B0", "A1", "B1", "A2", "B2", "A3", "B3", "A4", "B4"]
@@ -203,7 +206,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
 #[tokio::test]
 async fn test_layered_interleaving() -> DagResult<()> {
     // Test interleaving across DAG layers
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let layer_active = Arc::new(AtomicUsize::new(0));
 
     struct Layer1Task {
@@ -228,7 +231,6 @@ async fn test_layered_interleaving() -> DagResult<()> {
                 active: layer_active.clone(),
                 idx: i,
             })
-            .into()
         })
         .collect();
 
@@ -264,11 +266,12 @@ async fn test_layered_interleaving() -> DagResult<()> {
         })
         .depends_on((&layer2[0], &layer2[1]));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Sum of 0+1+2+3 = 6
-    assert_eq!(dag.get(layer3)?, 6);
+    assert_eq!(output.get(layer3)?, 6);
 
     Ok(())
 }
diff --git a/tests/lib_tests.rs b/tests/lib_tests.rs
index 3fecea2..fed5629 100644
--- a/tests/lib_tests.rs
+++ b/tests/lib_tests.rs
@@ -8,7 +8,7 @@ mod boundaries;
 mod dependencies;
 mod errors;
 mod execution;
+mod interleaving;
 mod parallelism;
 mod runtimes;
-mod timing;
 mod tracing;
diff --git a/tests/parallelism/concurrent.rs b/tests/parallelism/concurrent.rs
deleted file mode 100644
index fd746a6..0000000
--- a/tests/parallelism/concurrent.rs
+++ /dev/null
@@ -1,425 +0,0 @@
-//! Concurrent access tests
-
-use dagx::{task, DagRunner};
-use dagx_test::task_fn;
-
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
-use std::time::Duration;
-use tokio::task::JoinSet;
-
-struct SleepTask(usize);
-
-#[task]
-impl SleepTask {
-    async fn run(&self) -> usize {
-        tokio::time::sleep(Duration::from_millis(5)).await;
-        self.0
-    }
-}
-
-#[tokio::test]
-async fn test_concurrent_run_calls() {
-    // Test that multiple run() calls are handled correctly
-    let dag = Arc::new(DagRunner::new());
-
-    // Add some tasks
-    let tasks: Vec<_> = (0..10).map(|i| dag.add_task(SleepTask(i))).collect();
-
-    // Try to run the DAG multiple times concurrently
-    let mut handles = JoinSet::new();
-
-    for _ in 0..5 {
-        let dag = Arc::clone(&dag);
-        handles.spawn(async move {
-            dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-                .await
-        });
-    }
-
-    // Only one run should succeed, others should fail with concurrent execution error
-    let mut success_count = 0;
-    let mut error_count = 0;
-
-    while let Some(result) = handles.join_next().await {
-        match result.unwrap() {
-            Ok(_) => success_count += 1,
-            Err(_) => error_count += 1,
-        }
-    }
-
-    // Exactly one should succeed, the rest should fail
-    assert_eq!(success_count, 1);
-    assert_eq!(error_count, 4);
-
-    // Verify results are consistent from the successful run
-    for (i, task) in tasks.into_iter().enumerate() {
-        assert_eq!(dag.get(task).unwrap(), i);
-    }
-}
-
-#[tokio::test]
-async fn test_concurrent_execution_and_building() {
-    // Test that we can build and execute concurrently
-    let dag = Arc::new(DagRunner::new());
-
-    // Start with some initial tasks
-    let initial: Vec<_> = (0..10)
-        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
-        .collect();
-
-    // Execute the initial DAG
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await
-        .unwrap();
-
-    // Now try to add more tasks while also reading results
-    let mut handles = JoinSet::new();
-
-    // Reader tasks
-    for task in initial {
-        let dag = Arc::clone(&dag);
-        let task_handle: dagx::TaskHandle<i32> = task.into();
-        handles.spawn(async move { dag.get(task_handle).unwrap() });
-    }
-
-    // Builder tasks (adding new nodes)
-    for i in 10..20 {
-        let dag = Arc::clone(&dag);
-        handles.spawn(async move {
-            dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
-            i
-        });
-    }
-
-    // Wait for all operations
-    while let Some(result) = handles.join_next().await {
-        result.unwrap();
-    }
-
-    // Execute again with the new tasks
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await
-        .unwrap();
-}
-
-#[tokio::test]
-async fn test_dag_builder_thread_safety() {
-    // Test that the DAG builder is thread-safe
-    let dag = Arc::new(DagRunner::new());
-    let counter = Arc::new(AtomicUsize::new(0));
-
-    let mut handles = JoinSet::new();
-
-    // Spawn tasks that build complex sub-graphs concurrently
-    for thread_id in 0..10 {
-        let dag = Arc::clone(&dag);
-        let counter = Arc::clone(&counter);
-
-        handles.spawn(async move {
-            // Each thread creates a small pipeline
-            let source = dag.add_task(task_fn::<(), _, _>(move |_: ()| thread_id));
-
-            let transform1 = dag
-                .add_task(task_fn::<i32, _, _>(move |&x: &i32| x * 10))
-                .depends_on(source);
-
-            let transform2 = dag
-                .add_task(task_fn::<i32, _, _>(move |&x: &i32| x + thread_id))
-                .depends_on(transform1);
-
-            counter.fetch_add(1, Ordering::SeqCst);
-            transform2
-        });
-    }
-
-    let mut final_tasks = vec![];
-    while let Some(result) = handles.join_next().await {
-        final_tasks.push(result.unwrap());
-    }
-
-    // Verify all threads completed
-    assert_eq!(counter.load(Ordering::SeqCst), 10);
-
-    // Execute the combined DAG
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await
-        .unwrap();
-
-    // Verify each pipeline computed correctly
-    for task in final_tasks.iter() {
-        let result = dag.get(task).unwrap();
-        // Each pipeline: thread_id * 10 + thread_id
-        assert!((0..100).contains(&result));
-    }
-}
-
-#[tokio::test]
-async fn test_concurrent_get_operations() {
-    // Test concurrent get() operations
-    let dag = Arc::new(DagRunner::new());
-
-    // Create and execute tasks
-    let tasks: Vec<_> = (0..100)
-        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
-        .collect();
-
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await
-        .unwrap();
-
-    // Now access results concurrently
-    let mut handles = JoinSet::new();
-
-    for task in tasks.into_iter() {
-        let dag = Arc::clone(&dag);
-        let task_handle: dagx::TaskHandle<i32> = task.into();
-        handles.spawn(async move { dag.get(task_handle).unwrap() });
-    }
-
-    // Collect and verify results
-    let mut results = vec![];
-    while let Some(result) = handles.join_next().await {
-        results.push(result.unwrap());
-    }
-
-    // Results should contain all values 0..100
-    results.sort();
-    for (i, val) in results.iter().enumerate() {
-        assert_eq!(*val, i as i32);
-    }
-}
-
-#[tokio::test]
-async fn test_no_race_in_dependency_tracking() {
-    // Test that concurrent dependency modifications don't cause races
-    let dag = Arc::new(DagRunner::new());
-    let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
-
-    let mut handles = JoinSet::new();
-
-    // Multiple threads create tasks depending on the same source
-    let source_handle: dagx::TaskHandle<i32> = source.into();
-    for i in 0..50 {
-        let dag = Arc::clone(&dag);
-        let source_clone = source_handle;
-
-        handles.spawn(async move {
-            dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
-                .depends_on(source_clone)
-        });
-    }
-
-    let mut dependents = vec![];
-    while let Some(result) = handles.join_next().await {
-        dependents.push(result.unwrap());
-    }
-
-    // Execute
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await
-        .unwrap();
-
-    // All dependents should have correct values
-    for dep in dependents.iter() {
-        let result = dag.get(dep).unwrap();
-        assert!((42..92).contains(&result)); // 42 + (0..50)
-    }
-}
-
-#[tokio::test]
-async fn test_high_concurrency_stress() {
-    // Stress test with high concurrency
-    let dag = Arc::new(DagRunner::new());
-    let completed = Arc::new(AtomicUsize::new(0));
-
-    let mut handles = JoinSet::new();
-
-    // 1000 concurrent operations
-    for i in 0..1000 {
-        let dag = Arc::clone(&dag);
-        let completed = Arc::clone(&completed);
-
-        handles.spawn(async move {
-            // Mix of operations
-            match i % 3 {
-                0 => {
-                    // Add a source task
-                    dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
-                }
-                1 => {
-                    // Add a task with a simple dependency
-                    let source = dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
-                    dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x * 2))
-                        .depends_on(source);
-                }
-                2 => {
-                    // Add a more complex subgraph
-                    let a = dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
-                    let b = dag.add_task(task_fn::<(), _, _>(move |_: ()| i + 1));
-                    dag.add_task(task_fn::<(i32, i32), _, _>(move |(x, y): (&i32, &i32)| {
-                        x + y
-                    }))
-                    .depends_on((a, b));
-                }
-                _ => unreachable!(),
-            }
-            completed.fetch_add(1, Ordering::SeqCst);
-        });
-    }
-
-    // Wait for all operations to complete
-    while let Some(result) = handles.join_next().await {
-        result.unwrap();
-    }
-
-    assert_eq!(completed.load(Ordering::SeqCst), 1000);
-
-    // The DAG should still be executable
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await
-        .unwrap();
-}
-
-#[tokio::test]
-async fn test_lock_contention_measurement() {
-    // Test to measure lock contention under concurrent load
-    let dag = Arc::new(DagRunner::new());
-    let start = tokio::time::Instant::now();
-    let operations = Arc::new(AtomicUsize::new(0));
-
-    let mut handles = JoinSet::new();
-
-    // Create many concurrent operations that will contend for locks
-    for i in 0..100 {
-        let dag = Arc::clone(&dag);
-        let ops = Arc::clone(&operations);
-
-        handles.spawn(async move {
-            for j in 0..10 {
-                dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 10 + j));
-                ops.fetch_add(1, Ordering::SeqCst);
-
-                // Small yield to increase contention likelihood
-                tokio::task::yield_now().await;
-            }
-        });
-    }
-
-    // Wait for completion
-    while let Some(result) = handles.join_next().await {
-        result.unwrap();
-    }
-
-    let duration = start.elapsed();
-    let total_ops = operations.load(Ordering::SeqCst);
-
-    println!(
-        "Completed {} operations in {:?} ({:.0} ops/sec)",
-        total_ops,
-        duration,
-        total_ops as f64 / duration.as_secs_f64()
-    );
-
-    // Should complete reasonably quickly even with contention
-    assert!(
-        duration < Duration::from_secs(5),
-        "Too much lock contention"
-    );
-}
-
-#[tokio::test]
-async fn test_atomic_dag_operations() {
-    // Test that DAG operations are atomic (all-or-nothing)
-    let dag = Arc::new(DagRunner::new());
-
-    // Create a base task
-    let base = dag.add_task(task_fn::<(), _, _>(|_: ()| 100));
-
-    // Try to create dependent tasks concurrently
-    let mut handles = JoinSet::new();
-    let success_count = Arc::new(AtomicUsize::new(0));
-
-    let base_handle: dagx::TaskHandle<i32> = base.into();
-    for i in 0..50 {
-        let dag = Arc::clone(&dag);
-        let base_clone = base_handle;
-        let success = Arc::clone(&success_count);
-
-        handles.spawn(async move {
-            let dep = dag
-                .add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
-                .depends_on(base_clone);
-
-            success.fetch_add(1, Ordering::SeqCst);
-            dep
-        });
-    }
-
-    let mut deps = vec![];
-    while let Some(result) = handles.join_next().await {
-        deps.push(result.unwrap());
-    }
-
-    // All operations should have succeeded
-    assert_eq!(success_count.load(Ordering::SeqCst), 50);
-    assert_eq!(deps.len(), 50);
-
-    // Execute and verify
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await
-        .unwrap();
-
-    for dep in deps.iter() {
-        let result = dag.get(dep).unwrap();
-        assert!((100..150).contains(&result));
-    }
-}
-
-#[tokio::test]
-async fn test_no_deadlock_on_circular_waiting() {
-    // Test that we don't deadlock even with complex concurrent operations
-    let dag = Arc::new(DagRunner::new());
-
-    // Create multiple threads that might cause circular waiting patterns
-    let mut handles = JoinSet::new();
-
-    for i in 0..10 {
-        let dag = Arc::clone(&dag);
-
-        handles.spawn(async move {
-            // Each thread creates a chain of tasks
-            let first = dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
-            let mut chain = vec![first.into()];
-
-            for j in 0..5 {
-                let next = dag
-                    .add_task(task_fn::<i32, _, _>(move |&x: &i32| x + j))
-                    .depends_on(chain.last().unwrap());
-                chain.push(next);
-
-                // Yield to potentially cause interleaving
-                tokio::task::yield_now().await;
-            }
-
-            *chain.last().unwrap()
-        });
-    }
-
-    // Set a timeout to detect deadlocks
-    let result = tokio::time::timeout(Duration::from_secs(5), async {
-        let mut finals = vec![];
-        while let Some(result) = handles.join_next().await {
-            finals.push(result.unwrap());
-        }
-        finals
-    })
-    .await;
-
-    assert!(result.is_ok(), "Deadlock detected - operation timed out");
-
-    // Should be able to execute the DAG
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await
-        .unwrap();
-}
diff --git a/tests/parallelism/mod.rs b/tests/parallelism/mod.rs
index 93421db..adf6362 100644
--- a/tests/parallelism/mod.rs
+++ b/tests/parallelism/mod.rs
@@ -1,6 +1,5 @@
 //! Parallelism tests module
 
-pub mod concurrent;
 pub mod spawning;
 pub mod threads;
 pub mod timing;
diff --git a/tests/parallelism/spawning.rs b/tests/parallelism/spawning.rs
index 1ddf148..a6070c4 100644
--- a/tests/parallelism/spawning.rs
+++ b/tests/parallelism/spawning.rs
@@ -9,7 +9,7 @@ use std::sync::Arc;
 #[tokio::test]
 async fn test_spawner_actually_spawns_tasks() -> DagResult<()> {
     // Verify that the spawner function is actually called for each task
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let spawn_count = Arc::new(AtomicUsize::new(0));
 
     // Create 10 independent tasks
@@ -27,12 +27,13 @@ async fn test_spawner_actually_spawns_tasks() -> DagResult<()> {
         .collect();
 
     // Custom spawner that counts invocations
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Verify all tasks were spawned
     for (i, task) in tasks.into_iter().enumerate() {
-        assert_eq!(dag.get(task)?, i as i32);
+        assert_eq!(output.get(task)?, i as i32);
     }
 
     // The spawner should have been called once for each task
diff --git a/tests/parallelism/threads.rs b/tests/parallelism/threads.rs
index 59cbd29..6636ebb 100644
--- a/tests/parallelism/threads.rs
+++ b/tests/parallelism/threads.rs
@@ -1,6 +1,6 @@
 //! Tests for multi-threading and concurrent execution
 
-use dagx::{task, DagResult, DagRunner, TaskHandle};
+use dagx::{task, DagResult, DagRunner};
 use dagx_test::task_fn;
 
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -11,7 +11,7 @@ use tokio::time::sleep;
 #[tokio::test]
 async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> {
     // Prove concurrent execution using atomic counters to track simultaneous tasks
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
     let concurrent_count = Arc::new(AtomicUsize::new(0));
     let max_concurrent = Arc::new(AtomicUsize::new(0));
 
@@ -62,12 +62,13 @@ async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> {
         })
         .collect();
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
     // Verify all tasks completed
     for (i, task) in tasks.into_iter().enumerate() {
-        assert_eq!(dag.get(task)?, i);
+        assert_eq!(output.get(task)?, i);
     }
 
     // Check that we achieved real parallelism
@@ -90,13 +91,13 @@ async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> {
 #[tokio::test]
 async fn test_massive_parallel_fanout() -> DagResult<()> {
     // Test extreme parallelism with 1000 parallel tasks
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let completed = Arc::new(AtomicUsize::new(0));
     let max_concurrent = Arc::new(AtomicUsize::new(0));
     let current = Arc::new(AtomicUsize::new(0));
 
-    let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 1)).into();
+    let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
 
     struct FanoutTask {
         comp: Arc<AtomicUsize>,
@@ -147,7 +148,8 @@ async fn test_massive_parallel_fanout() -> DagResult<()> {
         .collect();
 
     let start = Instant::now();
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
     let elapsed = start.elapsed();
 
@@ -155,9 +157,9 @@ async fn test_massive_parallel_fanout() -> DagResult<()> {
     assert_eq!(completed.load(Ordering::SeqCst), 1000);
 
     // Check some results
-    assert_eq!(dag.get(tasks[0])?, 1);
-    assert_eq!(dag.get(tasks[500])?, 501);
-    assert_eq!(dag.get(tasks[999])?, 1000);
+    assert_eq!(output.get(tasks[0])?, 1);
+    assert_eq!(output.get(tasks[500])?, 501);
+    assert_eq!(output.get(tasks[999])?, 1000);
 
     let max_seen = max_concurrent.load(Ordering::SeqCst);
     println!(
diff --git a/tests/parallelism/timing.rs b/tests/parallelism/timing.rs
index 8903ea0..9ecc4d7 100644
--- a/tests/parallelism/timing.rs
+++ b/tests/parallelism/timing.rs
@@ -22,7 +22,7 @@ impl SleepTask {
 #[tokio::test]
 async fn test_parallel_execution_speedup() -> DagResult<()> {
     // Prove parallel execution by comparing timing
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let work_duration = Duration::from_millis(50);
     let num_tasks = 10;
@@ -33,13 +33,14 @@ async fn test_parallel_execution_speedup() -> DagResult<()> {
         .collect();
 
     let start = Instant::now();
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
     let elapsed = start.elapsed();
 
     // Verify all tasks completed
     for (i, task) in tasks.into_iter().enumerate() {
-        assert_eq!(dag.get(task)?, i);
+        assert_eq!(output.get(task)?, i);
     }
 
     // If tasks ran sequentially: 10 * 50ms = 500ms
@@ -53,7 +54,7 @@ async fn test_parallel_execution_speedup() -> DagResult<()> {
 
     // Parallel execution should be significantly faster
     assert!(
-        elapsed < sequential_time / 4,
+        elapsed < sequential_time / 2,
         "Parallel execution too slow: {:?} vs sequential {:?}",
         elapsed,
         sequential_time
@@ -65,7 +66,7 @@ async fn test_parallel_execution_speedup() -> DagResult<()> {
 #[tokio::test]
 async fn test_layers_execute_in_parallel() -> DagResult<()> {
     // Test that tasks in the same topological layer execute in parallel
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let layer_timing = Arc::new(Mutex::new(Vec::new()));
 
@@ -142,10 +143,11 @@ async fn test_layers_execute_in_parallel() -> DagResult<()> {
         .depends_on((&layer1[0], &layer1[1], &layer1[2]))
     };
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await?;
 
-    assert_eq!(dag.get(final_task)?, 33); // (0+1) + (10+1) + (20+1)
+    assert_eq!(output.get(final_task)?, 33); // (0+1) + (10+1) + (20+1)
 
     // Analyze timing
     let timing = layer_timing.lock().unwrap();
diff --git a/tests/runtimes/mod.rs b/tests/runtimes/mod.rs
index 740a1da..ca424fe 100644
--- a/tests/runtimes/mod.rs
+++ b/tests/runtimes/mod.rs
@@ -95,15 +95,15 @@ impl Add {
 #[test_case(AsyncExecutorRuntimeTest)]
 fn test_basic_dag(runner: impl RuntimeTest) {
     runner.run_dagx_test(|spawner| async {
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         let x = dag.add_task(Value(2));
         let y = dag.add_task(Value(3));
         let sum = dag.add_task(Add).depends_on((x, y));
 
-        dag.run(spawner).await.unwrap();
+        let mut output = dag.run(spawner).await.unwrap();
 
-        assert_eq!(dag.get(sum).unwrap(), 5);
+        assert_eq!(output.get(sum).unwrap(), 5);
     })
 }
 
@@ -113,16 +113,16 @@ fn test_basic_dag(runner: impl RuntimeTest) {
 #[test_case(AsyncExecutorRuntimeTest)]
 fn test_parallel_execution(runner: impl RuntimeTest) {
     runner.run_dagx_test(|spawner| async {
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         let tasks: Vec<_> = (0..10)
             .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 2)))
             .collect();
 
-        dag.run(spawner).await.unwrap();
+        let mut output = dag.run(spawner).await.unwrap();
 
         for (i, task) in tasks.into_iter().enumerate() {
-            assert_eq!(dag.get(task).unwrap(), i * 2);
+            assert_eq!(output.get(task).unwrap(), i * 2);
         }
     });
 }
@@ -133,7 +133,7 @@ fn test_parallel_execution(runner: impl RuntimeTest) {
 #[test_case(AsyncExecutorRuntimeTest)]
 fn test_complex_dependencies(runner: impl RuntimeTest) {
     runner.run_dagx_test(|spawner| async {
-        let dag = DagRunner::new();
+        let mut dag = DagRunner::new();
 
         let a = dag.add_task(Value(10));
         let b = dag.add_task(Value(20));
@@ -142,8 +142,8 @@ fn test_complex_dependencies(runner: impl RuntimeTest) {
             .add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
             .depends_on(sum);
 
-        dag.run(spawner).await.unwrap();
+        let mut output = dag.run(spawner).await.unwrap();
 
-        assert_eq!(dag.get(double).unwrap().clone(), 60);
+        assert_eq!(output.get(double).unwrap().clone(), 60);
     });
 }
diff --git a/tests/timing/concurrent_access.rs b/tests/timing/concurrent_access.rs
deleted file mode 100644
index ad5b147..0000000
--- a/tests/timing/concurrent_access.rs
+++ /dev/null
@@ -1,331 +0,0 @@
-//! Tests for concurrent access patterns and race conditions
-
-use dagx::{task, DagResult, DagRunner, TaskHandle};
-use dagx_test::task_fn;
-
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::sync::Arc;
-use std::time::Duration;
-use tokio::sync::RwLock;
-
-#[tokio::test]
-async fn test_concurrent_result_retrieval() -> DagResult<()> {
-    // Test concurrent get() calls after execution
-    let dag = Arc::new(DagRunner::new());
-
-    struct DelayTask(usize, Duration);
-
-    #[task]
-    impl DelayTask {
-        async fn run(&self) -> usize {
-            tokio::time::sleep(self.1).await;
-            self.0
-        }
-    }
-
-    // Add tasks with minimal delays
-    let tasks: Vec<dagx::TaskHandle<usize>> = (0..10)
-        .map(|i| {
-            let task = dag.add_task(DelayTask(i, Duration::from_millis(1)));
-            task.into()
-        })
-        .collect();
-
-    // Execute the DAG
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await?;
-
-    // Now try to get results concurrently from multiple threads
-    let mut get_handles = Vec::new();
-    for (i, task) in tasks.iter().enumerate() {
-        let dag_clone = dag.clone();
-        let task = *task;
-        get_handles.push(tokio::spawn(async move {
-            // Results should be available immediately
-            dag_clone.get(task).map(|v| (i, v))
-        }));
-    }
-
-    // Verify all gets succeeded
-    for handle in get_handles {
-        let (i, result) = handle.await.unwrap()?;
-        assert_eq!(result, i);
-    }
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_simultaneous_dag_building_and_execution() -> DagResult<()> {
-    // Test building DAG while another part is executing
-    let dag = Arc::new(DagRunner::new());
-    let building_complete = Arc::new(AtomicBool::new(false));
-
-    // Start with a few initial tasks
-    let initial: Vec<_> = (0..5)
-        .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
-        .collect();
-
-    // Start building more tasks in background
-    let dag_build = dag.clone();
-    let building = building_complete.clone();
-    let build_handle = tokio::spawn(async move {
-        let mut tasks: Vec<dagx::TaskHandle<i32>> = Vec::new();
-        for i in 5..50 {
-            let task = dag_build.add_task(task_fn::<(), _, _>(move |_: ()| i));
-            tasks.push(task.into());
-            tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
-        }
-        building.store(true, Ordering::SeqCst);
-        tasks
-    });
-
-    // Wait a bit for some tasks to be added
-    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
-
-    // Execute the DAG
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await?;
-
-    // Get the additional tasks
-    let additional_tasks = build_handle.await.unwrap();
-
-    // Initial tasks should have results
-    for (i, task) in initial.into_iter().enumerate() {
-        assert_eq!(dag.get(task)?, i);
-    }
-
-    // Additional tasks added during execution won't have results yet
-    // (they need another run cycle)
-    if building_complete.load(Ordering::SeqCst) {
-        // Run again for the new tasks
-        dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-            .await?;
-
-        // Now they should have results
-        for (i, task) in additional_tasks.iter().enumerate() {
-            assert_eq!(dag.get(task)?, (i + 5) as i32);
-        }
-    }
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_concurrent_dependency_resolution() -> DagResult<()> {
-    // Test multiple tasks resolving dependencies concurrently
-    let dag = DagRunner::new();
-    let resolution_counter = Arc::new(AtomicUsize::new(0));
-
-    // Create a source task
-    let source: TaskHandle<_> = dag
-        .add_task(task_fn::<(), _, _>({
-            let counter = resolution_counter.clone();
-            move |_: ()| {
-                let counter = counter.clone();
-                counter.fetch_add(1, Ordering::SeqCst);
-                42
-            }
-        }))
-        .into();
-
-    // Create many tasks depending on the same source
-    let dependents: Vec<_> = (0..100)
-        .map(|i| {
-            let counter = resolution_counter.clone();
-            dag.add_task(task_fn::<i32, _, _>(move |val: &i32| {
-                let counter = counter.clone();
-                counter.fetch_add(1, Ordering::SeqCst);
-                val + i
-            }))
-            .depends_on(source)
-        })
-        .collect();
-
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await?;
-
-    // Check all completed
-    assert_eq!(resolution_counter.load(Ordering::SeqCst), 101);
-
-    // Verify results
-    for (i, task) in dependents.iter().enumerate() {
-        assert_eq!(dag.get(task)?, 42 + i as i32);
-    }
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_read_write_race_patterns() -> DagResult<()> {
-    // Test read-write patterns with shared state
-    let dag = DagRunner::new();
-    let shared_state = Arc::new(RwLock::new(Vec::new()));
-
-    struct WriterTask {
-        state: Arc<RwLock<Vec<i32>>>,
-        idx: i32,
-    }
-
-    #[task]
-    impl WriterTask {
-        async fn run(&self) -> i32 {
-            let mut guard = self.state.write().await;
-            guard.push(self.idx);
-            tokio::task::yield_now().await; // Force potential race
-            guard.push(self.idx * 10);
-            self.idx
-        }
-    }
-
-    // Writers
-    let writers: Vec<_> = (0..10)
-        .map(|i| {
-            dag.add_task(WriterTask {
-                state: shared_state.clone(),
-                idx: i,
-            })
-            .into()
-        })
-        .collect();
-
-    struct ReaderTask {
-        state: Arc<RwLock<Vec<i32>>>,
-    }
-
-    struct ReaderTaskOutput((i32, bool, usize));
-
-    #[task]
-    impl ReaderTask {
-        async fn run(&self, writer_id: &i32) -> ReaderTaskOutput {
-            let guard = self.state.read().await;
-            let contains_writer = guard.contains(&writer_id);
-            let len = guard.len();
-            ReaderTaskOutput((*writer_id, contains_writer, len))
-        }
-    }
-
-    // Readers (depend on writers to ensure ordering)
-    let readers: Vec<_> = writers
-        .iter()
-        .map(|writer| {
-            dag.add_task(ReaderTask {
-                state: shared_state.clone(),
-            })
-            .depends_on(writer)
-        })
-        .collect();
-
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await?;
-
-    // Verify readers saw their writer's data
-    for reader in &readers {
-        let ReaderTaskOutput((writer_id, saw_writer, _len)) = dag.get(reader)?;
-        assert!(saw_writer, "Reader didn't see writer {}", writer_id);
-    }
-
-    // Final state should have all writes
-    let final_state = shared_state.read().await;
-    assert_eq!(final_state.len(), 20); // 10 writers * 2 writes each
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_atomic_counter_races() -> DagResult<()> {
-    // Test atomic operations under high concurrency
-    let dag = DagRunner::new();
-    let counter = Arc::new(AtomicUsize::new(0));
-    let max_seen = Arc::new(AtomicUsize::new(0));
-
-    // Create tasks that increment and track max
-    let _tasks: Vec<_> = (0..200)
-        .map(|_| {
-            let counter = counter.clone();
-            let max_seen = max_seen.clone();
-            dag.add_task(task_fn::<(), _, _>(move |_: ()| {
-                let counter = counter.clone();
-                let max_seen = max_seen.clone();
-                let mut local_max = 0;
-                for _ in 0..100 {
-                    let val = counter.fetch_add(1, Ordering::SeqCst);
-                    local_max = local_max.max(val);
-
-                    // Update global max
-                    let mut current_max = max_seen.load(Ordering::SeqCst);
-                    while val > current_max {
-                        match max_seen.compare_exchange_weak(
-                            current_max,
-                            val,
-                            Ordering::SeqCst,
-                            Ordering::SeqCst,
-                        ) {
-                            Ok(_) => break,
-                            Err(x) => current_max = x,
-                        }
-                    }
-                }
-            }))
-        })
-        .collect();
-
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-        .await?;
-
-    // Final counter should be 200 * 100 = 20000
-    assert_eq!(counter.load(Ordering::SeqCst), 20000);
-
-    // Max seen should be at least 19999 (last increment - 1)
-    assert!(max_seen.load(Ordering::SeqCst) >= 19999);
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_multiple_dag_runners_interaction() -> DagResult<()> {
-    // Test multiple DAG runners potentially interfering
-    let shared_counter = Arc::new(AtomicUsize::new(0));
-
-    let mut handles = Vec::new();
-
-    for dag_id in 0..5 {
-        let counter = shared_counter.clone();
-        let handle = tokio::spawn(async move {
-            let dag = DagRunner::new();
-
-            let tasks: Vec<_> = (0..20)
-                .map(|task_id| {
-                    let counter = counter.clone();
-                    dag.add_task(task_fn::<(), _, _>(move |_: ()| {
-                        let counter = counter.clone();
-                        counter.fetch_add(1, Ordering::SeqCst);
-                        (dag_id, task_id)
-                    }))
-                })
-                .collect();
-
-            dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
-                .await?;
-
-            // Collect results
-            let mut results = Vec::new();
-            for task in tasks {
-                results.push(dag.get(task)?);
-            }
-
-            Ok::<_, dagx::DagError>(results)
-        });
-        handles.push(handle);
-    }
-
-    // Wait for all DAGs to complete
-    for handle in handles {
-        handle.await.unwrap()?;
-    }
-
-    // Should have executed 5 * 20 = 100 tasks total
-    assert_eq!(shared_counter.load(Ordering::SeqCst), 100);
-
-    Ok(())
-}
diff --git a/tests/timing/mod.rs b/tests/timing/mod.rs
deleted file mode 100644
index 58d63cf..0000000
--- a/tests/timing/mod.rs
+++ /dev/null
@@ -1,6 +0,0 @@
-//! Timing and race condition tests
-//!
-//! Tests for temporal edge cases, execution ordering, and race conditions.
-
-mod concurrent_access;
-mod interleaving;
diff --git a/tests/tracing/with_tracing.rs b/tests/tracing/with_tracing.rs
index c087143..d8ef38a 100644
--- a/tests/tracing/with_tracing.rs
+++ b/tests/tracing/with_tracing.rs
@@ -39,17 +39,18 @@ async fn test_tracing_with_subscriber() {
         .with_test_writer()
         .try_init();
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     let a = dag.add_task(Value(2));
     let b = dag.add_task(Value(3));
     let sum = dag.add_task(Add).depends_on((a, b));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
-    assert_eq!(dag.get(sum).unwrap(), 5);
+    assert_eq!(output.get(sum).unwrap(), 5);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -60,19 +61,20 @@ async fn test_tracing_inline_execution() {
         .with_test_writer()
         .try_init();
 
-    let dag = DagRunner::new();
+    let mut dag = DagRunner::new();
 
     // Create a linear chain (should trigger inline execution)
-    let a = dag.add_task(Value(1)).into();
+    let a = dag.add_task(Value(1));
     let b = dag.add_task(Add).depends_on((&a, &a));
     let c = dag.add_task(Multiply).depends_on((&b, &b));
 
-    dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
+    let mut output = dag
+        .run(|fut| async move { tokio::spawn(fut).await.unwrap() })
         .await
         .unwrap();
 
     // a = 1
     // b = 1 + 1 = 2
     // c = 2 * 2 = 4
-    assert_eq!(dag.get(c).unwrap(), 4);
+    assert_eq!(output.get(c).unwrap(), 4);
 }