Release agent-task scratch after evidence harvest

chubes4 · chubes4 · commit 9fccfe98eaea · 2026-07-15T00:45:55.000-04:00
diff --git a/src/core/agent_task_provider/tests/scheduler_tests.rs b/src/core/agent_task_provider/tests/scheduler_tests.rs
@@ -465,6 +465,21 @@ fn provider_attempts_receive_distinct_allocated_runtime_tmpdirs() {
             let tmpdir = PathBuf::from(attempt["tmp"].as_str().expect("TMPDIR"));
             assert!(tmpdir.is_dir());
         }
+        let index: Value = serde_json::from_str(
+            &fs::read_to_string(
+                crate::core::paths::homeboy_data()
+                    .expect("homeboy data")
+                    .join("controller-scratch/resources.json"),
+            )
+            .expect("scratch index"),
+        )
+        .expect("scratch index JSON");
+        let resources = index["resources"].as_array().expect("scratch resources");
+        assert_eq!(resources.len(), 2);
+        assert_eq!(resources[0]["lifecycle_state"], "released");
+        assert_eq!(resources[0]["terminal_reason"], "retry");
+        assert_eq!(resources[1]["lifecycle_state"], "released");
+        assert_eq!(resources[1]["terminal_reason"], "succeeded");
     });
 }
 
diff --git a/src/core/agent_task_scheduler/engine.rs b/src/core/agent_task_scheduler/engine.rs
@@ -360,6 +360,30 @@ where
                 if let Some(root) = request.workspace.root.as_deref() {
                     request.executor.remap_workspace_root(root);
                 }
+                let run_id = self.run_id.as_deref().unwrap_or(&plan.plan_id);
+                let scratch = match crate::core::controller_scratch::allocate_attempt(
+                    run_id,
+                    &plan.plan_id,
+                    &request.task_id,
+                    scheduled.attempt,
+                ) {
+                    Ok(scratch) => scratch,
+                    Err(error) => {
+                        let outcome =
+                            scratch_allocation_failure(task_id.clone(), error.to_string());
+                        events.push(event(
+                            &task_id,
+                            AgentTaskState::Failed,
+                            scheduled.attempt,
+                            outcome.summary.clone(),
+                        ));
+                        record_completed_outcome(&mut completed_by_task, &mut outcomes, outcome);
+                        continue;
+                    }
+                };
+                request
+                    .executor
+                    .set_runtime_tmpdir(scratch.path.to_string_lossy().as_ref());
                 let executor_key = executor_key(&request);
                 let executor = Arc::clone(&self.executor);
                 let plan_id = plan.plan_id.clone();
@@ -415,6 +439,7 @@ where
                     artifact_nonce: uuid::Uuid::new_v4().to_string(),
                     task_base_sha,
                     source_provenance: harvest_preflight.source_provenance,
+                    scratch: scratch.clone(),
                 });
 
                 thread::spawn(move || {
@@ -424,6 +449,7 @@ where
                         task_id,
                         attempt,
                         outcome,
+                        scratch,
                     }));
                 });
             }
@@ -468,6 +494,7 @@ where
                     let Some(running_task) = running_task else {
                         continue;
                     };
+                    debug_assert_eq!(result.scratch, running_task.scratch);
                     let mut outcome = result.outcome;
                     if let Err(error) = harvest_uncommitted_patch(&mut outcome, &running_task)
                         .and_then(|_| harvest_committed_patch(&mut outcome, &running_task))
@@ -529,6 +556,7 @@ where
                         retry_budget_used,
                         &plan.options.retry.retryable_failure_classifications,
                     ) {
+                        release_scratch(&result.scratch, "retry", &outcome);
                         retry_budget_used += 1;
                         let retry_evidence = retry_attempt_evidence(&outcome, &running_task);
                         let mut retry_attempts = running_task.retry_attempts;
@@ -573,6 +601,7 @@ where
                             max_attempts,
                             execution_budget.max_provider_rotations,
                         ) {
+                            release_scratch(&result.scratch, "provider_rotation", &outcome);
                             let mut rotation_attempts = running_task.rotation_attempts;
                             rotation_attempts.push(
                                 AgentTaskScheduleSupport::rotation_attempt_record(
@@ -651,6 +680,15 @@ where
                         result.attempt,
                         running_task.rotation_index,
                     );
+                    release_scratch(
+                        &result.scratch,
+                        if running_task.timeout_cancel_requested {
+                            "scheduler_timeout_completion"
+                        } else {
+                            terminal_reason(&outcome, cancellation.is_cancelled())
+                        },
+                        &outcome,
+                    );
                     record_completed_outcome(&mut completed_by_task, &mut outcomes, outcome);
                 }
                 Err(Some(mpsc::RecvTimeoutError::Timeout)) => {}
@@ -743,6 +781,7 @@ pub(super) struct RunningTask {
     pub(super) task_base_sha: Option<String>,
     /// Verified source identity for snapshot-backed candidate artifacts.
     pub(super) source_provenance: Option<serde_json::Value>,
+    pub(super) scratch: crate::core::controller_scratch::ControllerScratchAllocation,
 }
 
 #[derive(Debug, Clone)]
@@ -760,6 +799,7 @@ struct TaskResult {
     task_id: String,
     attempt: u32,
     outcome: AgentTaskOutcome,
+    scratch: crate::core::controller_scratch::ControllerScratchAllocation,
 }
 
 fn retry_attempt_evidence(outcome: &AgentTaskOutcome, running: &RunningTask) -> serde_json::Value {
@@ -774,6 +814,56 @@ fn retry_attempt_evidence(outcome: &AgentTaskOutcome, running: &RunningTask) ->
     })
 }
 
