//! Scheduler dispatch, concurrency, retry, dependency-binding, matrix, and
//! cancellation behavior.

use super::super::*;
use super::fixtures::*;
use crate::core::agent_task::{
    expand_agent_task_matrix, AgentTaskArtifact, AgentTaskArtifactDeclaration,
    AgentTaskMatrixAggregate, AgentTaskMatrixAxis, AgentTaskTypedArtifact,
    AGENT_TASK_ARTIFACT_SCHEMA, AGENT_TASK_OUTCOME_SCHEMA,
};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::fs;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;

mod adaptive_concurrency_tests {
    use super::*;

    #[test]
    fn adaptive_concurrency_scales_up_when_runner_slots_are_available() {
        let executor = RecordingExecutor::new(HashMap::new(), Duration::from_millis(25));
        let max_seen = Arc::clone(&executor.max_seen);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(4);
        plan.options.max_concurrency = 1;
        plan.options.adaptive_concurrency = Some(AgentTaskAdaptiveConcurrencyPolicy {
            max_concurrency: Some(3),
            runner_capacity: Some(3),
            ..AgentTaskAdaptiveConcurrencyPolicy::default()
        });

        let aggregate = scheduler.run(plan);
        let adaptive = aggregate
            .queue
            .adaptive_concurrency
            .expect("adaptive status");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert!(max_seen.load(Ordering::SeqCst) > 1);
        assert!(max_seen.load(Ordering::SeqCst) <= 3);
        assert_eq!(adaptive.configured_max_concurrency, 1);
        assert_eq!(adaptive.max_concurrency, 3);
        assert!(adaptive.decisions.iter().any(|decision| {
            decision.action == AgentTaskAdaptiveConcurrencyAction::Increased
                && decision.effective_concurrency == 3
                && decision.reason.contains("runner slots are available")
        }));
    }

    #[test]
    fn adaptive_concurrency_scales_down_under_runner_pressure() {
        let executor = RecordingExecutor::new(HashMap::new(), Duration::from_millis(25));
        let max_seen = Arc::clone(&executor.max_seen);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(3);
        plan.options.max_concurrency = 4;
        plan.options.adaptive_concurrency = Some(AgentTaskAdaptiveConcurrencyPolicy {
            max_concurrency: Some(4),
            runner_capacity: Some(3),
            active_leases: 2,
            ..AgentTaskAdaptiveConcurrencyPolicy::default()
        });

        let aggregate = scheduler.run(plan);
        let adaptive = aggregate
            .queue
            .adaptive_concurrency
            .expect("adaptive status");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert!(max_seen.load(Ordering::SeqCst) <= 1);
        assert_eq!(adaptive.effective_concurrency, 1);
        assert!(adaptive.decisions.iter().any(|decision| {
            decision.action == AgentTaskAdaptiveConcurrencyAction::Decreased
                && decision.reason.contains("available runner slots 1")
        }));
    }

    #[test]
    fn adaptive_concurrency_pauses_and_blocks_when_runner_capacity_is_unavailable() {
        let executor = RecordingExecutor {
            statuses: HashMap::new(),
            delay: Duration::from_millis(0),
            running: Arc::new(AtomicUsize::new(0)),
            max_seen: Arc::new(AtomicUsize::new(0)),
            cancel_calls: Arc::new(Mutex::new(Vec::new())),
        };
        let max_seen = Arc::clone(&executor.max_seen);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(2);
        plan.options.max_concurrency = 2;
        plan.options.adaptive_concurrency = Some(AgentTaskAdaptiveConcurrencyPolicy {
            runner_capacity: Some(1),
            active_leases: 1,
            ..AgentTaskAdaptiveConcurrencyPolicy::default()
        });

        let aggregate = scheduler.run(plan);
        let adaptive = aggregate
            .queue
            .adaptive_concurrency
            .expect("adaptive status");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Failed
        );
        assert_eq!(aggregate.totals.blocked, 2);
        assert_eq!(max_seen.load(Ordering::SeqCst), 0);
        assert_eq!(adaptive.effective_concurrency, 0);
        assert!(adaptive.decisions.iter().any(|decision| {
            decision.action == AgentTaskAdaptiveConcurrencyAction::Paused
                && decision.reason.contains("consume runner_capacity=1")
        }));
        assert!(aggregate
            .queue
            .backpressure
            .iter()
            .any(|status| status.kind == "adaptive_concurrency"));
    }

    #[test]
    fn adaptive_concurrency_status_records_held_decision() {
        let scheduler = AgentTaskScheduler::new(RecordingExecutor::new(
            HashMap::new(),
            Duration::from_millis(0),
        ));
        let mut plan = plan_with_tasks(1);
        plan.options.max_concurrency = 2;
        plan.options.adaptive_concurrency = Some(AgentTaskAdaptiveConcurrencyPolicy::default());

        let aggregate = scheduler.run(plan);
        let adaptive = aggregate
            .queue
            .adaptive_concurrency
            .expect("adaptive status");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(adaptive.effective_concurrency, 2);
        assert!(adaptive.decisions.iter().any(|decision| {
            decision.action == AgentTaskAdaptiveConcurrencyAction::Held
                && decision.reason.contains("configured ceiling")
        }));
    }
}

mod concurrency_tests {
    use super::*;

    #[test]
    fn schedules_tasks_with_bounded_concurrency_and_success_aggregate() {
        let executor = RecordingExecutor::new(HashMap::new(), Duration::from_millis(25));
        let max_seen = Arc::clone(&executor.max_seen);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(4);
        plan.options.max_concurrency = 2;

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(aggregate.totals.queued, 0);
        assert_eq!(aggregate.totals.succeeded, 4);
        assert!(max_seen.load(Ordering::SeqCst) <= 2);
        assert!(aggregate
            .events
            .iter()
            .any(|event| event.state == AgentTaskState::Running));
    }

