mahbot 0.7.3

//! Joint verdict comments for pipeline stages (analysis, verification).
//!
//! Replaces the per-agent verdict comments with ONE comment per round —
//! written even on fully clean rounds so the audit trail is uniform. The
//! merge backbone is the shared LLM grouping core ([`crate::consensus`]):
//! a progress-preserving repair synthesis pass groups the agents' exact issue
//! statements — accepted groups freeze, repair rounds only touch the
//! remainder — contradiction groups carry a `— DISPUTED` marker on their
//! heading. The DISPUTED cross-reference appears in the ungrouped section
//! only when a member contradicts a frozen group (contradiction:true — never
//! for solo findings). Per-agent attribution (brackets, "Agent N:" /
//! "[blocker]" prefixes) and free-form critiques are stripped from comments —
//! scores + issues are persisted with each agent's stored verdict outcome
//! (`agents.outcome` on the phase-job roster) instead. The Analysis
//! stage shares this renderer, so analyst critiques are dropped too (analysis
//! uses the score-less `AnalysisVerdict` with per-issue grades; the
//! verification stage's reviewers and tester use the score-based `Verdict` —
//! critiques were never persisted anywhere).
//!
//! Items are referenced by stable numeric ids (global flat numbering across
//! all agents); validation is strictly structural (id range, duplicate
//! placement, completeness, contradiction ≥2 agents) and termination
//! deterministically places every remaining item in the ungrouped section,
//! eventually falling back to a deterministic raw member dump when nothing
//! ever freezes.
//!
//! Beyond comment rendering, this module also holds the parallel-verdict
//! data model ([`ParallelVerdict`], [`AgentSlot`], [`ExtractionMode`]) and the
//! reviewer-count calibration.

use std::fmt::Write as _;
use std::sync::Arc;

use crate::jobs::RowStatus;
use crate::retry::RetryExhausted;
use crate::util::{panic_message, scrub_credentials};
use crate::{
    AnalysisVerdict, BlockerVerificationVerdict, ChatMessage, ChatRequest, ChatRequestMeta, Role,
    Verdict, Workspace,
};

use super::raw_response_dump_section;

// ── Hardcoded review-count calibration defaults (no config surface) ──────

pub(crate) const DEFAULT_REVIEW_COUNT_TINY_CHURN: i64 = 200;
pub(crate) const DEFAULT_REVIEW_COUNT_LOW_CHURN: i64 = 1000;
pub(crate) const DEFAULT_REVIEW_COUNT_HIGH_CHURN: i64 = 3000;

// ── Round data ─────────────────────────────────────────────────────────

/// One valid verdict from a parallel round, unified across the verification
/// round (score-based Score) and analysis (score-less Graded). The verdict data is
/// OWNED so the round can be split away from the base_results vec — the
/// analysis consolidation must run before escalation and be reused afterward.
pub(crate) enum JointVerdict {
    Score { verdict: crate::Verdict },
    Graded { verdict: AnalysisVerdict },
}

impl JointVerdict {
    #[must_use]
    pub(crate) fn is_empty_issues(&self) -> bool {
        match self {
            Self::Score { verdict, .. } => verdict.issues_detected.is_empty(),
            Self::Graded { verdict, .. } => verdict.issues_detected.is_empty(),
        }
    }
}

/// One failed agent (no response / parse failure) with its rendered dump.
pub(crate) struct JointFailure {
    pub dump: String,
}

/// Everything the joint-comment renderer needs about a round.
pub(crate) struct JointRound {
    /// Stage name: "Analysis" or "Verification" (comment role = stage name).
    pub stage: &'static str,
    /// Valid verdicts, one per responding agent.
    pub verdicts: Vec<JointVerdict>,
    /// Failed agents (no response / parse failure).
    pub failures: Vec<JointFailure>,
    /// Per-agent issue texts, indexed by dispatch index (empty for failures /
    /// no-verdict slots). This is the text-only input to the grouping core.
    pub issues: Vec<Vec<String>>,
    /// Per-agent per-issue grade (analysis only; `None` for verification and for
    /// issues that carry no grade).
    pub grades: Vec<Vec<Option<crate::IssueGrade>>>,
}