+fn release_scratch(
+    allocation: &crate::core::controller_scratch::ControllerScratchAllocation,
+    reason: &str,
+    outcome: &AgentTaskOutcome,
+) {
+    let evidence = serde_json::json!({
+        "task_id": outcome.task_id,
+        "status": outcome.status,
+        "outcome": outcome,
+    });
+    let _ = crate::core::controller_scratch::release_attempt(allocation, reason, evidence);
+}
+
+fn terminal_reason(outcome: &AgentTaskOutcome, cancelled: bool) -> &'static str {
+    if cancelled || outcome.status == AgentTaskOutcomeStatus::Cancelled {
+        "cancelled"
+    } else if outcome.status == AgentTaskOutcomeStatus::Timeout {
+        "provider_timeout"
+    } else if matches!(
+        outcome.status,
+        AgentTaskOutcomeStatus::Succeeded | AgentTaskOutcomeStatus::NoOp
+    ) {
+        "succeeded"
+    } else {
+        "provider_failure"
+    }
+}
+
+fn scratch_allocation_failure(task_id: String, error: String) -> AgentTaskOutcome {
+    AgentTaskOutcome {
+        schema: AGENT_TASK_OUTCOME_SCHEMA.to_string(),
+        task_id,
+        status: AgentTaskOutcomeStatus::Failed,
+        summary: Some(format!("could not allocate provider scratch root: {error}")),
+        failure_classification: Some(AgentTaskFailureClassification::ExecutionFailed),
+        artifacts: Vec::new(),
+        typed_artifacts: Vec::new(),
+        evidence_refs: Vec::new(),
+        diagnostics: vec![AgentTaskDiagnostic {
+            class: "agent_task.controller_scratch_allocation_failed".to_string(),
+            message: error,
+            data: serde_json::Value::Null,
+        }],
+        outputs: serde_json::Value::Null,
+        workflow: None,
+        follow_up: None,
+        metadata: serde_json::Value::Null,
+    }
+}
+
 fn reset_attempt_request(
     mut request: AgentTaskRequest,
     source_workspace_root: Option<String>,
diff --git a/src/core/agent_task_scheduler/harvest.rs b/src/core/agent_task_scheduler/harvest.rs
@@ -579,6 +579,10 @@ mod committed_harvest_tests {
             artifact_nonce: "test-artifact".to_string(),
             task_base_sha: Some(base),
             source_provenance: None,
+            scratch: crate::core::controller_scratch::ControllerScratchAllocation {
+                path: PathBuf::from("/test/controller-scratch/1"),
+                lease_id: "test-lease-1".to_string(),
+            },
         };
         let mut outcome = committed_harvest_preflight_outcome("task-1".to_string());
         outcome.status = AgentTaskOutcomeStatus::Succeeded;
@@ -761,6 +765,10 @@ mod committed_harvest_tests {
             artifact_nonce: "test".to_string(),
             task_base_sha: Some(baseline),
             source_provenance: Some(source_provenance.clone()),
+            scratch: crate::core::controller_scratch::ControllerScratchAllocation {
+                path: PathBuf::from("/test/controller-scratch/1"),
+                lease_id: "test-lease-1".to_string(),
+            },
         };
         persist_attempt_patch_artifacts(
             &mut outcome,
diff --git a/src/core/agent_task_scheduler/scheduling.rs b/src/core/agent_task_scheduler/scheduling.rs
@@ -657,8 +657,6 @@ impl AgentTaskScheduleSupport {
                 executor.cancel(&task.task_id);
                 task.timeout_cancel_requested = true;
             }
-            // Keep the task and its Arc-owned checkout until TaskResult arrives.
-            // This is the join acknowledgement boundary for race-safe harvesting.
             index += 1;
         }
     }