    #[test]
    fn blocks_tasks_over_queue_depth_and_reports_backpressure() {
        let scheduler = AgentTaskScheduler::new(RecordingExecutor::new(
            HashMap::new(),
            Duration::from_millis(0),
        ));
        let mut plan = plan_with_tasks(3);
        plan.options.max_queue_depth = Some(2);

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::PartialFailure
        );
        assert_eq!(aggregate.totals.succeeded, 2);
        assert_eq!(aggregate.totals.blocked, 1);
        assert_eq!(aggregate.queue.blocked, 1);
        assert_eq!(aggregate.queue.max_queue_depth, Some(2));
        assert!(aggregate
            .queue
            .backpressure
            .iter()
            .any(|status| status.kind == "queue_depth"));
        assert!(aggregate
            .events
            .iter()
            .any(|event| { event.task_id == "task-3" && event.state == AgentTaskState::Blocked }));
    }

    #[test]
    fn applies_per_executor_concurrency_below_global_limit() {
        let executor = RecordingExecutor::new(HashMap::new(), Duration::from_millis(25));
        let max_seen = Arc::clone(&executor.max_seen);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(3);
        plan.options.max_concurrency = 3;
        plan.options
            .per_executor_concurrency
            .insert("test".to_string(), 1);

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert!(max_seen.load(Ordering::SeqCst) <= 1);
        assert_eq!(
            aggregate.queue.per_executor_concurrency.get("test"),
            Some(&1)
        );
    }

    #[test]
    fn applies_per_model_concurrency_below_global_limit() {
        let executor = RecordingExecutor::new(HashMap::new(), Duration::from_millis(25));
        let max_seen = Arc::clone(&executor.max_seen);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(3);
        for task in &mut plan.tasks {
            task.executor.model = Some("model-a".to_string());
        }
        plan.options.max_concurrency = 3;
        plan.options
            .per_model_concurrency
            .insert("test:model-a".to_string(), 1);

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert!(max_seen.load(Ordering::SeqCst) <= 1);
        assert_eq!(
            aggregate.queue.per_model_concurrency.get("test:model-a"),
            Some(&1)
        );
    }
}

mod resource_budget_tests {
    use super::*;

    #[test]
    fn resource_budget_limits_concurrent_task_cost() {
        let executor = RecordingExecutor::new(HashMap::new(), Duration::from_millis(25));
        let max_seen = Arc::clone(&executor.max_seen);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(4);
        plan.options.max_concurrency = 4;
        plan.options.resource_budget.max_active_units = Some(2);
        plan.options.resource_budget.default_task_units = 1;

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(aggregate.totals.succeeded, 4);
        assert!(max_seen.load(Ordering::SeqCst) <= 2);
        assert_eq!(aggregate.queue.resource_budget.max_active_units, Some(2));
        assert_eq!(aggregate.queue.resource_budget.default_task_units, 1);
    }

    #[test]
    fn resource_budget_blocks_task_that_cannot_fit() {
        let scheduler = AgentTaskScheduler::new(RecordingExecutor::new(
            HashMap::new(),
            Duration::from_millis(0),
        ));
        let mut plan = plan_with_tasks(1);
        plan.options.resource_budget.max_active_units = Some(2);
        plan.options.resource_budget.default_task_units = 3;

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Failed
        );
        assert_eq!(aggregate.totals.blocked, 1);
        assert_eq!(aggregate.queue.blocked, 1);
        assert!(aggregate
            .queue
            .backpressure
            .iter()
            .any(|status| status.kind == "resource_budget"));
    }

    #[test]
    fn retry_budget_and_failure_classifications_gate_retries() {
        let executor = RetryOnceExecutor::default();
        let attempts = Arc::clone(&executor.attempts);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(1);
        plan.options.retry.max_attempts = 3;
        plan.options.retry.max_retries_total = Some(0);
        plan.options.retry.retryable_failure_classifications =
            vec![AgentTaskFailureClassification::Provider];

        let aggregate = scheduler.run(plan);

        assert_eq!(aggregate.status, AgentTaskAggregateStatus::Failed);
        assert_eq!(aggregate.totals.failed, 1);
        assert_eq!(attempts.load(Ordering::SeqCst), 1);
        assert_eq!(aggregate.queue.retry_budget_remaining, Some(0));
    }
}

mod timeout_tests {
    use super::*;