impl JointRound {
    /// Number of valid verdicts (responding agents with parseable verdicts).
    #[must_use]
    fn n_valid(&self) -> usize {
        self.verdicts.len()
    }

    /// True when every valid verdict carries no detected issues.
    #[must_use]
    pub fn has_no_issues(&self) -> bool {
        self.verdicts.iter().all(JointVerdict::is_empty_issues)
    }
}

// ── Synthesis request ──────────────────────────────────────────────────

/// Max output tokens for the pipeline grouping pass: raised above the old
/// budget (8K) — probes truncated at 4K output on 36-issue rounds; 16K is the
/// floor for claim-length rounds.
const PIPELINE_GROUPING_MAX_TOKENS: u32 = 16_000;

/// Build the synthesis chat request for a stage role (the stage role's own
/// model, reasoning effort, and provider routing — no separate grouping
/// model). The shared contradiction package is appended to the system prompt;
/// the general workspace context is prepended by the consensus core. The
/// system prompt is byte-stable across rounds — repair-round schema selection
/// lives in the appended user-section instructions.
fn synthesis_request(
    round: &JointRound,
    role: Role,
    ws: &Workspace,
    items: &[Vec<String>],
) -> ChatRequest {
    let system = format!(
        "{}\n\n{}",
        crate::prompt::load_prompt("synthesis/synthesis.md"),
        crate::prompt::load_prompt("synthesis/grouping_contradictions.md"),
    );
    // Scores are deliberately NOT included: grouping needs issue text only —
    // scores are persisted with each agent's stored verdict outcome, never in
    // the comment.
    let material = crate::consensus::numbered_items_material(items);
    let user = format!(
        "{}\n\nStage: {}\nAgent issues (id-numbered):\n{}",
        crate::prompt::load_prompt("synthesis/synthesis_input.md"),
        round.stage,
        material,
    );
    let derived = crate::agent::role_chat_params(role);
    ChatRequest {
        messages: vec![ChatMessage::system(&system), ChatMessage::user(&user)],
        tools: None,
        model: derived.model,
        // Override the default 32K budget with the 16K aggregation budget.
        max_tokens: Some(PIPELINE_GROUPING_MAX_TOKENS),
        reasoning_effort: derived.reasoning_effort,
        provider_order: derived.provider_order,
        meta: Some(ChatRequestMeta {
            purpose: "synthesis",
            agent_id: format!("verdict_{}", crate::generate_suffix()),
            role: role.as_str().to_string(),
            workspace: ws.name.clone(),
            ticket_id: None,
        }),
    }
}

/// Run the repair-mode synthesis pass through the shared consensus core
/// (1 full call + up to N-1 repair rounds; frozen groups; per-group
/// acceptance; deterministic remainder placement; narrowed fail-open).
async fn run_synthesis(
    round: &JointRound,
    role: Role,
    ws: &Workspace,
    ticket_id: &str,
    ticket_title: &str,
) -> crate::consensus::RepairOutcome {
    let request = synthesis_request(round, role, ws, &round.issues);
    crate::consensus::run_grouping_repair(
        ws,
        "synthesis",
        request,
        &round.issues,
        Some(crate::agent::registry::ParentKey::Ticket(
            ticket_id.to_string(),
        )),
        Some(ticket_title.to_string()),
    )
    .await
}

// ── Joint comment rendering ─────────────────────────────────────────────