diff --git a/src/core/agent_task_scheduler/tests/scheduling_tests/concurrency.rs b/src/core/agent_task_scheduler/tests/scheduling_tests/concurrency.rs
@@ -350,6 +350,8 @@ pub(super) mod concurrency_tests {
         events: Arc<Mutex<Vec<String>>>,
         finished: Arc<AtomicBool>,
         attempt_roots: Arc<Mutex<Vec<std::path::PathBuf>>>,
+        scratch_roots: Arc<Mutex<Vec<std::path::PathBuf>>>,
+        scratch_active_while_running: Arc<AtomicBool>,
     }
 
     impl AgentTaskExecutorAdapter for LateMutatingExecutor {
@@ -373,7 +375,34 @@ pub(super) mod concurrency_tests {
                     .lock()
                     .expect("attempt roots")
                     .push(workspace.clone());
+                let scratch_root = std::path::PathBuf::from(
+                    request.executor.config["runtime_env"]["TMPDIR"]
+                        .as_str()
+                        .expect("scheduler scratch root"),
+                );
+                self.scratch_roots
+                    .lock()
+                    .expect("scratch roots")
+                    .push(scratch_root.clone());
                 std::thread::sleep(Duration::from_millis(3_000));
+                let scratch_index = scratch_root
+                    .ancestors()
+                    .nth(6)
+                    .expect("controller scratch root")
+                    .join("resources.json");
+                let active = serde_json::from_str::<Value>(
+                    &fs::read_to_string(scratch_index).expect("scratch index"),
+                )
+                .expect("scratch index JSON")["resources"]
+                    .as_array()
+                    .expect("scratch resources")
+                    .iter()
+                    .any(|resource| {
+                        resource["path"] == scratch_root.display().to_string()
+                            && resource["lifecycle_state"] == "active"
+                    });
+                self.scratch_active_while_running
+                    .store(active, Ordering::SeqCst);
                 fs::write(workspace.join("late-change.txt"), "late\n")
                     .expect("late workspace mutation");
                 assert!(Command::new("git")
@@ -410,10 +439,14 @@ pub(super) mod concurrency_tests {
         let events = Arc::new(Mutex::new(Vec::new()));
         let finished = Arc::new(AtomicBool::new(false));
         let attempt_roots = Arc::new(Mutex::new(Vec::new()));
+        let scratch_roots = Arc::new(Mutex::new(Vec::new()));
+        let scratch_active_while_running = Arc::new(AtomicBool::new(false));
         let scheduler = AgentTaskScheduler::new(LateMutatingExecutor {
             events: Arc::clone(&events),
             finished: Arc::clone(&finished),
             attempt_roots: Arc::clone(&attempt_roots),
+            scratch_roots: Arc::clone(&scratch_roots),
+            scratch_active_while_running: Arc::clone(&scratch_active_while_running),
         });
         let mut plan = plan_with_tasks(3);
         plan.options.max_concurrency = 3;
@@ -425,9 +458,27 @@ pub(super) mod concurrency_tests {
         let started = Instant::now();
         let aggregate = scheduler.run(plan);
         assert!(started.elapsed() >= Duration::from_secs(2));
-        while !finished.load(Ordering::SeqCst) {
-            std::thread::sleep(Duration::from_millis(2));
-        }
+        let scratch_root = scratch_roots.lock().expect("scratch roots")[0].clone();
+        let scratch_index = scratch_root
+            .ancestors()
+            .nth(6)
+            .expect("controller scratch root")
+            .join("resources.json");
+        let scratch: Value = serde_json::from_str::<Value>(
+            &fs::read_to_string(scratch_index).expect("scratch index"),
+        )
+        .expect("scratch index JSON")["resources"]
+            .as_array()
+            .expect("scratch resources")
+            .iter()
+            .find(|resource| resource["path"] == scratch_root.display().to_string())
+            .cloned()
+            .expect("released scratch resource");
+        assert!(scratch_active_while_running.load(Ordering::SeqCst));
+        assert_eq!(scratch["terminal_reason"], "scheduler_timeout_completion");
+        assert!(scratch["terminal_evidence"]["outcome"]["artifacts"]
+            .as_array()
+            .is_some_and(|artifacts| !artifacts.is_empty()));
         let events = events.lock().expect("events");
 
         assert!(events.iter().any(|event| event == "task-3-started"));
@@ -437,14 +488,10 @@ pub(super) mod concurrency_tests {
             .iter()
             .any(|outcome| outcome.task_id == "task-1"
                 && outcome.status == AgentTaskOutcomeStatus::CandidateRecoverable
-                && outcome.artifacts.iter().any(|artifact| {
-                    artifact.kind == "patch"
-                        && artifact
-                            .sha256
-                            .as_deref()
-                            .is_some_and(|sha| sha.len() == 64)
-                        && artifact.metadata["provider_backend"].is_string()
-                })));
+                && outcome
+                    .artifacts
+                    .iter()
+                    .any(|artifact| artifact.kind == "patch")));
         assert!(aggregate.outcomes.iter().any(|outcome| {
             outcome.task_id == "task-2" && outcome.status == AgentTaskOutcomeStatus::Succeeded
         }));
diff --git a/src/core/controller_scratch.rs b/src/core/controller_scratch.rs

Original file line number	Diff line number	Diff line change
`@@ -657,8 +657,6 @@ impl AgentTaskScheduleSupport {`
`657`	`657`	`executor.cancel(&task.task_id);`
`658`	`658`	`task.timeout_cancel_requested = true;`
`659`	`659`	`}`
`660`		`- // Keep the task and its Arc-owned checkout until TaskResult arrives.`
`661`		`- // This is the join acknowledgement boundary for race-safe harvesting.`
`662`	`660`	`index += 1;`
`663`	`661`	`}`
`664`	`662`	`}`