    #[test]
    fn normalizes_slow_task_to_timeout() {
        let scheduler = AgentTaskScheduler::new(RecordingExecutor::new(
            HashMap::new(),
            Duration::from_millis(25),
        ));
        let mut plan = plan_with_tasks(1);
        plan.tasks[0].limits.timeout_ms = Some(1);

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Failed
        );
        assert_eq!(aggregate.totals.timed_out, 1);
        assert_eq!(
            aggregate.outcomes[0].status,
            AgentTaskOutcomeStatus::Timeout
        );
        assert_eq!(
            aggregate.outcomes[0].failure_classification,
            Some(AgentTaskFailureClassification::Timeout)
        );
    }

    #[test]
    fn timeout_with_completed_runtime_artifacts_is_discoverable_and_promotable() {
        let temp = tempfile::tempdir().expect("tempdir");
        let artifact_root = temp.path().join("task-1-artifacts");
        fs::create_dir_all(&artifact_root).expect("artifact root");
        let patch_path = artifact_root.join("fix.patch");
        fs::write(&patch_path, "diff --git a/a.txt b/a.txt\n").expect("patch");
        fs::write(artifact_root.join("transcript.log"), "runtime completed").expect("log");
        let agent_result_path = artifact_root.join("agent-result.json");
        fs::write(
            &agent_result_path,
            serde_json::to_string(&AgentTaskOutcome {
                schema: AGENT_TASK_OUTCOME_SCHEMA.to_string(),
                task_id: "task-1".to_string(),
                status: AgentTaskOutcomeStatus::Succeeded,
                summary: Some("patch ready".to_string()),
                failure_classification: None,
                artifacts: vec![AgentTaskArtifact {
                    schema: AGENT_TASK_ARTIFACT_SCHEMA.to_string(),
                    id: "fix".to_string(),
                    kind: "patch".to_string(),
                    name: Some("fix.patch".to_string()),
                    label: None,
                    role: None,
                    semantic_key: None,
                    path: Some(patch_path.display().to_string()),
                    url: None,
                    mime: Some("text/x-patch".to_string()),
                    size_bytes: None,
                    sha256: None,
                    metadata: json!({ "role": "patch" }),
                }],
                typed_artifacts: Vec::new(),
                evidence_refs: vec![AgentTaskEvidenceRef {
                    kind: "runtime_bundle".to_string(),
                    uri: artifact_root.display().to_string(),
                    label: Some("runtime bundle".to_string()),
                }],
                diagnostics: Vec::new(),
                outputs: Value::Null,
                workflow: None,
                follow_up: None,
                metadata: json!({}),
            })
            .expect("agent result json"),
        )
        .expect("agent result");

        let scheduler = AgentTaskScheduler::new(RecordingExecutor::new(
            HashMap::new(),
            Duration::from_millis(250),
        ));
        let mut plan = plan_with_tasks(1);
        plan.tasks[0].limits.timeout_ms = Some(1);
        plan.tasks[0].metadata = json!({ "artifact_root": artifact_root });

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(aggregate.totals.succeeded, 1);
        assert_eq!(aggregate.totals.timed_out, 0);
        assert!(aggregate
            .events
            .iter()
            .any(|event| event.task_id == "task-1" && event.state == AgentTaskState::Succeeded));
        let outcome = &aggregate.outcomes[0];
        assert_eq!(outcome.status, AgentTaskOutcomeStatus::Succeeded);
        assert!(outcome.artifacts.iter().any(|artifact| {
            artifact.kind == "patch"
                && artifact.path.as_deref() == Some(&patch_path.to_string_lossy())
        }));
        assert!(outcome
            .artifacts
            .iter()
            .any(|artifact| artifact.kind == "transcript"));
        assert!(outcome
            .evidence_refs
            .iter()
            .any(|evidence| evidence.kind == "agent_result"
                && evidence.uri == agent_result_path.display().to_string()));
        assert!(outcome.diagnostics.iter().any(|diagnostic| {
            diagnostic.class == "completed_runtime_late_provider_race"
                && diagnostic.data.get("timeout_kind").and_then(Value::as_str)
                    == Some("scheduler_timeout")
                && diagnostic
                    .data
                    .get("actionable_patch")
                    .and_then(Value::as_bool)
                    == Some(true)
        }));
    }

    #[test]
    fn timeout_with_empty_patch_artifacts_and_actionable_false_stays_timed_out() {
        let temp = tempfile::tempdir().expect("tempdir");
        let artifact_root = temp.path().join("task-1-artifacts");
        fs::create_dir_all(&artifact_root).expect("artifact root");
        let patch_path = artifact_root.join("patch.diff");
        let mounted_patch_path = artifact_root.join("mount-5.patch");
        fs::write(&patch_path, "").expect("patch diff");
        fs::write(&mounted_patch_path, "").expect("mounted patch");
        fs::write(artifact_root.join("transcript.log"), "runtime completed").expect("log");
        fs::write(
            artifact_root.join("agent-result.json"),
            serde_json::to_string(&json!({
                "schema": AGENT_TASK_OUTCOME_SCHEMA,
                "task_id": "task-1",
                "status": "succeeded",
                "summary": "runtime produced no actionable patch",
                "actionable": false,
                "artifacts": [
                    {
                        "schema": AGENT_TASK_ARTIFACT_SCHEMA,
                        "id": "patch",
                        "kind": "patch",
                        "name": "patch.diff",
                        "path": patch_path.display().to_string(),
                        "mime": "text/x-diff",
                        "metadata": { "role": "patch" }
                    },
                    {
                        "schema": AGENT_TASK_ARTIFACT_SCHEMA,
                        "id": "mount-5",
                        "kind": "patch",
                        "name": "mount-5.patch",
                        "path": mounted_patch_path.display().to_string(),
                        "mime": "text/x-patch",
                        "metadata": { "role": "patch" }
                    }
                ],
                "evidence_refs": [],
                "diagnostics": [],
                "metadata": {}
            }))
            .expect("agent result json"),
        )
        .expect("agent result");

        let scheduler = AgentTaskScheduler::new(RecordingExecutor::new(
            HashMap::new(),
            Duration::from_millis(250),
        ));
        let mut plan = plan_with_tasks(1);
        plan.tasks[0].limits.timeout_ms = Some(1);
        plan.tasks[0].metadata = json!({ "artifact_root": artifact_root });

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Failed
        );
        assert_eq!(aggregate.totals.succeeded, 0);
        assert_eq!(aggregate.totals.timed_out, 1);
        assert!(aggregate
            .events
            .iter()
            .any(|event| event.task_id == "task-1" && event.state == AgentTaskState::TimedOut));
        let outcome = &aggregate.outcomes[0];
        assert_eq!(outcome.status, AgentTaskOutcomeStatus::Timeout);
        assert_eq!(
            outcome.failure_classification,
            Some(AgentTaskFailureClassification::Timeout)
        );
        assert!(outcome.artifacts.iter().any(|artifact| {
            artifact.kind == "patch"
                && artifact.path.as_deref() == Some(&patch_path.to_string_lossy())
        }));
        assert!(outcome.diagnostics.iter().any(|diagnostic| {
            diagnostic.class == "completed_runtime_late_provider_race"
                && diagnostic
                    .data
                    .get("actionable_patch")
                    .and_then(Value::as_bool)
                    == Some(false)
        }));
    }
}

mod artifact_binding_tests {
    use super::*;