/// Render the joint comment for a round given the repair-mode synthesis
/// outcome.
///
/// Structure: groups of issue statements (frozen by the repair protocol
/// when synthesis succeeded, raw member dump otherwise; contradiction groups
/// carry a `— DISPUTED` marker), the code-computed
/// ungrouped remainder in a deterministic trailing section (DISPUTED
/// cross-references for items that flag a contradiction against a frozen
/// group), the first-accepted LLM summary prose (or, on the fallback path,
/// a deterministic no-issues / no-verdict summary), and a raw-dump appendix
/// for failed agents. Per-agent attribution (brackets,
/// "Agent N:" / "[blocker]" prefixes) and free-form critiques are noise and
/// are not rendered — scores + issues are already persisted with each agent's
/// stored verdict outcome.
#[must_use]
pub(crate) fn render_joint_comment(
    round: &JointRound,
    outcome: &crate::consensus::RepairOutcome,
    table: &crate::consensus::ItemTable<'_>,
) -> String {
    let mut out = String::new();

    // Issues — grouped by the LLM when available.
    let has_issues = table.len() > 0;
    if has_issues {
        match outcome {
            crate::consensus::RepairOutcome::Repaired { output, references } => {
                for group in &output.groups {
                    let _ = write!(out, "\n\n**{}**", group.heading);
                    if group.contradiction {
                        out.push_str(" — DISPUTED");
                    }
                    for member in &group.members {
                        let _ = write!(out, "\n- {}", member_bullet(round, table, member));
                    }
                }
                out.push_str(&crate::consensus::render_ungrouped_section(
                    output,
                    references,
                    |member, disputed| member_bullet(round, table, member) + disputed,
                ));
            }
            crate::consensus::RepairOutcome::Fallback => {
                // Deterministic fail-open: raw per-agent issue dump + marker
                // (global flat id order = (agent, item) order).
                out.push_str("\n\n**Issues**");
                for id in 0..table.len() {
                    if let Some((_, text)) = table.resolve(id) {
                        let line = match issue_grade(round, table, id) {
                            Some(grade) => format!("{}: {}", grade.as_str(), text),
                            None => text.to_string(),
                        };
                        let _ = write!(out, "\n- {line}");
                    }
                }
            }
        }
    }

    // First-accepted LLM summary or explicit marker.
    match outcome {
        crate::consensus::RepairOutcome::Repaired { output, .. } => {
            let summary = output.summary.trim();
            if !summary.is_empty() {
                out.push_str("\n\n### Summary");
                let _ = write!(out, "\n{summary}");
            }
        }
        crate::consensus::RepairOutcome::Fallback => {
            // Without issues there is nothing to merge: the synthesis pass was
            // deliberately skipped, so the summary must not imply it failed.
            // (Whether the round actually passes is decided separately by the
            // stage finalizer.) With issues, the deterministic per-agent dump
            // above is the whole artifact and no summary is added.
            if !has_issues {
                let summary = if round.n_valid() > 0 || round.failures.is_empty() {
                    "\n\n### Summary\nNo issues found.".to_string()
                } else {
                    "\n\n### Summary\nNo issues to merge — no agent produced a verdict.".to_string()
                };
                out.push_str(&summary);
            }
        }
    }

    // Raw-dump appendix for failed agents.
    if !round.failures.is_empty() {
        out.push_str("\n\n### Plain participant responses");
        for f in &round.failures {
            let _ = write!(out, "\n- {}\n", f.dump);
        }
    }

    // The first section's leading separator would leave leading blank lines —
    // strip them.
    crate::util::failure_detail(out.trim_start_matches('\n'), "joint verdict comment")
}

/// Resolve a grouped member's issue text via the item table (no text
/// equality). Unknown ids (defensive) render with an explicit marker.
fn member_text(
    table: &crate::consensus::ItemTable<'_>,
    member: &crate::consensus::GroupingMember,
) -> String {
    table.resolve(member.id).map_or_else(
        || format!("<unknown item id {}>", member.id),
        |(_, text)| text.to_string(),
    )
}

