use super::{
    decision::*, pr_state::*, runtime_feedback::*, signals::*, wait_budget::ReviewWaitBudget,
};
use crate::task_executor::agent_review::jaccard_word_similarity;
use crate::task_executor::helpers::{
    build_task_event, run_agent_streaming, run_on_error, run_post_execute, run_pre_execute,
    telemetry_for_timeout, update_status,
};
use crate::task_executor::run_test_gate;
use crate::task_runner::{
    mutate_and_persist, CreateTaskRequest, RoundResult, TaskId, TaskStatus, TaskStore,
};
use chrono::Utc;
use harness_core::agent::{AgentRequest, AgentResponse, CodeAgent};
use harness_core::prompts;
use harness_core::tool_isolation::validate_tool_usage;
use harness_core::types::{Decision, ExecutionPhase, TurnFailure, TurnFailureKind};
use harness_core::validation::ShellValidationExecutor;
use harness_workflow::runtime::PrFeedbackOutcome;
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use tokio::time::{sleep, Duration, Instant};
#[allow(clippy::too_many_arguments)]
pub(crate) async fn run_review_loop(
    store: &TaskStore,
    task_id: &TaskId,
    agent: &dyn CodeAgent,
    review_config: &harness_core::config::agents::AgentReviewConfig,
    project_config: &harness_core::config::project::ProjectConfig,
    issue_workflow_store: Option<&harness_workflow::issue_lifecycle::IssueWorkflowStore>,
    workflow_runtime_store: Option<&harness_workflow::runtime::WorkflowRuntimeStore>,
    project_root: &Path,
    req: &CreateTaskRequest,
    events: &Arc<harness_observe::event_store::EventStore>,
    interceptors: &Arc<Vec<Arc<dyn harness_core::interceptor::TurnInterceptor>>>,
    context_items: &[harness_core::types::ContextItem],
    project: &Path,
    cargo_env: &HashMap<String, String>,
    pr_url: Option<String>,
    pr_num: u64,
    effective_max_turns: Option<u32>,
    _effective_max_rounds: u32,
    wait_secs: u64,
    max_rounds: u32,
    agent_pushed_commit: bool,
    rebase_pushed: bool,
    turn_timeout: Duration,
    turns_used: &mut u32,
    turns_used_acc: &mut u32,
    task_start: Instant,
    review_wait_started_at: Instant,
    repo_slug: String,
    jaccard_threshold: f64,
    github_token: Option<&str>,
) -> anyhow::Result<()> {
    let review_phase_start = Instant::now();
    let mut prev_fixed = agent_pushed_commit || rebase_pushed;
    let mut issue_counts: Vec<Option<u32>> = Vec::new();
    let mut impasse = false;
    let mut pending_test_failure: Option<String> = None;
    let mut lgtm_test_gate_rejected = false;
    let mut prev_review_output: Option<String> = None;
    let mut silence_rounds = 0u32;
    let mut last_bot_activity_at = None;
    let mut waiting_on_bot = None;
    let bot_chain = bot_fallback_chain(review_config);
    let review_wait_budget = ReviewWaitBudget::new(
        review_wait_started_at,
        review_config.review_wait_budget_secs,
    );
    let mut round: u32 = 1;
    let mut waiting_count: u32 = 1;
    while round <= max_rounds {
        match fetch_pr_external_state(pr_num, project, github_token).await {
            PrExternalState::Merged => {
                tracing::info!(
                    task_id = %task_id,
                    pr = pr_num,
                    round,
                    "review loop exit: PR merged externally"
                );
                mutate_and_persist(store, task_id, |s| {
                    s.status = TaskStatus::Done;
                    s.turn = round;
                    s.error = Some(
                        "PR merged externally; review loop exited without waiting for LGTM"
                            .to_string(),
                    );
                })
                .await?;
                store.log_event(crate::event_replay::TaskEvent::Completed {
                    task_id: task_id.0.clone(),
                    ts: crate::event_replay::now_ts(),
                });
                tracing::info!(
                    task_id = %task_id,
                    phase = "reviewing",
                    elapsed_secs = review_phase_start.elapsed().as_secs(),
                    "phase_completed"
                );
                tracing::info!(
                    task_id = %task_id,
                    status = "done",
                    turns = round,
                    pr_url = pr_url.as_deref().unwrap_or(""),
                    total_elapsed_secs = task_start.elapsed().as_secs(),
                    "task_completed"
                );
                return Ok(());
            }
            PrExternalState::Closed => {
                tracing::info!(
                    task_id = %task_id,
                    pr = pr_num,
                    round,
                    "review loop exit: PR closed externally without merge"
                );
                mutate_and_persist(store, task_id, |s| {
                    s.status = TaskStatus::Cancelled;
                    s.turn = round;
                    s.error =
                        Some("PR closed externally without merge; review loop exited".to_string());
                })
                .await?;
                return Ok(());
            }
            PrExternalState::Open | PrExternalState::Unknown => {}
        }
        let mut decision = ReviewLoopDecision {
            active_bot: bot_chain
                .first()
                .cloned()
                .ok_or_else(|| anyhow::anyhow!("review fallback chain is empty"))?,
            fallback: None,
            wait_for_bot: false,
        };
        match fetch_pull_request_signals(&repo_slug, pr_num, github_token, &bot_chain).await {
            Ok(signals) => {
                (silence_rounds, last_bot_activity_at) = update_silence_rounds(
                    silence_rounds,
                    last_bot_activity_at,
                    signals.latest_bot_activity_at,
                );
                decision =
                    decide_review_loop_action(&bot_chain, &signals, silence_rounds, review_config)?;
                if let Some(fallback) = &decision.fallback {
                    let event = build_task_event(
                        task_id,
                        round,
                        "review",
                        "pr_review_fallback",
                        Decision::Warn,
                        Some(format!(
                            "tier={:?}; trigger={:?}; active_bot={}",
                            fallback.tier,
                            fallback.trigger,
                            fallback
                                .active_bot
                                .map(ReviewBotKey::as_str)
                                .unwrap_or("none")
                        )),
                        Some(format!("pr={pr_num}")),
                        None,
                        None,
                        None,
                    );
                    if let Err(error) = events.log(&event).await {
                        tracing::warn!("failed to log pr_review_fallback event: {error}");
                    }
                }
                if review_wait_budget
                    .fail_terminal_fallback_if_exceeded(&decision, store, task_id, round)
                    .await?
                {
                    return Ok(());
                }
                if decision
                    .fallback
                    .as_ref()
                    .is_some_and(ReviewFallbackState::is_terminal)
                {
                    let fallback = decision.fallback.expect("tier C fallback");
                    let detail = fallback.detail();
                    mutate_and_persist(store, task_id, |s| {
                        s.status = TaskStatus::Done;
                        s.turn = round;
                        s.error = Some(detail.clone());
                        s.rounds.push(RoundResult::new(
                            round,
                            "review",
                            "ready_to_merge",
                            Some(detail.clone()),
                            None,
                            None,
                        ));
                    })
                    .await?;
                    if let Some(workflows) = issue_workflow_store {
                        let project_id = project_root.to_string_lossy().into_owned();
                        let snapshot = harness_workflow::issue_lifecycle::ReviewFallbackSnapshot {
                            tier: fallback.tier,
                            trigger: fallback.trigger,
                            active_bot: fallback.active_bot.map(|bot| bot.as_str().to_string()),
                            activated_at: Utc::now(),
                        };
                        let _ = workflows
                            .record_ready_to_merge_with_fallback(
                                &project_id,
                                req.repo.as_deref(),
                                pr_num,
                                Some(&detail),
                                snapshot,
                            )
                            .await;
                    }
                    record_runtime_pr_feedback(
                        issue_workflow_store,
                        workflow_runtime_store,
                        project_root,
                        req,
                        task_id,
                        pr_num,
                        pr_url.as_deref(),
                        PrFeedbackOutcome::ReadyToMerge,
                        &detail,
                    )
                    .await;
                    store.log_event(crate::event_replay::TaskEvent::Completed {
                        task_id: task_id.0.clone(),
                        ts: crate::event_replay::now_ts(),
                    });
                    return Ok(());
                }
                if decision.wait_for_bot {
                    if review_wait_budget
                        .fail_if_exceeded(store, task_id, round)
                        .await?
                    {
                        return Ok(());
                    }
                    if decision.active_bot.key == ReviewBotKey::Codex
                        && waiting_on_bot != Some(decision.active_bot.key)
                    {
                        if let Err(error) = post_review_bot_comment(
                            &repo_slug,
                            pr_num,
                            &decision.active_bot.review_command,
                            github_token,
                        )
                        .await
                        {
                            tracing::warn!(pr = pr_num, "failed to summon Codex reviewer: {error}");
                        } else {
                            waiting_on_bot = Some(decision.active_bot.key);
                        }
                    }
                    tracing::info!(
                        pr = pr_num,
                        active_bot = decision.active_bot.key.as_str(),
                        silence_rounds,
                        "waiting for review bot activity"
                    );
                    record_runtime_pr_feedback(
                        issue_workflow_store,
                        workflow_runtime_store,
                        project_root,
                        req,
                        task_id,
                        pr_num,
                        pr_url.as_deref(),
                        PrFeedbackOutcome::NoActionableFeedback,
                        "Review bot has not produced actionable feedback yet.",
                    )
                    .await;
                    waiting_count = waiting_count.saturating_add(1);
                    if waiting_count >= MAX_CONSECUTIVE_WAITS {
                        round += 1;
                        waiting_count = 0;
                    }
                    update_status(store, task_id, TaskStatus::Waiting, waiting_count).await?;
                    sleep(review_wait_budget.sleep_duration(wait_secs)).await;
                    continue;
                }
                waiting_on_bot = None;
            }
            Err(error) => {
                waiting_on_bot = None;
                if review_wait_budget
                    .fail_if_exceeded(store, task_id, round)
                    .await?
                {
                    return Ok(());
                }
                tracing::warn!(
                    pr = pr_num,
                    error = %error,
                    "review loop: GitHub signal fetch failed, falling back to prompt-driven review"
                );
            }
        }
        update_status(store, task_id, TaskStatus::Reviewing, round).await?;
        let base_prompt = prompts::review_prompt(
            req.issue,
            pr_num,
            round,
            prev_fixed,
            &decision.active_bot.review_command,
            &decision.active_bot.reviewer_name,
            &repo_slug,
            impasse,
        );
        let round_prompt = if let Some(failure) = pending_test_failure.take() {
            prompts::test_gate_failure_prompt(&failure, &base_prompt)
        } else {
            base_prompt
        };
        let prompt_built_at = Utc::now();
        let check_req = AgentRequest {
            prompt: round_prompt,
            project_root: project.to_path_buf(),
            context: context_items.to_vec(),
            execution_phase: Some(ExecutionPhase::Execution),
            env_vars: cargo_env.clone(),
            ..Default::default()
        };
        let check_req = run_pre_execute(interceptors, check_req).await?;
        if let Some(max) = effective_max_turns {
            if *turns_used >= max {
                let msg = format!(
                    "Turn budget exhausted: used {} of {} allowed turns",
                    turns_used, max
                );
                tracing::warn!(task_id = %task_id, turns_used, max, "turn budget exhausted in review loop");
                mutate_and_persist(store, task_id, |s| {
                    s.status = TaskStatus::Failed;
                    s.error = Some(msg.clone());
                })
                .await?;
                return Ok(());
            }
        }
        let review_started_at = Utc::now();
        let resp = tokio::time::timeout(
            turn_timeout,
            run_agent_streaming(
                agent,
                check_req.clone(),
                task_id,
                store,
                round,
                prompt_built_at,
                review_started_at,
            ),
        )
        .await;
        *turns_used += 1;
        *turns_used_acc = *turns_used;
        let resp = match resp {
            Ok(Ok(success)) => {
                let r = success.response;
                let tool_violations = validate_tool_usage(
                    &r.output,
                    check_req.allowed_tools.as_deref().unwrap_or(&[]),
                );
                if !tool_violations.is_empty() {
                    let msg = format!(
                        "Tool isolation violation in review check round {round}: agent used disallowed tools: [{}]",
                        tool_violations.join(", ")
                    );
                    tracing::warn!(
                        round,
                        ?tool_violations,
                        "review check: agent used tools outside allowed list"
                    );
                    run_on_error(interceptors, &check_req, &msg).await;
                    return Err(anyhow::anyhow!("{msg}"));
                }
                if let Some(val_err) = run_post_execute(interceptors, &check_req, &r).await {
                    tracing::error!(
                        round,
                        error = %val_err,
                        "post-execute validation failed in review check; will re-enter review"
                    );
                    pending_test_failure =
                        Some(format!("Post-execution validation failed:\n{val_err}"));
                }
                (r, success.telemetry)
            }
            Ok(Err(failure)) => {
                if matches!(
                    failure.failure.kind,
                    TurnFailureKind::Quota | TurnFailureKind::Billing
                ) {
                    tracing::error!(round, error = %failure.error, "quota/billing failure during review — aborting review loop");
                    run_on_error(interceptors, &check_req, &failure.error.to_string()).await;
                    mutate_and_persist(store, task_id, |s| {
                        s.status = TaskStatus::Failed;
                        s.error = Some(failure.error.to_string());
                        s.rounds.push(RoundResult::new(
                            round,
                            "review",
                            match failure.failure.kind {
                                TurnFailureKind::Quota => "quota_exhausted",
                                TurnFailureKind::Billing => "billing_failed",
                                TurnFailureKind::Upstream => "upstream_failure",
                                _ => "failed",
                            },
                            None,
                            Some(failure.telemetry.clone()),
                            Some(failure.failure.clone()),
                        ));
                    })
                    .await?;
                    let event = build_task_event(
                        task_id,
                        round,
                        "review",
                        "pr_review",
                        Decision::Block,
                        Some("review round failed".to_string()),
                        Some(format!("pr={pr_num}")),
                        Some(failure.telemetry),
                        Some(failure.failure),
                        None,
                    );
                    if let Err(error) = events.log(&event).await {
                        tracing::warn!("failed to log pr_review event: {error}");
                    }
                    return Ok(());
                }
                run_on_error(interceptors, &check_req, &failure.error.to_string()).await;
                mutate_and_persist(store, task_id, |s| {
                    s.rounds.push(RoundResult::new(
                        round,
                        "review",
                        "failed",
                        None,
                        Some(failure.telemetry.clone()),
                        Some(failure.failure.clone()),
                    ));
                })
                .await?;
                let event = build_task_event(
                    task_id,
                    round,
                    "review",
                    "pr_review",
                    Decision::Block,
                    Some("review round failed".to_string()),
                    Some(format!("pr={pr_num}")),
                    Some(failure.telemetry),
                    Some(failure.failure),
                    None,
                );
                if let Err(error) = events.log(&event).await {
                    tracing::warn!("failed to log pr_review event: {error}");
                }
                return Err(failure.error.into());
            }
            Err(_) => {
                let msg = format!(
                    "Review check round {round} timed out after {}s",
                    turn_timeout.as_secs()
                );
                run_on_error(interceptors, &check_req, &msg).await;
                let telemetry =
                    telemetry_for_timeout(prompt_built_at, review_started_at, Utc::now(), None);
                let failure = TurnFailure {
                    kind: TurnFailureKind::Timeout,
                    provider: Some(agent.name().to_string()),
                    upstream_status: None,
                    message: Some(msg.clone()),
                    body_excerpt: None,
                };
                mutate_and_persist(store, task_id, |s| {
                    s.rounds.push(RoundResult::new(
                        round,
                        "review",
                        "timeout",
                        None,
                        Some(telemetry.clone()),
                        Some(failure.clone()),
                    ));
                })
                .await?;
                let event = build_task_event(
                    task_id,
                    round,
                    "review",
                    "pr_review",
                    Decision::Block,
                    Some("review round timed out".to_string()),
                    Some(format!("pr={pr_num}")),
                    Some(telemetry),
                    Some(failure),
                    None,
                );
                if let Err(error) = events.log(&event).await {
                    tracing::warn!("failed to log pr_review event: {error}");
                }
                return Err(anyhow::anyhow!("{msg}"));
            }
        };
        let (AgentResponse { output, stderr, .. }, review_telemetry) = resp;
        if !stderr.is_empty() {
            tracing::warn!(round, stderr = %stderr, "agent stderr during review check");
        }
        let raw_lgtm = prompts::is_lgtm(&output);
        let waiting = prompts::is_waiting(&output);
        let lgtm = raw_lgtm && pending_test_failure.is_none();
        let fixed = !lgtm && !waiting;
        let current_issues = prompts::parse_issue_count(&output);
        if !waiting && !raw_lgtm {
            if let Some(ref prev) = prev_review_output {
                let score = jaccard_word_similarity(prev, &output);
                if score >= jaccard_threshold {
                    let last_count = issue_counts.last().and_then(|x| *x);
                    let progressing = matches!(
                        (last_count, current_issues),
                        (Some(prev_n), Some(curr_n)) if curr_n < prev_n
                    );
                    if !progressing {
                        let msg = format!(
                            "review loop detected: output similarity {score:.2} >= threshold {jaccard_threshold:.2}"
                        );
                        tracing::warn!(task_id = %task_id, round, score, jaccard_threshold, "jaccard loop detected in review");
                        mutate_and_persist(store, task_id, |s| {
                            s.status = TaskStatus::Failed;
                            s.error = Some(msg.clone());
                        })
                        .await?;
                        return Ok(());
                    }
                }
            }
            prev_review_output = Some(output.clone());
        } else if raw_lgtm {
            prev_review_output = None;
        }
        if !waiting {
            issue_counts.push(current_issues);
        }
        if !impasse && issue_count_not_decreasing(&issue_counts) {
            impasse = true;
            let tail = &issue_counts[issue_counts.len() - 3..];
            if let [Some(a), Some(b), Some(c)] = tail {
                tracing::warn!(
                    task_id = %task_id,
                    round,
                    issues = %format_args!("[{a}, {b}, {c}]"),
                    "review loop impasse detected: issue count not decreasing"
                );
            }
        }
        if waiting {
            if review_wait_budget
                .fail_if_exceeded(store, task_id, round)
                .await?
            {
                return Ok(());
            }
            if waiting_count >= MAX_CONSECUTIVE_WAITS {
                tracing::warn!(
                    round,
                    waiting_count,
                    "PR #{pr_num} review bot has not responded after {MAX_CONSECUTIVE_WAITS} waits; \
                     consuming a round to prevent infinite loop"
                );
                round += 1;
                waiting_count = 0;
            } else {
                tracing::info!(
                    round,
                    waiting_count,
                    "PR #{pr_num} review bot has not responded yet; sleeping without consuming round"
                );
                record_runtime_pr_feedback(
                    issue_workflow_store,
                    workflow_runtime_store,
                    project_root,
                    req,
                    task_id,
                    pr_num,
                    pr_url.as_deref(),
                    PrFeedbackOutcome::NoActionableFeedback,
                    "Review output asked the workflow to keep waiting for fresh feedback.",
                )
                .await;
                waiting_count += 1;
                update_status(store, task_id, TaskStatus::Waiting, waiting_count).await?;
                sleep(review_wait_budget.sleep_duration(wait_secs)).await;
                continue;
            }
        }
        let result_label = if lgtm { "lgtm" } else { "fixed" };
        mutate_and_persist(store, task_id, |s| {
            s.rounds.push(RoundResult::new(
                round,
                "review",
                result_label,
                None,
                Some(review_telemetry.clone()),
                None,
            ));
        })
        .await?;
        store.log_event(crate::event_replay::TaskEvent::RoundCompleted {
            task_id: task_id.0.clone(),
            ts: crate::event_replay::now_ts(),
            round,
            result: result_label.to_string(),
        });
        let event = build_task_event(
            task_id,
            round,
            "review",
            "pr_review",
            if lgtm {
                Decision::Complete
            } else {
                Decision::Warn
            },
            Some(format!("round {round}: {result_label}")),
            Some(format!("pr={pr_num}")),
            Some(review_telemetry.clone()),
            None,
            Some(output.clone()),
        );
        if let Err(e) = events.log(&event).await {
            tracing::warn!("failed to log pr_review event: {e}");
        }
        if lgtm {
            let validation_executor = ShellValidationExecutor::new();
            match run_test_gate(
                &validation_executor,
                project,
                &project_config.validation.pre_push,
                project_config.validation.test_gate_timeout_secs,
                cargo_env,
            )
            .await
            {
                Ok(()) => {
                    tracing::info!("PR #{pr_num} approved at round {round}");
                    record_runtime_pr_feedback(
                        issue_workflow_store,
                        workflow_runtime_store,
                        project_root,
                        req,
                        task_id,
                        pr_num,
                        pr_url.as_deref(),
                        PrFeedbackOutcome::ReadyToMerge,
                        "Reviewer approved and validation passed.",
                    )
                    .await;
                    mutate_and_persist(store, task_id, |s| {
                        s.status = TaskStatus::Done;
                        s.turn = round.saturating_add(1);
                    })
                    .await?;
                    store.log_event(crate::event_replay::TaskEvent::Completed {
                        task_id: task_id.0.clone(),
                        ts: crate::event_replay::now_ts(),
                    });
                    tracing::info!(
                        task_id = %task_id,
                        phase = "reviewing",
                        elapsed_secs = review_phase_start.elapsed().as_secs(),
                        "phase_completed"
                    );
                    tracing::info!(
                        task_id = %task_id,
                        status = "done",
                        turns = round.saturating_add(1),
                        pr_url = pr_url.as_deref().unwrap_or(""),
                        total_elapsed_secs = task_start.elapsed().as_secs(),
                        "task_completed"
                    );
                    return Ok(());
                }
                Err(output) => {
                    tracing::warn!(
                        task_id = %task_id,
                        round,
                        "LGTM rejected: tests failed in test gate, re-entering review round {}",
                        round.saturating_add(1)
                    );
                    lgtm_test_gate_rejected = true;
                    pending_test_failure = Some(output);
                    record_runtime_pr_feedback(
                        issue_workflow_store,
                        workflow_runtime_store,
                        project_root,
                        req,
                        task_id,
                        pr_num,
                        pr_url.as_deref(),
                        PrFeedbackOutcome::BlockingFeedback,
                        "Reviewer approved, but local validation failed and must be addressed.",
                    )
                    .await;
                }
            }
        }
        prev_fixed = fixed || pending_test_failure.is_some();
        if !lgtm {
            record_runtime_pr_feedback(
                issue_workflow_store,
                workflow_runtime_store,
                project_root,
                req,
                task_id,
                pr_num,
                pr_url.as_deref(),
                PrFeedbackOutcome::BlockingFeedback,
                "Review round produced actionable feedback that requires another pass.",
            )
            .await;
        }
        tracing::info!("PR #{pr_num} fixed at round {round}; waiting for bot re-review");
        if round < max_rounds {
            if review_wait_budget
                .fail_if_exceeded(store, task_id, round)
                .await?
            {
                return Ok(());
            }
            waiting_count += 1;
            update_status(store, task_id, TaskStatus::Waiting, waiting_count).await?;
            sleep(review_wait_budget.sleep_duration(wait_secs)).await;
        }
        round += 1;
    }
    let last_issue_count = issue_counts.iter().rev().find_map(|c| *c);
    let graduated = last_issue_count.is_some_and(|n| n <= 2) && !lgtm_test_gate_rejected;
    if graduated {
        tracing::info!(
            task_id = %task_id,
            remaining_issues = last_issue_count.unwrap_or(0),
            "review loop exhausted but few issues remain — marking done with warning"
        );
        record_runtime_pr_feedback(
            issue_workflow_store,
            workflow_runtime_store,
            project_root,
            req,
            task_id,
            pr_num,
            pr_url.as_deref(),
            PrFeedbackOutcome::ReadyToMerge,
            "Review loop graduated with few remaining issues.",
        )
        .await;
        mutate_and_persist(store, task_id, |s| {
            s.status = TaskStatus::Done;
            s.turn = max_rounds.saturating_add(1);
            s.error = Some(format!(
                "Graduated: LGTM not received after {} rounds but only {} issues remain.",
                max_rounds,
                last_issue_count.unwrap_or(0),
            ));
        })
        .await?;
    } else {
        mutate_and_persist(store, task_id, |s| {
            s.status = TaskStatus::Failed;
            s.turn = max_rounds.saturating_add(1);
            s.error = Some(format!(
                "Task did not receive LGTM after {} review rounds (last issues: {}).",
                max_rounds,
                last_issue_count.map_or("unknown".to_string(), |n| n.to_string()),
            ));
        })
        .await?;
    }
    tracing::info!(
        task_id = %task_id,
        phase = "reviewing",
        elapsed_secs = review_phase_start.elapsed().as_secs(),
        "phase_completed"
    );
    tracing::info!(
        task_id = %task_id,
        status = if graduated { "done" } else { "failed" },
        turns = max_rounds.saturating_add(1),
        pr_url = pr_url.as_deref().unwrap_or(""),
        total_elapsed_secs = task_start.elapsed().as_secs(),
        "task_completed"
    );
    Ok(())
}