    #[test]
    fn runtime_bundle_artifacts_materialize_required_typed_artifacts() {
        let temp = tempfile::tempdir().expect("tempdir");
        let bundle = temp.path().join("runtime-123");
        let files = bundle.join("files");
        fs::create_dir_all(&files).expect("runtime files");
        let patch_path = files.join("patch.diff");
        let transcript_path = files.join("transcript.json");
        fs::write(&patch_path, "diff --git a/a.txt b/a.txt\n").expect("patch");
        fs::write(&transcript_path, "{\"events\":[]}").expect("transcript");

        let scheduler = AgentTaskScheduler::new(RuntimeBundleOutcomeExecutor {
            patch_path: patch_path.clone(),
            transcript_path: transcript_path.clone(),
        });
        let mut plan = plan_with_tasks(1);
        plan.tasks[0].expected_artifacts = vec![
            "patch".to_string(),
            "agent_result".to_string(),
            "transcript".to_string(),
        ];

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(aggregate.totals.succeeded, 1);
        let outcome = &aggregate.outcomes[0];
        assert_eq!(outcome.status, AgentTaskOutcomeStatus::Succeeded);
        for name in ["patch", "agent_result", "transcript"] {
            assert!(
                outcome
                    .typed_artifacts
                    .iter()
                    .any(|artifact| artifact.name == name),
                "missing typed artifact {name}"
            );
        }
        let patch = outcome
            .typed_artifacts
            .iter()
            .find(|artifact| artifact.name == "patch")
            .expect("patch typed artifact");
        assert_eq!(
            patch
                .artifact
                .as_ref()
                .and_then(|artifact| artifact.path.as_deref()),
            Some(patch_path.to_str().expect("patch path"))
        );
        assert!(outcome.diagnostics.iter().any(|diagnostic| {
            diagnostic.class == "agent_task.required_typed_artifacts_normalized"
        }));
    }