/// Resolve a flat item id's per-issue grade (analysis only): `None` for
/// verification rounds, for unknown ids, and for issues that carry no grade.
#[must_use]
pub(crate) fn issue_grade(
    round: &JointRound,
    table: &crate::consensus::ItemTable<'_>,
    id: usize,
) -> Option<crate::IssueGrade> {
    let (agent, item) = table.resolve_index(id)?;
    round
        .grades
        .get(agent)
        .and_then(|g| g.get(item))
        .copied()
        .flatten()
}

fn member_bullet(
    round: &JointRound,
    table: &crate::consensus::ItemTable<'_>,
    member: &crate::consensus::GroupingMember,
) -> String {
    match issue_grade(round, table, member.id) {
        Some(grade) => format!("{}: {}", grade.as_str(), member_text(table, member)),
        None => member_text(table, member),
    }
}

// ── Calibrated dynamic agent counts ─────────────────────────────────────

/// Compute the reviewer-count base from total working-tree churn (added +
/// deleted lines, including lines of new files): 1 for churn < tiny, 2 for
/// churn < low, 3 for churn < high, 4 otherwise — each threshold belongs to
/// the higher band (e.g. churn == low calibrates to 3).
#[must_use]
pub(crate) fn review_base_from_signals(
    total_churn: i64,
    tiny_churn: i64,
    low_churn: i64,
    high_churn: i64,
) -> usize {
    if total_churn < tiny_churn {
        1
    } else if total_churn < low_churn {
        2
    } else if total_churn < high_churn {
        3
    } else {
        4
    }
}

/// Apply the P0 floor: priority-0 tickets never drop below 2 reviewers
/// (floor 2). Bounces do not change the count.
#[must_use]
pub(crate) fn review_agent_count(base: usize, priority: i64) -> usize {
    if priority == 0 { base.max(2) } else { base }
}

/// Resolve a stage-name comment role back to the role whose badge and icon the
/// GUI renders for it. The stages' own names come from the stage modules
/// ([`crate::pipeline::analysis::STAGE`],
/// [`crate::pipeline::verification::STAGE`]) so a rename cannot silently leave a
/// comment unmapped; the merged stage renders with the functional tester's
/// identity — the stage's final gate — while "Review"/"QA" are the retired
/// stages' own labels, kept so historical comments keep their visuals.
#[must_use]
pub(crate) fn stage_role(name: &str) -> Option<Role> {
    match name {
        crate::pipeline::analysis::STAGE => Some(Role::Analyst),
        crate::pipeline::verification::STAGE | "QA" => Some(Role::Qa),
        "Review" => Some(Role::Reviewer),
        _ => None,
    }
}

// ── Parallel-round data shapes & joint-comment builder ──────────────────
//
// These are the verdict-round types the renderer above consumes, so they
// live with it rather than in the orchestrator.

/// Result from a single participant of a parallel round.
#[derive(Clone)]
pub(crate) enum ParallelVerdict {
    /// Agent failed to produce any response (crashed, timed out, empty output).
    NoResponse(String),
    /// Agent produced a response but structured verdict extraction failed.
    ParseFailed(RetryExhausted),
    /// Agent produced a successfully-parsed verdict.
    Verdict(Verdict),
    /// Agent produced a score-less analysis verdict (analysis base round).
    Analysis(AnalysisVerdict),
    /// Agent produced a blocker-verification verdict (analysis escalation).
    BlockerVerification(BlockerVerificationVerdict),
}

impl ParallelVerdict {
    /// True when the round never produced a result at all (no response or
    /// unparseable output) — a hard technical failure, as opposed to a
    /// completed-but-rework verdict.
    #[must_use]
    pub(crate) fn is_technical_failure(&self) -> bool {
        matches!(self, Self::NoResponse(_) | Self::ParseFailed(_))
    }
}

/// Structured-extraction behavior for a parallel round member.
#[derive(Clone)]
pub(crate) enum ExtractionMode {
    /// Standard score+issues verdict (the verification stage's participants).
    ScoreVerdict,
    /// Score-less analysis base round (no score; each issue graded).
    ScorelessVerdict,
    /// Blocker verification (analysis escalation).
    BlockerVerification { blockers: Arc<[String]> },
}