    #[test]
    fn templates_prior_output_into_downstream_task_request() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(OutputTemplateExecutor {
            observed: Arc::clone(&observed),
            include_issue_number: true,
        });
        let mut plan =
            AgentTaskPlan::new("plan-output-dag", vec![request("idea"), request("design")]);
        plan.options.max_concurrency = 2;
        plan.tasks[1].instructions = "Build design for issue #{{outputs.issue_number}}".to_string();
        plan.tasks[1].executor.config = json!({
            "github_issue": "{{outputs.issue_number}}",
            "instructions": "Use issue {{ outputs.issue_number }}"
        });
        plan.output_dependencies.insert(
            "design".to_string(),
            AgentTaskOutputDependencies {
                depends_on: Vec::new(),
                bindings: HashMap::from([(
                    "issue_number".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: "/outputs/issue_number".to_string(),
                        artifact: None,
                        required: true,
                        default: Value::Null,
                    },
                )]),
            },
        );

        let aggregate = scheduler.run(plan);
        let observed = observed.lock().expect("observed requests");
        let design = observed
            .iter()
            .find(|request| request.task_id == "design")
            .expect("design request dispatched");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(aggregate.totals.succeeded, 2);
        assert_eq!(design.instructions, "Build design for issue #3447");
        assert_eq!(design.executor.config["github_issue"], json!(3447));
        assert_eq!(design.executor.config["instructions"], "Use issue 3447");
        assert_eq!(design.metadata["generated_from_outputs"], json!(true));
        assert_eq!(
            design.metadata["resolved_output_bindings"]["issue_number"],
            json!(3447)
        );
        let idea_succeeded_index = aggregate
            .events
            .iter()
            .position(|event| event.task_id == "idea" && event.state == AgentTaskState::Succeeded)
            .expect("idea succeeded event");
        let design_running_index = aggregate
            .events
            .iter()
            .position(|event| event.task_id == "design" && event.state == AgentTaskState::Running)
            .expect("design running event");
        assert!(idea_succeeded_index < design_running_index);
    }

    #[test]
    fn binds_typed_artifact_payload_into_downstream_task_request() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(OutputTemplateExecutor {
            observed: Arc::clone(&observed),
            include_issue_number: true,
        });
        let mut plan = AgentTaskPlan::new(
            "plan-artifact-dag",
            vec![request("idea"), request("design")],
        );
        plan.options.max_concurrency = 2;
        plan.tasks[1].inputs = json!({ "packet": "{{outputs.concept_packet}}" });
        plan.artifact_outputs.insert(
            "idea".to_string(),
            vec![AgentTaskArtifactOutputDeclaration {
                name: "concept_packet".to_string(),
                kind: "concept_packet".to_string(),
                schema: Some("example/concept-packet/v1".to_string()),
                artifact_id: None,
                payload_path: Some("/title".to_string()),
            }],
        );
        plan.output_dependencies.insert(
            "design".to_string(),
            AgentTaskOutputDependencies {
                depends_on: Vec::new(),
                bindings: HashMap::from([(
                    "concept_packet".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: String::new(),
                        artifact: Some(AgentTaskArtifactBinding {
                            kind: "concept_packet".to_string(),
                            schema: Some("example/concept-packet/v1".to_string()),
                            artifact_id: Some("concept".to_string()),
                            payload_path: Some("/title".to_string()),
                        }),
                        required: true,
                        default: Value::Null,
                    },
                )]),
            },
        );

        let aggregate = scheduler.run(plan);
        let observed = observed.lock().expect("observed requests");
        let design = observed
            .iter()
            .find(|request| request.task_id == "design")
            .expect("design request dispatched");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(design.inputs["packet"], json!("Demo concept"));
        assert_eq!(aggregate.artifact_lineage.len(), 1);
        assert_eq!(aggregate.artifact_lineage[0].name, "concept_packet");
        assert_eq!(aggregate.artifact_lineage[0].payload, json!("Demo concept"));
    }

    #[test]
    fn required_concept_packet_binding_uses_canonical_typed_artifact() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(ConceptPacketExecutor {
            observed: Arc::clone(&observed),
            emit_concept_packet: true,
        });
        let mut plan = AgentTaskPlan::new(
            "plan-concept-packet-typed-artifact",
            vec![request("idea"), request("build")],
        );
        plan.options.max_concurrency = 2;
        plan.tasks[0].artifact_declarations = vec![concept_packet_declaration()];
        plan.tasks[1].inputs = json!({ "concept_packet": "{{outputs.concept_packet}}" });
        plan.output_dependencies.insert(
            "build".to_string(),
            AgentTaskOutputDependencies {
                depends_on: Vec::new(),
                bindings: HashMap::from([(
                    "concept_packet".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: String::new(),
                        artifact: Some(AgentTaskArtifactBinding {
                            kind: "concept_packet".to_string(),
                            schema: Some("wp-site-generator/ConceptPacket/v1".to_string()),
                            artifact_id: None,
                            payload_path: None,
                        }),
                        required: true,
                        default: Value::Null,
                    },
                )]),
            },
        );

        let aggregate = scheduler.run(plan);
        let observed = observed.lock().expect("observed requests");
        let build = observed
            .iter()
            .find(|request| request.task_id == "build")
            .expect("build request dispatched");

        assert_eq!(aggregate.status, AgentTaskAggregateStatus::Succeeded);
        assert_eq!(aggregate.totals.succeeded, 2);
        assert_eq!(build.inputs["concept_packet"]["title"], "Typed concept");
    }

    #[test]
    fn required_concept_packet_binding_fails_without_canonical_typed_artifact() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(ConceptPacketExecutor {
            observed: Arc::clone(&observed),
            emit_concept_packet: false,
        });
        let mut plan = AgentTaskPlan::new(
            "plan-concept-packet-missing-typed-artifact",
            vec![request("idea"), request("build")],
        );
        plan.options.max_concurrency = 2;
        plan.tasks[0].artifact_declarations = vec![concept_packet_declaration()];
        plan.output_dependencies.insert(
            "build".to_string(),
            AgentTaskOutputDependencies {
                depends_on: Vec::new(),
                bindings: HashMap::from([(
                    "concept_packet".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: String::new(),
                        artifact: Some(AgentTaskArtifactBinding {
                            kind: "concept_packet".to_string(),
                            schema: Some("wp-site-generator/ConceptPacket/v1".to_string()),
                            artifact_id: None,
                            payload_path: None,
                        }),
                        required: true,
                        default: Value::Null,
                    },
                )]),
            },
        );

        let aggregate = scheduler.run(plan);
        let observed = observed.lock().expect("observed requests");

        assert_eq!(aggregate.status, AgentTaskAggregateStatus::Failed);
        assert!(observed.iter().all(|request| request.task_id != "build"));
        let idea = aggregate
            .outcomes
            .iter()
            .find(|outcome| outcome.task_id == "idea")
            .expect("idea outcome");
        assert!(idea.diagnostics.iter().any(|diagnostic| {
            diagnostic.class == "agent_task.required_typed_artifacts_missing"
                && diagnostic.message.contains("concept_packet")
        }));
        let build = aggregate
            .outcomes
            .iter()
            .find(|outcome| outcome.task_id == "build")
            .expect("build skipped outcome");
        assert!(build.diagnostics.iter().any(|diagnostic| {
            diagnostic.class == "output_dependency_missing"
                && diagnostic.message.contains("required artifact binding")
                && diagnostic.message.contains("concept_packet")
        }));
    }

    #[test]
    fn binds_artifacts_to_generic_child_run_ids_for_durable_fanout() {
        let scheduler = AgentTaskScheduler::new(GenericChildRunExecutor);
        let mut plan = AgentTaskPlan::new(
            "fuzz/campaign-1",
            vec![request("case-a"), request("case-b")],
        );
        plan.options.max_concurrency = 2;

        let aggregate = scheduler.run(plan);

        assert_eq!(aggregate.status, AgentTaskAggregateStatus::Succeeded);
        assert_eq!(aggregate.child_runs.len(), 2);
        let mut child_runs = aggregate.child_runs.clone();
        child_runs.sort_by(|left, right| left.task_id.cmp(&right.task_id));
        assert_eq!(child_runs[0].task_id, "case-a");
        assert_eq!(child_runs[0].run_id, "child-case-a");
        assert_eq!(
            child_runs[0].provider.as_deref(),
            Some("generic-fuzz")
        );
        assert_eq!(child_runs[0].state, AgentTaskState::Succeeded);
        assert_eq!(aggregate.artifact_bindings.len(), 2);
        let mut artifact_bindings = aggregate.artifact_bindings.clone();
        artifact_bindings.sort_by(|left, right| left.task_id.cmp(&right.task_id));
        assert_eq!(artifact_bindings[0].task_id, "case-a");
        assert_eq!(artifact_bindings[0].run_id, "child-case-a");
        assert_eq!(
            artifact_bindings[0].artifact_id,
            "artifact-case-a"
        );
        assert_eq!(artifact_bindings[0].kind, "fuzz-report");
        assert_eq!(
            artifact_bindings[0].path.as_deref(),
            Some("artifacts/case-a/report.json")
        );
    }

    #[test]
    fn skips_required_typed_artifact_binding_when_artifact_is_missing() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(OutputTemplateExecutor {
            observed: Arc::clone(&observed),
            include_issue_number: false,
        });
        let mut plan = AgentTaskPlan::new(
            "plan-artifact-skip",
            vec![request("idea"), request("design")],
        );
        plan.output_dependencies.insert(
            "design".to_string(),
            AgentTaskOutputDependencies {
                depends_on: Vec::new(),
                bindings: HashMap::from([(
                    "finding_packet".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: String::new(),
                        artifact: Some(AgentTaskArtifactBinding {
                            kind: "finding_packet".to_string(),
                            schema: None,
                            artifact_id: None,
                            payload_path: None,
                        }),
                        required: true,
                        default: Value::Null,
                    },
                )]),
            },
        );

        let aggregate = scheduler.run(plan);
        let observed = observed.lock().expect("observed requests");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::PartialFailure
        );
        assert!(observed.iter().all(|request| request.task_id != "design"));
        let skipped = aggregate
            .outcomes
            .iter()
            .find(|outcome| outcome.task_id == "design")
            .expect("skipped outcome");
        assert!(skipped.diagnostics.iter().any(|diagnostic| {
            diagnostic.class == "output_dependency_missing"
                && diagnostic.message.contains("required artifact binding")
        }));
    }

    #[test]
    fn optional_typed_artifact_binding_uses_default() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(OutputTemplateExecutor {
            observed: Arc::clone(&observed),
            include_issue_number: false,
        });
        let mut plan = AgentTaskPlan::new(
            "plan-artifact-default",
            vec![request("idea"), request("design")],
        );
        plan.tasks[1].inputs = json!({ "packet": "{{outputs.finding_packet}}" });
        plan.output_dependencies.insert(
            "design".to_string(),
            AgentTaskOutputDependencies {
                depends_on: Vec::new(),
                bindings: HashMap::from([(
                    "finding_packet".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: String::new(),
                        artifact: Some(AgentTaskArtifactBinding {
                            kind: "finding_packet".to_string(),
                            schema: None,
                            artifact_id: None,
                            payload_path: None,
                        }),
                        required: false,
                        default: json!({ "findings": [] }),
                    },
                )]),
            },
        );

        let aggregate = scheduler.run(plan);
        let observed = observed.lock().expect("observed requests");
        let design = observed
            .iter()
            .find(|request| request.task_id == "design")
            .expect("design request dispatched");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(design.inputs["packet"], json!({ "findings": [] }));
    }

    #[test]
    fn skips_downstream_task_when_required_output_is_missing() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(OutputTemplateExecutor {
            observed: Arc::clone(&observed),
            include_issue_number: false,
        });
        let mut plan =
            AgentTaskPlan::new("plan-output-skip", vec![request("idea"), request("design")]);
        plan.options.max_concurrency = 2;
        plan.tasks[1].instructions = "Build design for issue #{{outputs.issue_number}}".to_string();
        plan.output_dependencies.insert(
            "design".to_string(),
            AgentTaskOutputDependencies {
                depends_on: Vec::new(),
                bindings: HashMap::from([(
                    "issue_number".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: "/outputs/issue_number".to_string(),
                        artifact: None,
                        required: true,
                        default: Value::Null,
                    },
                )]),
            },
        );

        let aggregate = scheduler.run(plan);
        let observed = observed.lock().expect("observed requests");

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::PartialFailure
        );
        assert_eq!(aggregate.totals.succeeded, 1);
        assert_eq!(aggregate.totals.skipped, 1);
        assert_eq!(aggregate.totals.failed, 0);
        assert!(observed.iter().all(|request| request.task_id != "design"));
        assert!(aggregate
            .events
            .iter()
            .any(|event| { event.task_id == "design" && event.state == AgentTaskState::Skipped }));
        let skipped = aggregate
            .outcomes
            .iter()
            .find(|outcome| outcome.task_id == "design")
            .expect("skipped outcome");
        assert_eq!(skipped.status, AgentTaskOutcomeStatus::Failed);
        assert_eq!(
            skipped.failure_classification,
            Some(AgentTaskFailureClassification::InvalidInput)
        );
        assert!(skipped.diagnostics.iter().any(|diagnostic| {
            diagnostic.class == "output_dependency_missing"
                && diagnostic.message.contains("required output binding")
        }));
    }
}