/// One roster slot of a ticket phase round.
#[derive(Clone)]
pub(crate) struct AgentSlot {
    /// Dispatch slot index (0-based; escalation continues at 3, 4).
    pub idx: i64,
    pub agent_id: String,
    /// FINAL per-agent rendered prompt (angle appended).
    pub task: String,
    pub status: RowStatus,
    /// Stored agents.outcome (tagged JSON) — set on replay of a done slot.
    pub outcome: Option<String>,
}

/// Map a panicked round member's [`tokio::task::JoinError`] to a contained
/// [`ParallelVerdict::NoResponse`] (round continues fail-open).
pub(crate) fn round_member_failed(e: tokio::task::JoinError) -> ParallelVerdict {
    let reason = scrub_credentials(&panic_message(&*e.into_panic()));
    tracing::warn!(%reason, "round member task failed");
    ParallelVerdict::NoResponse(reason)
}

/// Validate a verdict score is within [0, 10].
pub(crate) fn validate_verdict_score(v: &Verdict) -> Result<(), String> {
    if v.score <= 10 {
        Ok(())
    } else {
        Err(format!("verdict score {} out of range [0,10]", v.score))
    }
}

/// Validate a blocker-verification verdict.
pub(crate) fn validate_blocker_verification(
    v: &BlockerVerificationVerdict,
    blockers: &[String],
) -> Result<(), String> {
    if v.verdicts.is_empty() {
        return Err("blocker verification returned no verdicts".to_string());
    }
    if v.verdicts.len() != blockers.len() {
        return Err(format!(
            "blocker verification returned {} verdicts for {} blockers",
            v.verdicts.len(),
            blockers.len()
        ));
    }
    let mut seen = vec![false; blockers.len()];
    for item in &v.verdicts {
        if item.index >= blockers.len() {
            return Err(format!("blocker index {} out of range", item.index));
        }
        if seen[item.index] {
            return Err(format!("duplicate blocker index {}", item.index));
        }
        seen[item.index] = true;
        if item.reasoning.trim().is_empty() {
            return Err(format!("blocker {} missing reasoning", item.index));
        }
        if item.impact.trim().is_empty() {
            return Err(format!("blocker {} missing impact", item.index));
        }
    }
    Ok(())
}

/// Serialize a [`ParallelVerdict`] into the agents.outcome column.
#[must_use]
pub(crate) fn serialize_verdict_outcome(result: &ParallelVerdict) -> String {
    match result {
        ParallelVerdict::Verdict(v) => serde_json::json!({ "verdict": v }).to_string(),
        ParallelVerdict::Analysis(v) => serde_json::json!({ "verdict": v }).to_string(),
        ParallelVerdict::NoResponse(reason) => {
            serde_json::json!({ "no_response": reason }).to_string()
        }
        ParallelVerdict::ParseFailed(f) => {
            serde_json::json!({ "parse_failed": raw_response_dump_section(f) }).to_string()
        }
        ParallelVerdict::BlockerVerification(v) => {
            serde_json::json!({ "blocker_verification": v }).to_string()
        }
    }
}