mod retry_failure_tests {
    use super::*;

    #[test]
    fn preserves_partial_failure_evidence() {
        let mut statuses = HashMap::new();
        statuses.insert("task-2".to_string(), AgentTaskOutcomeStatus::Failed);
        let scheduler =
            AgentTaskScheduler::new(RecordingExecutor::new(statuses, Duration::from_millis(0)));
        let mut plan = plan_with_tasks(3);
        plan.options.max_concurrency = 3;

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::PartialFailure
        );
        assert_eq!(aggregate.totals.queued, 0);
        assert_eq!(aggregate.totals.succeeded, 2);
        assert_eq!(aggregate.totals.failed, 1);
        let failed = aggregate
            .outcomes
            .iter()
            .find(|outcome| outcome.task_id == "task-2")
            .expect("failed task outcome");
        assert_eq!(failed.evidence_refs[0].kind, "log");
    }

    #[test]
    fn failed_single_task_is_not_also_counted_as_queued() {
        let mut statuses = HashMap::new();
        statuses.insert("task-1".to_string(), AgentTaskOutcomeStatus::Failed);
        let scheduler =
            AgentTaskScheduler::new(RecordingExecutor::new(statuses, Duration::from_millis(0)));

        let aggregate = scheduler.run(plan_with_tasks(1));

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Failed
        );
        assert_eq!(aggregate.totals.failed, 1);
        assert_eq!(aggregate.totals.queued, 0);
        assert_eq!(aggregate.queue.queued, 0);
    }

    #[test]
    fn retries_failed_tasks_until_success() {
        let executor = RetryOnceExecutor::default();
        let attempts = Arc::clone(&executor.attempts);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(1);
        plan.options.retry.max_attempts = 2;

        let aggregate = scheduler.run(plan);

        assert_eq!(aggregate.status, AgentTaskAggregateStatus::Succeeded);
        assert_eq!(aggregate.totals.succeeded, 1);
        assert_eq!(attempts.load(Ordering::SeqCst), 2);
        assert!(aggregate.events.iter().any(|event| {
            event.task_id == "task-1" && event.state == AgentTaskState::Queued && event.attempt == 2
        }));
    }
}

mod cancellation_tests {
    use super::*;

    #[test]
    fn cancellation_token_callbacks_fire_once_and_after_existing_cancel() {
        let token = AgentTaskCancellationToken::default();
        let callback_count = Arc::new(AtomicUsize::new(0));
        let callback_count_for_token = Arc::clone(&callback_count);
        token.on_cancel(Arc::new(move || {
            callback_count_for_token.fetch_add(1, Ordering::SeqCst);
        }));

        token.cancel();
        token.cancel();

        assert_eq!(callback_count.load(Ordering::SeqCst), 1);

        let immediate_count = Arc::new(AtomicUsize::new(0));
        let immediate_count_for_token = Arc::clone(&immediate_count);
        token.on_cancel(Arc::new(move || {
            immediate_count_for_token.fetch_add(1, Ordering::SeqCst);
        }));

        assert_eq!(immediate_count.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn cancellation_stops_queued_tasks_and_notifies_running_executor() {
        let executor = RecordingExecutor::new(HashMap::new(), Duration::from_millis(100));
        let cancel_calls = Arc::clone(&executor.cancel_calls);
        let running = Arc::clone(&executor.running);
        let scheduler = AgentTaskScheduler::new(executor);
        let mut plan = plan_with_tasks(3);
        plan.options.max_concurrency = 1;
        let token = AgentTaskCancellationToken::default();
        let worker_token = token.clone();

        let handle = thread::spawn(move || scheduler.run_with_cancellation(plan, worker_token));
        while running.load(Ordering::SeqCst) == 0 {
            thread::sleep(Duration::from_millis(1));
        }
        token.cancel();
        let aggregate = handle.join().expect("scheduler thread");

        assert_eq!(aggregate.status, AgentTaskAggregateStatus::Cancelled);
        assert!(aggregate.totals.cancelled >= 2);
        assert!(cancel_calls
            .lock()
            .expect("cancel calls")
            .contains(&"task-1".to_string()));
    }
}

mod plan_projection_tests {
    use super::*;

    #[test]
    fn runs_matrix_cells_through_generic_scheduler_and_preserves_axes() {
        let mut statuses = HashMap::new();
        statuses.insert(
            "fanout/site-smoke[model=gpt-5.5,prompt=site-b]".to_string(),
            AgentTaskOutcomeStatus::Failed,
        );
        let scheduler =
            AgentTaskScheduler::new(RecordingExecutor::new(statuses, Duration::from_millis(0)));
        let matrix_plan = expand_agent_task_matrix(
            "fanout/site-smoke",
            vec![
                AgentTaskMatrixAxis {
                    name: "model".to_string(),
                    values: vec!["gpt-5.5".to_string(), "claude".to_string()],
                },
                AgentTaskMatrixAxis {
                    name: "prompt".to_string(),
                    values: vec!["site-a".to_string(), "site-b".to_string()],
                },
            ],
            request("template"),
        )
        .expect("matrix expands");
        let mut schedule_plan = AgentTaskPlan::new(
            matrix_plan.plan_id.clone(),
            matrix_plan
                .cells
                .iter()
                .map(|cell| cell.task.clone())
                .collect(),
        );
        schedule_plan.options.max_concurrency = 2;

        let schedule = scheduler.run(schedule_plan);
        let matrix = AgentTaskMatrixAggregate::from_outcomes(&matrix_plan, &schedule.outcomes);

        assert_eq!(schedule.plan_id, "fanout/site-smoke");
        assert_eq!(schedule.totals.succeeded, 3);
        assert_eq!(schedule.totals.failed, 1);
        assert_eq!(matrix.cells.len(), 4);
        assert!(!matrix.passed);
        let failed = matrix
            .cells
            .iter()
            .find(|cell| cell.status == Some(AgentTaskOutcomeStatus::Failed))
            .expect("failed matrix cell");
        assert_eq!(failed.axes["model"], "gpt-5.5");
        assert_eq!(failed.axes["prompt"], "site-b");
        assert_eq!(failed.evidence_refs[0].kind, "log");
    }

    #[test]
    fn static_batch_plans_remain_compatible_without_output_dependencies() {
        let scheduler = AgentTaskScheduler::new(RecordingExecutor::new(
            HashMap::new(),
            Duration::from_millis(0),
        ));
        let plan_json = serde_json::to_string(&plan_with_tasks(2)).expect("plan json");
        let plan: AgentTaskPlan = serde_json::from_str(&plan_json).expect("static plan decodes");

        let aggregate = scheduler.run(plan);

        assert_eq!(
            aggregate.status,
            crate::core::agent_task_scheduler::AgentTaskAggregateStatus::Succeeded
        );
        assert_eq!(aggregate.totals.succeeded, 2);
        assert_eq!(aggregate.totals.skipped, 0);
        assert_eq!(aggregate.queue.max_concurrency, 1);
        assert!(aggregate.queue.adaptive_concurrency.is_none());
    }

    #[test]
    fn plan_level_component_contracts_are_preserved_on_executor_requests() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(OutputTemplateExecutor {
            observed: Arc::clone(&observed),
            include_issue_number: true,
        });
        let raw = serde_json::json!({
            "schema": AGENT_TASK_PLAN_SCHEMA,
            "plan_id": "plan-components",
            "component_contracts": [{
                "slug": "generic-component",
                "path": "/workspace/generic-component",
                "loadAs": "plugin",
                "activate": true,
                "opaque_executor_hint": { "preserve": true }
            }],
            "tasks": [{
                "task_id": "task-components",
                "executor": { "backend": "test" },
                "instructions": "run"
            }]
        });
        let plan: AgentTaskPlan = serde_json::from_value(raw).expect("plan parses");

        let aggregate = scheduler.run(plan);
        let observed = observed.lock().expect("observed requests");
        let request = observed.first().expect("request dispatched");

        assert_eq!(aggregate.status, AgentTaskAggregateStatus::Succeeded);
        assert_eq!(request.component_contracts.len(), 1);
        assert_eq!(
            request.component_contracts[0].slug.as_deref(),
            Some("generic-component")
        );
        assert_eq!(
            request.component_contracts[0].path.as_deref(),
            Some("/workspace/generic-component")
        );
        assert_eq!(
            request.component_contracts[0].load_as.as_deref(),
            Some("plugin")
        );
        assert_eq!(request.component_contracts[0].activate, Some(true));
        assert_eq!(
            request.component_contracts[0].extra["opaque_executor_hint"]["preserve"],
            true
        );
    }

    #[test]
    fn legacy_agent_task_plan_json_round_trips_through_homeboy_plan_projection() {
        let mut plan =
            AgentTaskPlan::new("plan-projection", vec![request("idea"), request("design")]);
        plan.group_key = Some("group-a".to_string());
        plan.options.max_concurrency = 2;
        plan.metadata = json!({ "source": "compat" });
        plan.output_dependencies.insert(
            "design".to_string(),
            AgentTaskOutputDependencies {
                depends_on: vec!["idea".to_string()],
                bindings: HashMap::from([(
                    "issue_number".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: "/outputs/issue_number".to_string(),
                        artifact: None,
                        required: true,
                        default: Value::Null,
                    },
                )]),
            },
        );
        plan.artifact_outputs.insert(
            "idea".to_string(),
            vec![AgentTaskArtifactOutputDeclaration {
                name: "concept_packet".to_string(),
                kind: "concept_packet".to_string(),
                schema: Some("example/concept-packet/v1".to_string()),
                artifact_id: Some("concept".to_string()),
                payload_path: Some("/title".to_string()),
            }],
        );

        let raw = serde_json::to_string(&plan).expect("serialize legacy contract");
        let value: Value = serde_json::from_str(&raw).expect("serialized json");
        assert_eq!(value["schema"], AGENT_TASK_PLAN_SCHEMA);
        assert!(value.get("homeboy_plan").is_none());

        let decoded: AgentTaskPlan = serde_json::from_str(&raw).expect("legacy plan decodes");
        let projected = AgentTaskPlan::from_homeboy_plan(decoded.homeboy_plan.clone());

        assert_eq!(
            decoded.homeboy_plan.kind,
            crate::core::plan::PlanKind::AgentTask
        );
        assert_eq!(projected.schema, AGENT_TASK_PLAN_SCHEMA);
        assert_eq!(projected.plan_id, plan.plan_id);
        assert_eq!(projected.group_key, plan.group_key);
        assert_eq!(projected.tasks, plan.tasks);
        assert_eq!(projected.output_dependencies, plan.output_dependencies);
        assert_eq!(projected.artifact_outputs, plan.artifact_outputs);
        assert_eq!(projected.options, plan.options);
        assert_eq!(projected.metadata, plan.metadata);
    }