/// Reconstruct a [`ParallelVerdict`] from the agents.outcome column.
#[must_use]
pub(crate) fn deserialize_verdict_outcome(outcome: &str) -> ParallelVerdict {
    let Ok(v) = serde_json::from_str::<serde_json::Value>(outcome) else {
        return ParallelVerdict::NoResponse("unreadable stored outcome".to_string());
    };
    if let Some(verdict) = v.get("verdict") {
        // A numeric `score` is the verification-round shape; score-less graded objects
        // are analysis verdicts. The two are NOT distinguishable at the type
        // level for empty-issue rows (`AnalysisVerdict` ignores the unknown
        // `score` field and accepts `{"issues":[]}`), so the presence of
        // `score` is the discriminating signal.
        if verdict.get("score").is_some() {
            if let Ok(vv) = serde_json::from_value::<crate::Verdict>(verdict.clone()) {
                return ParallelVerdict::Verdict(vv);
            }
        } else if let Ok(av) = serde_json::from_value::<crate::AnalysisVerdict>(verdict.clone()) {
            return ParallelVerdict::Analysis(av);
        }
        ParallelVerdict::NoResponse("unreadable stored verdict".to_string())
    } else if let Some(r) = v.get("no_response").and_then(serde_json::Value::as_str) {
        ParallelVerdict::NoResponse(r.to_string())
    } else if let Some(p) = v.get("parse_failed").and_then(serde_json::Value::as_str) {
        ParallelVerdict::NoResponse(p.to_string())
    } else if let Some(v) = v.get("blocker_verification") {
        match serde_json::from_value(v.clone()) {
            Ok(bv) => ParallelVerdict::BlockerVerification(bv),
            Err(_) => {
                ParallelVerdict::NoResponse("unreadable stored blocker verification".to_string())
            }
        }
    } else {
        ParallelVerdict::NoResponse("unrecognized stored outcome".to_string())
    }
}

/// Run the grouping pass ONCE for a round, returning the round plus its
/// synthesis outcome so the caller can reuse both (escalation selection and
/// final rendering). Skips the LLM pass when every verdict is clean, or for a
/// round the caller declares single-participant — there is nothing to merge, so
/// the deterministic dump is the honest artifact. Only the merged verification
/// stage can declare one: its skipped code review leaves the functional tester
/// as the round's only participant, whereas the analysis stage always runs the
/// pass (it tolerates partially-failed rounds, where one surviving verdict is
/// not a complete picture).
pub(crate) async fn build_round_grouping(
    stage: &'static str,
    results: &[ParallelVerdict],
    role: Role,
    single_participant_round: bool,
    ws: &Workspace,
    ticket_id: &str,
    ticket_title: &str,
) -> (JointRound, crate::consensus::RepairOutcome) {
    let round = build_joint_round(stage, results);
    let has_no_issues = round.has_no_issues();
    if has_no_issues || single_participant_round {
        (round, crate::consensus::RepairOutcome::Fallback)
    } else {
        let outcome = run_synthesis(&round, role, ws, ticket_id, ticket_title).await;
        (round, outcome)
    }
}

/// Build a [`JointRound`] from raw parallel results, cloning any valid verdict
/// data and leaving failed / no-verdict slots empty in every per-agent table.
fn build_joint_round(stage: &'static str, results: &[ParallelVerdict]) -> JointRound {
    let mut verdicts: Vec<JointVerdict> = Vec::new();
    let mut failures: Vec<JointFailure> = Vec::new();
    let mut issues: Vec<Vec<String>> = vec![Vec::new(); results.len()];
    let mut grades: Vec<Vec<Option<crate::IssueGrade>>> = vec![Vec::new(); results.len()];
    for (i, r) in results.iter().enumerate() {
        match r {
            ParallelVerdict::Verdict(v) => {
                verdicts.push(JointVerdict::Score { verdict: v.clone() });
                issues[i].clone_from(&v.issues_detected);
                grades[i] = vec![None; v.issues_detected.len()];
            }
            ParallelVerdict::Analysis(v) => {
                verdicts.push(JointVerdict::Graded { verdict: v.clone() });
                issues[i] = v.issues_detected.iter().map(|a| a.text.clone()).collect();
                grades[i] = v.issues_detected.iter().map(|a| Some(a.grade)).collect();
            }
            ParallelVerdict::NoResponse(reason) => {
                failures.push(JointFailure {
                    dump: reason.clone(),
                });
            }
            ParallelVerdict::ParseFailed(f) => {
                failures.push(JointFailure {
                    dump: scrub_credentials(&raw_response_dump_section(f)),
                });
            }
            ParallelVerdict::BlockerVerification(_) => {}
        }
    }
    JointRound {
        stage,
        verdicts,
        failures,
        issues,
        grades,
    }
}