    #[test]
    fn scheduler_executes_from_projected_homeboy_plan() {
        let observed = Arc::new(Mutex::new(Vec::new()));
        let scheduler = AgentTaskScheduler::new(OutputTemplateExecutor {
            observed: Arc::clone(&observed),
            include_issue_number: true,
        });
        let mut plan = AgentTaskPlan::new(
            "plan-homeboy-projection",
            vec![request("idea"), request("design")],
        );
        plan.options.max_concurrency = 2;
        plan.tasks[1].instructions = "Build design for issue #{{outputs.issue_number}}".to_string();
        plan.output_dependencies.insert(
            "design".to_string(),
            AgentTaskOutputDependencies {
                depends_on: vec!["idea".to_string()],
                bindings: HashMap::from([(
                    "issue_number".to_string(),
                    AgentTaskOutputBinding {
                        task_id: "idea".to_string(),
                        path: "/outputs/issue_number".to_string(),
                        artifact: None,
                        required: true,
                        default: Value::Null,
                    },
                )]),
            },
        );
        plan.rebuild_homeboy_plan();
        let projected = AgentTaskPlan::from_homeboy_plan(plan.homeboy_plan.clone());

        let aggregate = scheduler.run(projected);
        let observed = observed.lock().expect("observed requests");
        let design = observed
            .iter()
            .find(|request| request.task_id == "design")
            .expect("design request dispatched");

        assert_eq!(aggregate.status, AgentTaskAggregateStatus::Succeeded);
        assert_eq!(aggregate.totals.succeeded, 2);
        assert_eq!(design.instructions, "Build design for issue #3447");
    }
}

fn concept_packet_declaration() -> AgentTaskArtifactDeclaration {
    AgentTaskArtifactDeclaration {
        name: "concept_packet".to_string(),
        artifact_type: Some("concept_packet".to_string()),
        artifact_schema: Some("wp-site-generator/ConceptPacket/v1".to_string()),
        path: None,
        required: true,
        description: None,
        metadata: Value::Null,
    }
}

struct ConceptPacketExecutor {
    observed: Arc<Mutex<Vec<AgentTaskRequest>>>,
    emit_concept_packet: bool,
}

impl AgentTaskExecutorAdapter for ConceptPacketExecutor {
    fn execute(
        &self,
        request: AgentTaskRequest,
        _context: AgentTaskExecutionContext,
    ) -> AgentTaskOutcome {
        self.observed
            .lock()
            .expect("observed requests")
            .push(request.clone());

        AgentTaskOutcome {
            schema: AGENT_TASK_OUTCOME_SCHEMA.to_string(),
            task_id: request.task_id,
            status: AgentTaskOutcomeStatus::Succeeded,
            summary: Some("ok".to_string()),
            failure_classification: None,
            artifacts: Vec::new(),
            typed_artifacts: if self.emit_concept_packet {
                vec![AgentTaskTypedArtifact {
                    name: "concept_packet".to_string(),
                    artifact_type: Some("concept_packet".to_string()),
                    artifact_schema: Some("wp-site-generator/ConceptPacket/v1".to_string()),
                    payload: json!({ "title": "Typed concept" }),
                    artifact: None,
                    metadata: json!({ "source": "wp-codebox/artifact-result-envelope/v1" }),
                }]
            } else {
                Vec::new()
            },
            evidence_refs: Vec::new(),
            diagnostics: Vec::new(),
            outputs: Value::Null,
            workflow: None,
            follow_up: None,
            metadata: Value::Null,
        }
    }
}

struct GenericChildRunExecutor;

impl AgentTaskExecutorAdapter for GenericChildRunExecutor {
    fn execute(
        &self,
        request: AgentTaskRequest,
        _context: AgentTaskExecutionContext,
    ) -> AgentTaskOutcome {
        AgentTaskOutcome {
            schema: AGENT_TASK_OUTCOME_SCHEMA.to_string(),
            task_id: request.task_id.clone(),
            status: AgentTaskOutcomeStatus::Succeeded,
            summary: Some("generic fuzz case completed".to_string()),
            failure_classification: None,
            artifacts: vec![AgentTaskArtifact {
                schema: AGENT_TASK_ARTIFACT_SCHEMA.to_string(),
                id: format!("artifact-{}", request.task_id),
                kind: "fuzz-report".to_string(),
                name: Some("report.json".to_string()),
                label: Some("Fuzz report".to_string()),
                role: Some("fuzz_report".to_string()),
                semantic_key: Some("fuzz.report".to_string()),
                path: Some(format!("artifacts/{}/report.json", request.task_id)),
                url: None,
                mime: Some("application/json".to_string()),
                size_bytes: Some(512),
                sha256: Some(format!("sha256:{}", request.task_id)),
                metadata: json!({ "case_id": request.task_id }),
            }],
            typed_artifacts: Vec::new(),
            evidence_refs: Vec::new(),
            diagnostics: Vec::new(),
            outputs: json!({ "case_id": request.task_id }),
            workflow: None,
            follow_up: None,
            metadata: json!({
                "provider": "generic-fuzz",
                "child_run_id": format!("child-{}", request.task_id)
            }),
        }
    }
}
