polyc-facts 2026.10.2

//! Committed conversation-security history facts.
//!
//! Approvals and attribution. This fold is the semantic boundary shared by
//! Projector and the end-to-end parity proof that holds the projected shapes.
//!
//! # One row per occurrence, not per event
//!
//! The embedded approvals table carried one row per event, so a request and
//! its response were separate rows and an occurrence's state was something a
//! reader reassembled. That shape is what let a superseded re-prompt request
//! look unanswered inside a committed turn, and hand a live capability to a
//! card that could never be answered.
//!
//! This fold resolves the occurrence and states the result once. A recovery
//! re-prompt appends a second `approval_request` under the same occurrence,
//! and the LATEST request wins — the same rule `match_pending_request`
//! applies on the authoritative side.
//!
//! # The committed-turn gate APPLIES here
//!
//! Unlike `conversation-delegation/v1`, whose records are committed as their
//! own signed commands under a synthetic identifier, approval and
//! attribution records are turn-tagged, and all four embedded views this
//! replaces filtered to committed turns. Dropping that filter would publish
//! uncommitted rows to a participant, which is a visibility regression
//! rather than a fidelity win. [`committed_turn_ids`] is the one definition
//! of committed, and both folds use it.
//!
//! `routine_grants` (POLY-361) is the one exception, and a narrow one: it
//! folds a bare `approval_response` ledger a routine's owner-mutation
//! commands write outside any turn at all, never a request/response
//! occurrence inside one — see [`RoutineGrantMutationFact`]'s own doc.
//!
//! An untagged attribution record names no turn. The embedded `WHERE EXISTS`
//! could never match one, so it was unreachable at every scope; it stays
//! unpublished.
//!
//! # `unanswered` is a value, not a gap
//!
//! A committed occurrence whose latest request was never answered is a fact.
//! Normalising it into absence would hide it, so [`ApprovalOutcome`] states
//! it. It is never actionable: whether an approval may still be answered is
//! authoritative State's to decide, never this artifact's.
//!
//! # Attribution is history here, and authority elsewhere
//!
//! Attribution rows are rendered, never consulted for a decision. Every
//! authorization consumer — the forensics read gate, the taint evidence, the
//! incognito fence, grant binding, the token-budget rollup — folds
//! [`attribution_events`] over authoritative replay and must keep doing so.
//! `scripts/check_conversation_security_cutover.py` holds that.

use std::collections::HashMap;

use polyc_eventlog_model::Event;
use polyc_proto::kinds;

use crate::{
    ApprovalFact, ApprovalSignatureStatus, AttributionScope, GrantReplaySignatureStatus,
    PreparedSource, attribution_events_with_positions, committed_turn_ids, fold_approval_event,
    fold_grant_replay_event, prepare_conversation_core,
};

/// How one approval occurrence ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApprovalOutcome {
    /// A response approved the occurrence's latest request.
    Approved,
    /// A response denied it.
    Denied,
    /// Its latest request carries no response.
    ///
    /// A real state, not a placeholder: a superseded request inside a
    /// committed turn reads this way. Never actionable.
    Unanswered,
}

impl ApprovalOutcome {
    /// Every outcome, in a fixed order.
    ///
    /// `every_outcome_appears_in_all` matches without a wildcard, so a new
    /// variant fails to compile rather than silently escaping a reader that
    /// restates this vocabulary.
    pub const ALL: [Self; 3] = [Self::Approved, Self::Denied, Self::Unanswered];

    /// Stable lowercase name for the projected `outcome` column.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Approved => "approved",
            Self::Denied => "denied",
            Self::Unanswered => "unanswered",
        }
    }
}

/// One approval occurrence, resolved.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovalHistoryFact {
    /// Journal position of the occurrence's LATEST request.
    pub position: u64,
    /// The committed turn the occurrence belongs to.
    pub turn_id: String,
    /// The occurrence identifier, unique within its turn.
    pub request_id: String,
    /// The tool the latest request named.
    pub tool_name: String,
    /// The arguments the latest request carried.
    pub args_json: String,
    /// How the occurrence ended.
    pub outcome: ApprovalOutcome,
    /// The responder's stated reason, empty when unanswered.
    pub response_reason: String,
    /// How the response's signature reads, empty when unanswered.
    pub signature_status: String,
    /// Whether the response minted a routine grant.
    pub routine_grant: bool,
    /// The granted tool's descriptor hash, empty unless `routine_grant`.
    pub tool_descriptor_hash: String,
    /// The grant's scope, empty unless `routine_grant`.
    pub grant_scope: String,
}

/// Fleet-only detail for one approval occurrence.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovalDetailFact {
    /// Journal position of the occurrence's latest request.
    pub position: u64,
    /// Why the request was raised.
    pub request_reason: String,
    /// The sandbox mode the request ran under.
    pub request_sandbox_mode: String,
    /// The responder's raw ed25519 key.
    ///
    /// All zeroes when the occurrence carries no verified response key — an
    /// unanswered occurrence, or a response whose key did not decode. A real
    /// ed25519 public key is never all zeroes, so the sentinel is
    /// unambiguous.
    pub signer_public_key: [u8; 32],
    /// Arguments the approver substituted, empty when none.
    pub modified_args_json: String,
    /// Whether the decision was scoped to the whole session.
    pub approved_for_session: bool,
    /// The persona that made the call.
    pub caller: String,
    /// The persona that answered.
    pub approver: String,
    /// The sandbox mode the response recorded.
    pub response_sandbox_mode: String,
    /// Context injected alongside the decision.
    pub injected_context: String,
}

/// One caller or participant attribution record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AttributionHistoryFact {
    /// Journal position of the record.
    pub position: u64,
    /// The turn it was tagged with, empty when untagged.
    pub turn_id: String,
    /// The persona attributed.
    pub persona_id: String,
    /// `initiator` or `participant`.
    pub role: String,
}

/// Fleet-only identity and edge evidence for one attribution record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AttributionProvenanceFact {
    /// Journal position of the record this describes.
    pub position: u64,
    /// The external identity provider.
    pub identity_provider: String,
    /// The identity's scope within that provider.
    pub identity_scope: String,
    /// The identity's external identifier.
    pub identity_external_id: String,
    /// The identity's display name.
    pub identity_display_name: String,
    /// The edge that asserted the attribution.
    pub asserting_edge_id: String,
    /// That edge's registered public key, hex.
    pub signer_pk_hex: String,
    /// The envelope signature, hex.
    pub signature_hex: String,
}

/// One `grant_replay` audit record, without signer material.
///
/// Mirrors the deleted embedded `grant_replays` view
/// (`crates/query/src/decode/grant_replays.rs`, removed with the engine in
/// #3109): every column that view showed a non-Fleet scope.
/// Unlike [`ApprovalHistoryFact`]/[`AttributionHistoryFact`], this is NOT
/// gated on the committed-turn set — neither `collect_grant_replays`
/// (`crates/control-plane/src/forensics.rs`) nor the embedded
/// `grant_replays` typed table filtered on commit status, so this fold
/// keeps parity with both rather than introducing a new filter.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GrantReplayHistoryFact {
    /// Journal position of the record.
    pub position: u64,
    /// The `:{turn_uuid}` suffix off the event's kind, empty when untagged.
    pub turn_id: String,
    /// The tool whose call the grant cleared.
    pub tool: String,
    /// The grant that cleared it: sha256-hex of its full signed payload.
    pub grant_ref: String,
    /// The capability names the grant kept against taint, as JSON-array
    /// text.
    pub covered_capabilities: String,
    /// The template-coverage hash the grant matched.
    pub coverage_hash: String,
    /// `verified` or `invalid` — see [`GrantReplaySignatureStatus`]. A
    /// record with an untrusted or forged signature is KEPT, tagged
    /// `invalid`, never dropped: the audit surface this replaces exists to
    /// show exactly that.
    pub signature_status: String,
}

/// Fleet-only signer evidence for one grant-replay record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GrantReplaySignerFact {
    /// Journal position of the record this describes.
    pub position: u64,
    /// The embedded signer public key, hex, empty when the payload carried
    /// no `signed_by`.
    pub signer_pk_hex: String,
    /// The embedded `signature_hex`, as recorded, empty when absent.
    pub signature_hex: String,
}

/// All security-history rows folded from one prepared source prefix.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ConversationSecurityFacts {
    /// One row per committed approval occurrence.
    pub approvals: Vec<ApprovalHistoryFact>,
    /// Fleet-only detail, one row per entry in [`Self::approvals`].
    pub approval_details: Vec<ApprovalDetailFact>,
    /// One row per attribution record.
    pub attribution: Vec<AttributionHistoryFact>,
    /// Fleet-only provenance, one row per entry in [`Self::attribution`].
    pub attribution_provenance: Vec<AttributionProvenanceFact>,
    /// One row per structurally-decodable `grant_replay` record.
    pub grant_replays: Vec<GrantReplayHistoryFact>,
    /// Fleet-only signer evidence, one row per entry in
    /// [`Self::grant_replays`].
    pub grant_replay_signers: Vec<GrantReplaySignerFact>,
    /// One row per routine-grant-mutation record (POLY-361) — see
    /// [`RoutineGrantMutationFact`]'s own doc for why this is the one
    /// exception to this module's committed-turn gate.
    pub routine_grants: Vec<RoutineGrantMutationFact>,
}

/// One `approval_response` record carrying the `routine_grant` marker.
///
/// Written by `crate::routine_nav`'s create-time blanket mint,
/// `AllowDenial`'s allow-from-denial path, and `RevokeGrant`, all through
/// `crates/control-plane/src/approval.rs`'s `append_routine_grant`.
///
/// # The one exception to this module's committed-turn gate
///
/// Every other row in this family requires both `TURN_START` and
/// `TURN_COMPLETE` for the turn tag it carries ([`committed_turn_ids`]) —
/// deliberately, per this module's own doc. A routine-grant mutation cannot
/// meet that bar: `append_routine_grant` commits ONLY the `approval_response`
/// event, under a synthetic turn id that never gets a `TURN_START` or
/// `TURN_COMPLETE` of its own. The embedded `routine_grants` view this
/// replaces already carried the identical exception, and documents why:
/// "A grant minted outside a paused turn... carries a synthetic turn id with
/// NO `turn_dispatched`/`turn_complete` events behind it, so the usual
/// committed-turn semijoin would hide exactly the records this ledger exists
/// to show."
///
/// # One row per RECORD, not per occurrence
///
/// [`ApprovalHistoryFact`] resolves an occurrence (request + response) to
/// one row, and drops a response with no matching request in the same
/// prefix — a deliberate anti-fabrication guard: never infer an approver
/// from a neighbouring event. A routine-grant mutation is a bare,
/// self-contained response with no request at all (`append_routine_grant`
/// commits no `approval_request`), so folding it through that machinery
/// would drop every one of these rows, not just the requestless minority.
/// This table is the raw ledger instead — one row per verified record,
/// newest and oldest alike, exactly mirroring `ROUTINE_GRANTS_VIEW_SQL`'s own
/// "Rows are the raw ledger" doc. The latest-wins fold
/// (`routine_owner_active_grants.sql`'s windowing) is the reader's job, not
/// this fold's.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoutineGrantMutationFact {
    /// Journal position.
    pub position: u64,
    /// The synthetic turn id `append_routine_grant` minted for this record.
    pub turn_id: String,
    /// The granted or revoked tool, empty for a blanket-scope record.
    pub tool_name: String,
    /// The granted tool's descriptor hash, empty for a blanket-scope record.
    pub tool_descriptor_hash: String,
    /// `"tool"`, `"blanket_below_high"`, or `"blanket_all"`.
    pub grant_scope: String,
    /// Whether this record grants (`true`) or revokes (`false`).
    pub approved: bool,
    /// The actor-supplied reason, empty when none was given.
    pub response_reason: String,
    /// How this record's embedded signature reads.
    pub signature_status: String,
}

/// Why a security generation was refused.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ConversationSecurityError {
    /// Two records claimed the same source position.
    #[error("the source prefix repeats position {position}")]
    RepeatedPosition {
        /// The repeated position.
        position: u64,
    },
    /// An APPROVAL record's payload did not decode.
    ///
    /// Attribution is deliberately not symmetric here.
    /// `attribution_events_with_positions` skips an undecodable payload and
    /// this fold inherits that, because the taint and completeness read paths
    /// share that primitive and a refusal there would change their behaviour.
    /// The consequence is real and worth stating: a corrupt attribution
    /// record disappears from a signed artifact with no refusal. Making both
    /// sides refuse is its own change, in the crate that owns the primitive.
    #[error("the {kind} record at position {position} did not decode")]
    Malformed {
        /// The event kind that failed.
        kind: &'static str,
        /// Its journal position.
        position: u64,
    },
}

fn refuse_repeated_positions(events: &[(u64, Event)]) -> Result<(), ConversationSecurityError> {
    let mut seen = std::collections::HashSet::with_capacity(events.len());
    for (position, _) in events {
        if !seen.insert(*position) {
            return Err(ConversationSecurityError::RepeatedPosition {
                position: *position,
            });
        }
    }
    Ok(())
}

/// Folds one prepared source prefix into its security history.
///
/// # Errors
///
/// Returns [`ConversationSecurityError`] for a repeated source position or a
/// malformed approval payload. An unusable responder key is NOT an error: the
/// occurrence publishes with a zeroed key, exactly as an unanswered one does.
/// `partition` is the source this generation is folding — `"conv-{id}"`
/// for a conversation family. `fold_routine_grants` is the one fold that
/// reads it: see that function's own doc for why.
pub fn fold_conversation_security(
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
    partition: &str,
) -> Result<ConversationSecurityFacts, ConversationSecurityError> {
    refuse_repeated_positions(events)?;
    let mut facts = ConversationSecurityFacts::default();
    fold_approvals(events, trusted_signers, &mut facts)?;
    fold_attribution(events, &mut facts);
    fold_grant_replays(events, trusted_signers, &mut facts);
    fold_routine_grants(events, trusted_signers, partition, &mut facts);
    Ok(facts)
}

/// Renders every `approval_response` record carrying the `routine_grant`
/// marker into [`RoutineGrantMutationFact`] — see that type's own doc for
/// why this is the one committed-turn and request-matching exception in
/// this module. Mirrors every skip rule Control's own
/// `collect_routine_tool_grants` (`crates/control-plane/src/grpc/
/// approvals.rs:850-889`) applies to the identical record shape, in the
/// same order:
///
/// - not an `approval_response` event: skipped (that function's line 868).
/// - untagged with a turn (`kinds::parse` returns no turn): skipped (line
///   871) — a routine-grant response is always turn-tagged in practice
///   (`append_routine_grant` mints a synthetic turn id for exactly this),
///   but this fold does not assume that; it enforces it, the same
///   uniform-key discipline every table in this workspace holds.
/// - fails even a structural decode: skipped, matching
///   [`fold_approval_event`]'s own posture for every other caller (lines
///   872-874).
/// - `routine_grant` marker absent, OR the response's own signed
///   `conversation_id` does not name the conversation this fold is
///   folding: skipped (line 875) — "a record naming a different routine's
///   conversation is unambiguously wrong"
///   (`routine_tool_grants_fold_drops_a_record_naming_a_different_
///   conversation`'s own doc), and this fold is bound to the SAME
///   partition-derived ground truth `collect_routine_tool_grants`'s
///   caller supplies, never to the payload's own unverified claim alone.
///
/// An untrusted or forged signature is KEPT and tagged invalid rather
/// than dropped, since [`RoutineGrantMutationFact::signature_status`]
/// states the verdict explicitly and a reader (the harness gate) checks
/// it before honoring the grant either way — line 875 also does not
/// filter on signature, for the identical reason (that function's own
/// `decode_response_full` call never reads one) — the same posture
/// `fold_grant_replays` already holds for its own audit trail.
fn fold_routine_grants(
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
    partition: &str,
    facts: &mut ConversationSecurityFacts,
) {
    for (position, event) in events {
        let (base, turn) = kinds::parse(&event.kind);
        if base != kinds::APPROVAL_RESPONSE {
            continue;
        }
        let Some(turn) = turn else {
            continue;
        };
        let Some(ApprovalFact::Response(response)) = fold_approval_event(event, trusted_signers)
        else {
            continue;
        };
        if response.routine_grant != Some(true) {
            continue;
        }
        let Some(conversation_id) = response.conversation_id.as_deref() else {
            continue;
        };
        if format!("conv-{conversation_id}") != partition {
            continue;
        }
        facts.routine_grants.push(RoutineGrantMutationFact {
            position: *position,
            turn_id: turn.to_string(),
            tool_name: response.tool_name.unwrap_or_default(),
            tool_descriptor_hash: response.tool_descriptor_hash.unwrap_or_default(),
            grant_scope: response.grant_scope.unwrap_or_default(),
            approved: response.approved,
            response_reason: response.response_reason.unwrap_or_default(),
            signature_status: signature_status(response.signature_status),
        });
    }
}

/// Renders every structurally-decodable `grant_replay` record, through the
/// shared fold ([`fold_grant_replay_event`]) every other consumer of this
/// event kind already reads through.
///
/// No committed-turn gate: see [`GrantReplayHistoryFact`]'s doc for why.
fn fold_grant_replays(
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
    facts: &mut ConversationSecurityFacts,
) {
    for (position, event) in events {
        let (base, turn) = kinds::parse(&event.kind);
        if base != kinds::GRANT_REPLAY {
            continue;
        }
        let Some(fact) = fold_grant_replay_event(event, trusted_signers) else {
            continue;
        };
        let signature_status = match fact.signature_status {
            GrantReplaySignatureStatus::Verified => "verified",
            GrantReplaySignatureStatus::Invalid => "invalid",
        }
        .to_owned();
        facts.grant_replays.push(GrantReplayHistoryFact {
            position: *position,
            turn_id: turn.map(|turn| turn.to_string()).unwrap_or_default(),
            tool: fact.tool,
            grant_ref: fact.grant_ref,
            covered_capabilities: serde_json::to_string(&fact.covered_capabilities)
                .unwrap_or_default(),
            coverage_hash: fact.coverage_hash,
            signature_status,
        });
        facts.grant_replay_signers.push(GrantReplaySignerFact {
            position: *position,
            signer_pk_hex: fact
                .signer_public_key
                .as_deref()
                .map(polyc_crypto::hex::lower)
                .unwrap_or_default(),
            signature_hex: fact.signature_hex.unwrap_or_default(),
        });
    }
}

/// Resolves every committed approval occurrence into one row.
///
/// Approvals are turn-tagged and the contract this replaces filtered to
/// committed turns. Keep that: an uncommitted row is not history yet.
fn fold_approvals(
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
    facts: &mut ConversationSecurityFacts,
) -> Result<(), ConversationSecurityError> {
    let committed = committed_turn_ids(events.iter().map(|(_, event)| event));

    // Resolve each occurrence. The latest request wins, so a recovery
    // re-prompt supersedes rather than shadows.
    let mut by_occurrence: HashMap<(String, String), usize> = HashMap::new();
    for (position, event) in events {
        let (base, turn) = kinds::parse(&event.kind);
        if base != kinds::APPROVAL_REQUEST && base != kinds::APPROVAL_RESPONSE {
            continue;
        }
        let Some(turn) = turn.filter(|turn| committed.contains(turn)) else {
            continue;
        };
        let kind = if base == kinds::APPROVAL_REQUEST {
            kinds::APPROVAL_REQUEST
        } else {
            kinds::APPROVAL_RESPONSE
        };
        let fact = fold_approval_event(event, trusted_signers).ok_or(
            ConversationSecurityError::Malformed {
                kind,
                position: *position,
            },
        )?;
        let turn_id = turn.to_string();
        match fact {
            ApprovalFact::Request(request) => {
                let occurrence = (turn_id.clone(), request.request_id.clone());
                let row = ApprovalHistoryFact {
                    position: *position,
                    turn_id,
                    request_id: request.request_id,
                    tool_name: request.tool_name,
                    args_json: request.args_json,
                    outcome: ApprovalOutcome::Unanswered,
                    response_reason: String::new(),
                    signature_status: String::new(),
                    routine_grant: false,
                    tool_descriptor_hash: String::new(),
                    grant_scope: String::new(),
                };
                let detail = ApprovalDetailFact {
                    position: *position,
                    request_reason: request.reason,
                    request_sandbox_mode: request.sandbox_mode,
                    signer_public_key: [0u8; 32],
                    modified_args_json: String::new(),
                    approved_for_session: false,
                    caller: String::new(),
                    approver: String::new(),
                    response_sandbox_mode: String::new(),
                    injected_context: String::new(),
                };
                if let Some(&index) = by_occurrence.get(&occurrence) {
                    // A re-minted request supersedes the earlier one, and
                    // clears any response the earlier one had collected.
                    facts.approvals[index] = row;
                    facts.approval_details[index] = detail;
                } else {
                    by_occurrence.insert(occurrence, facts.approvals.len());
                    facts.approvals.push(row);
                    facts.approval_details.push(detail);
                }
            }
            ApprovalFact::Response(response) => {
                let occurrence = (turn_id, response.request_id.clone());
                // A response with no preceding request in this prefix names no
                // occurrence to resolve. It is not invented here: inferring an
                // approver from a neighbouring event is exactly what the
                // directive forbids.
                let Some(&index) = by_occurrence.get(&occurrence) else {
                    continue;
                };
                let row = &mut facts.approvals[index];
                row.outcome = if response.approved {
                    ApprovalOutcome::Approved
                } else {
                    ApprovalOutcome::Denied
                };
                row.response_reason = response.response_reason.unwrap_or_default();
                row.signature_status = signature_status(response.signature_status);
                row.routine_grant = response.routine_grant.unwrap_or(false);
                row.tool_descriptor_hash = response.tool_descriptor_hash.unwrap_or_default();
                row.grant_scope = response.grant_scope.unwrap_or_default();

                let detail = &mut facts.approval_details[index];
                detail.signer_public_key = response
                    .signer_public_key
                    .and_then(|key| <[u8; 32]>::try_from(key.as_slice()).ok())
                    .unwrap_or([0u8; 32]);
                detail.modified_args_json = response.modified_args_json.unwrap_or_default();
                detail.approved_for_session = response.approved_for_session.unwrap_or(false);
                detail.caller = response.caller.unwrap_or_default();
                detail.approver = response.approver.unwrap_or_default();
                detail.response_sandbox_mode = response.sandbox_mode.unwrap_or_default();
                detail.injected_context = response.injected_context.unwrap_or_default();
            }
        }
    }

    Ok(())
}

/// Renders every attribution record of a committed turn.
///
/// The committed-turn gate applies here for the same reason it applies to
/// approvals, and against the same definition. Both embedded views this
/// replaces ended in `WHERE EXISTS (SELECT 1 FROM events e WHERE e.partition
/// = t.partition AND e.turn_id = t.turn_id)`, and `events` was the
/// committed-turns projection, so a record of an uncommitted or orphaned
/// turn was filtered out at every scope.
///
/// [`SECURITY_ATTRIBUTION`](polyc_projection::family::SECURITY_ATTRIBUTION)
/// is a VISIBLE table. Publishing an uncommitted record would let a
/// conversation-grant reader enumerate who else was in a turn that never
/// completed, which is a visibility regression rather than a fidelity win.
///
/// An UNTAGGED record names no turn. The SQL `EXISTS` could never match one,
/// because a null `turn_id` equals nothing, so those were unreachable at
/// every scope too. They stay unpublished here.
fn fold_attribution(events: &[(u64, Event)], facts: &mut ConversationSecurityFacts) {
    let committed = committed_turn_ids(events.iter().map(|(_, event)| event));
    for fact in attribution_events_with_positions(
        events.iter().map(|(position, event)| (*position, event)),
        AttributionScope::Both,
    ) {
        let Some(turn) = fact.turn_id.filter(|turn| committed.contains(turn)) else {
            continue;
        };
        facts.attribution.push(AttributionHistoryFact {
            position: fact.position,
            turn_id: turn.to_string(),
            persona_id: fact.persona_id,
            role: fact.role,
        });
        let identity = fact.identity.unwrap_or_default();
        facts
            .attribution_provenance
            .push(AttributionProvenanceFact {
                position: fact.position,
                identity_provider: identity.provider,
                identity_scope: identity.scope,
                identity_external_id: identity.external_id,
                identity_display_name: identity.display_name,
                asserting_edge_id: fact.asserting_edge_id,
                signer_pk_hex: fact.signer_pk_hex,
                signature_hex: fact.signature_hex,
            });
    }
}

/// The projected spelling of one signature verdict.
///
/// Exhaustive over [`ApprovalSignatureStatus`], so a new verdict is a compile
/// error here rather than a row that reads as something it is not.
fn signature_status(status: ApprovalSignatureStatus) -> String {
    match status {
        ApprovalSignatureStatus::Verified => "verified",
        ApprovalSignatureStatus::Invalid => "invalid",
        ApprovalSignatureStatus::LegacyUnverifiable => "legacy_unverifiable",
    }
    .to_owned()
}

/// Prepares one source prefix, then folds its security history.
///
/// # Errors
///
/// Returns [`ConversationSecurityError`] for every reason
/// [`fold_conversation_security`] does.
pub fn prepare_and_fold_conversation_security(
    events: &mut [(u64, Event)],
    partition: &str,
    trusted_signers: &[Vec<u8>],
) -> Result<(PreparedSource, ConversationSecurityFacts), ConversationSecurityError> {
    let prepared = prepare_conversation_core(events, partition);
    let facts = fold_conversation_security(events, trusted_signers, partition)?;
    Ok((prepared, facts))
}

#[cfg(test)]
mod tests {

    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use polyc_crypto::approval::{self, ApprovalSigner, response_payload, test_util};
    use uuid::Uuid;

    use super::*;

    fn turn() -> Uuid {
        Uuid::from_u128(0x5EC0)
    }

    fn tagged(base: &str) -> String {
        kinds::tagged(base, &turn())
    }

    /// `turn_start` + `turn_complete`, so the occurrence is committed.
    fn commit_markers(from: u64) -> Vec<(u64, Event)> {
        vec![
            (from, Event::new(tagged(kinds::TURN_START), Vec::new())),
            (
                from + 1,
                Event::new(tagged(kinds::TURN_COMPLETE), Vec::new()),
            ),
        ]
    }

    fn request(position: u64, request_id: &str, tool: &str) -> (u64, Event) {
        (
            position,
            Event::new(
                tagged(kinds::APPROVAL_REQUEST),
                approval::request_payload(
                    request_id,
                    tool,
                    "{}",
                    "default",
                    "",
                    &[],
                    "",
                    "",
                    "",
                    &[],
                    false,
                ),
            ),
        )
    }

    fn response(
        position: u64,
        request_id: &str,
        tool: &str,
        approved: bool,
        signer: &ApprovalSigner,
    ) -> (u64, Event) {
        let payload = response_payload(
            request_id,
            tool,
            "{}",
            "",
            approved,
            false,
            &[],
            "caller-1",
            "",
            "",
            "because",
            "",
            "conv-1",
            "nonce-1",
            "",
            signer,
        )
        .0;
        (
            position,
            Event::new(tagged(kinds::APPROVAL_RESPONSE), payload),
        )
    }

    fn fold(events: Vec<(u64, Event)>, signer: &ApprovalSigner) -> ConversationSecurityFacts {
        fold_conversation_security(&events, &[signer.public_key_bytes()], "conv-1")
            .expect("the fold")
    }

    /// `ALL` must list every outcome.
    ///
    /// The match has no wildcard, so a new variant fails to compile here.
    #[test]
    fn every_outcome_appears_in_all() {
        for outcome in ApprovalOutcome::ALL {
            match outcome {
                ApprovalOutcome::Approved
                | ApprovalOutcome::Denied
                | ApprovalOutcome::Unanswered => {}
            }
        }
        assert_eq!(ApprovalOutcome::ALL.len(), 3);
    }

    /// A recovery re-prompt supersedes the earlier request rather than
    /// shadowing it. This is the projected half of the phantom-token defect:
    /// the shadow row cannot exist here at all.
    #[test]
    fn a_re_prompted_occurrence_yields_one_answered_row() {
        let signer = ApprovalSigner::from_seed(7);
        let mut events = commit_markers(1);
        events.push(request(3, "call-1", "rm"));
        events.push(request(4, "call-1", "rm"));
        events.push(response(5, "call-1", "rm", true, &signer));

        let facts = fold(events, &signer);
        assert_eq!(facts.approvals.len(), 1, "one row per occurrence");
        assert_eq!(facts.approvals[0].outcome, ApprovalOutcome::Approved);
        assert_eq!(
            facts.approvals[0].position, 4,
            "the row carries the LATEST request's position"
        );
        assert_eq!(facts.approval_details.len(), 1, "detail rows stay paired");
    }

    /// A response BETWEEN two requests leaves the latest request unanswered.
    /// That is a real state and it is stated, not dropped.
    #[test]
    fn a_request_after_a_response_reads_unanswered() {
        let signer = ApprovalSigner::from_seed(8);
        let mut events = commit_markers(1);
        events.push(request(3, "call-1", "rm"));
        events.push(response(4, "call-1", "rm", true, &signer));
        events.push(request(5, "call-1", "rm"));

        let facts = fold(events, &signer);
        assert_eq!(facts.approvals.len(), 1);
        assert_eq!(
            facts.approvals[0].outcome,
            ApprovalOutcome::Unanswered,
            "the latest request supersedes the answered one"
        );
        assert_eq!(facts.approvals[0].position, 5);
        assert!(
            facts.approvals[0].response_reason.is_empty(),
            "a superseding request clears the earlier response"
        );
    }

    #[test]
    fn a_denial_reads_denied_and_keeps_its_reason() {
        let signer = ApprovalSigner::from_seed(9);
        let mut events = commit_markers(1);
        events.push(request(3, "call-1", "rm"));
        events.push(response(4, "call-1", "rm", false, &signer));

        let facts = fold(events, &signer);
        assert_eq!(facts.approvals[0].outcome, ApprovalOutcome::Denied);
        assert_eq!(facts.approvals[0].response_reason, "because");
        assert_eq!(facts.approvals[0].signature_status, "verified");
    }

    /// An uncommitted turn contributes no approval row. The embedded views
    /// filtered to committed turns and dropping that would publish
    /// uncommitted rows to a participant.
    #[test]
    fn an_uncommitted_turn_contributes_no_approval() {
        let signer = ApprovalSigner::from_seed(10);
        // `turn_start` only: never completed.
        let events = vec![
            (1, Event::new(tagged(kinds::TURN_START), Vec::new())),
            request(2, "call-1", "rm"),
            response(3, "call-1", "rm", true, &signer),
        ];

        let facts = fold(events, &signer);
        assert!(
            facts.approvals.is_empty(),
            "an uncommitted turn is not history yet"
        );
    }

    /// An orphan response names no occurrence, and none is invented for it.
    #[test]
    fn a_response_without_its_request_resolves_nothing() {
        let signer = ApprovalSigner::from_seed(11);
        let mut events = commit_markers(1);
        events.push(response(3, "call-ghost", "rm", true, &signer));

        let facts = fold(events, &signer);
        assert!(
            facts.approvals.is_empty(),
            "an approver is never inferred from a neighbouring event"
        );
    }

    fn caller_event(position: u64, kind: String, persona_id: &str) -> (u64, Event) {
        use buffa::Message as _;
        let event = polyc_proto::proto::polychrome::events::v1::AttributionEvent {
            persona_id: persona_id.to_owned(),
            identity: buffa::MessageField::none(),
            role: "initiator".to_owned(),
            asserting_edge_id: "edge-1".to_owned(),
            signer_pk_hex: "aa".repeat(32),
            signature_hex: "bb".repeat(64),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        (position, Event::new(kind, event.encode_to_vec()))
    }

    /// The committed-turn gate applies to attribution, not only approvals.
    ///
    /// `SECURITY_ATTRIBUTION` is a VISIBLE table, so an ungated fold lets a
    /// conversation-grant reader enumerate who else was in a turn that never
    /// completed. Both embedded views this replaces filtered on committed
    /// turns; dropping either half of this test's fixture makes the fold
    /// publish a row it must not.
    #[test]
    fn only_a_committed_turn_contributes_attribution() {
        let signer = ApprovalSigner::from_seed(20);
        let mut events = commit_markers(1);
        events.push(caller_event(3, tagged(kinds::CALLER), "persona-ada"));

        let facts = fold(events, &signer);
        assert_eq!(facts.attribution.len(), 1, "the committed record publishes");
        assert_eq!(facts.attribution[0].persona_id, "persona-ada");
        assert_eq!(facts.attribution[0].turn_id, turn().to_string());
        assert_eq!(
            facts.attribution_provenance.len(),
            1,
            "its Fleet-only provenance rides alongside it"
        );
    }

    #[test]
    fn an_uncommitted_turn_contributes_no_attribution() {
        let signer = ApprovalSigner::from_seed(21);
        // `turn_start` only: never completed.
        let events = vec![
            (1, Event::new(tagged(kinds::TURN_START), Vec::new())),
            caller_event(2, tagged(kinds::CALLER), "persona-ada"),
            caller_event(3, tagged(kinds::PARTICIPANT), "persona-ida"),
        ];

        let facts = fold(events, &signer);
        assert!(
            facts.attribution.is_empty(),
            "a participant of an uncommitted turn is not history yet"
        );
        assert!(
            facts.attribution_provenance.is_empty(),
            "and neither is their raw external identity"
        );
    }

    /// An untagged record names no turn, so the embedded `WHERE EXISTS` could
    /// never match it. It stays unpublished rather than publishing under an
    /// empty `turn_id`.
    #[test]
    fn an_untagged_attribution_record_is_not_published() {
        let signer = ApprovalSigner::from_seed(22);
        let mut events = commit_markers(1);
        events.push(caller_event(3, kinds::CALLER.to_owned(), "persona-ada"));

        let facts = fold(events, &signer);
        assert!(
            facts.attribution.is_empty(),
            "a record naming no turn was unreachable at every scope before"
        );
    }

    #[test]
    fn a_repeated_source_position_is_refused() {
        let signer = ApprovalSigner::from_seed(12);
        let events = vec![request(2, "call-1", "rm"), request(2, "call-2", "rm")];
        assert_eq!(
            fold_conversation_security(&events, &[signer.public_key_bytes()], "conv-1"),
            Err(ConversationSecurityError::RepeatedPosition { position: 2 })
        );
    }

    /// An unanswered occurrence publishes a zeroed responder key rather than
    /// refusing, and the sentinel is unambiguous.
    #[test]
    fn an_unanswered_occurrence_publishes_a_zeroed_key() {
        let signer = ApprovalSigner::from_seed(13);
        let mut events = commit_markers(1);
        events.push(request(3, "call-1", "rm"));

        let facts = fold(events, &signer);
        assert_eq!(facts.approvals[0].outcome, ApprovalOutcome::Unanswered);
        assert_eq!(facts.approval_details[0].signer_public_key, [0u8; 32]);
        assert_ne!(
            signer.public_key_bytes().as_slice(),
            [0u8; 32].as_slice(),
            "a real key is never the sentinel"
        );
    }

    fn grant_replay_event(
        turn_id: &str,
        tool: &str,
        grant_ref: &str,
        covered_capabilities: &[String],
        coverage_hash: &str,
        signer: &ApprovalSigner,
    ) -> Event {
        let (payload, _sig, _pk) = test_util::grant_replay_payload(
            "conv-1",
            turn_id,
            tool,
            grant_ref,
            covered_capabilities,
            coverage_hash,
            signer,
        );
        Event::new(
            kinds::tagged(kinds::GRANT_REPLAY, &Uuid::parse_str(turn_id).unwrap()),
            payload,
        )
    }

    /// A grant-replay signed by a trusted signer folds to a paired
    /// `verified` history row and its signer evidence, regardless of
    /// commit status — grant-replays carry no committed-turn gate.
    #[test]
    fn a_trusted_grant_replay_folds_with_its_signer_evidence() {
        let signer = ApprovalSigner::from_seed(30);
        let turn = Uuid::from_u128(0x6EA7);
        // No commit_markers: an uncommitted turn still contributes a row,
        // unlike approvals/attribution.
        let events = vec![(
            1,
            grant_replay_event(
                &turn.to_string(),
                "read_file",
                "grant-ref-1",
                &["fs.read".to_owned()],
                "hash-1",
                &signer,
            ),
        )];
        let facts = fold_conversation_security(&events, &[signer.public_key_bytes()], "conv-1")
            .expect("the fold");
        assert_eq!(facts.grant_replays.len(), 1);
        let row = &facts.grant_replays[0];
        assert_eq!(row.turn_id, turn.to_string());
        assert_eq!(row.tool, "read_file");
        assert_eq!(row.grant_ref, "grant-ref-1");
        assert_eq!(row.covered_capabilities, r#"["fs.read"]"#);
        assert_eq!(row.coverage_hash, "hash-1");
        assert_eq!(row.signature_status, "verified");

        assert_eq!(
            facts.grant_replay_signers.len(),
            1,
            "one signer row per history row"
        );
        let signer_row = &facts.grant_replay_signers[0];
        assert_eq!(signer_row.position, row.position);
        assert_eq!(
            signer_row.signer_pk_hex,
            polyc_crypto::hex::lower(&signer.public_key_bytes())
        );
        assert!(!signer_row.signature_hex.is_empty());
    }

    /// The contrast with `receipts`' drop test: an untrusted signer's
    /// grant-replay is KEPT, tagged `invalid`, not dropped — the audit
    /// surface this fold feeds exists to show exactly this.
    #[test]
    fn an_untrusted_grant_replay_is_kept_tagged_invalid() {
        let trusted = ApprovalSigner::from_seed(31);
        let untrusted = ApprovalSigner::from_seed(32);
        let turn = Uuid::from_u128(0x6EA8);
        let events = vec![(
            1,
            grant_replay_event(
                &turn.to_string(),
                "read_file",
                "grant-ref-2",
                &["fs.read".to_owned()],
                "hash-2",
                &untrusted,
            ),
        )];
        let facts = fold_conversation_security(&events, &[trusted.public_key_bytes()], "conv-1")
            .expect("the fold");
        assert_eq!(facts.grant_replays.len(), 1, "the row still surfaces");
        assert_eq!(facts.grant_replays[0].signature_status, "invalid");
        assert_eq!(
            facts.grant_replay_signers[0].signer_pk_hex,
            polyc_crypto::hex::lower(&untrusted.public_key_bytes()),
            "the untrusted signer's own key is still surfaced, for audit"
        );
    }

    /// Only a structurally-malformed payload drops the row — the same gate
    /// `polyc_facts::fold_grant_replay_event` applies.
    #[test]
    fn a_structurally_malformed_grant_replay_drops_the_row() {
        let event = Event::new(kinds::GRANT_REPLAY.to_owned(), vec![0xFF, 0xFE]);
        let facts = fold_conversation_security(&[(1, event)], &[], "conv-1").expect("the fold");
        assert!(facts.grant_replays.is_empty());
        assert!(facts.grant_replay_signers.is_empty());
    }

    fn routine_grant_event(
        tool: &str,
        approved: bool,
        tool_descriptor_hash: &str,
        grant_scope: &str,
        conversation_id: &str,
        signer: &ApprovalSigner,
    ) -> Event {
        let (payload, ..) = polyc_crypto::approval::routine_grant_payload(
            "call-routine-grant",
            tool,
            "{}",
            "",
            approved,
            "owner-1",
            "",
            "default",
            "",
            &[],
            conversation_id,
            "nonce-routine-grant",
            "",
            tool_descriptor_hash,
            grant_scope,
            signer,
        );
        Event::new(tagged(kinds::APPROVAL_RESPONSE), payload)
    }

    /// POLY-361: a routine-grant mutation folds even though it carries NO
    /// `TURN_START`/`TURN_COMPLETE` at all — the one exception to this
    /// module's committed-turn gate — and even though it has no preceding
    /// `approval_request` in the prefix, which would drop an ordinary
    /// approval response.
    #[test]
    fn a_routine_grant_folds_with_no_commit_markers_and_no_preceding_request() {
        let signer = ApprovalSigner::from_seed(41);
        let events = vec![(
            1,
            routine_grant_event("fs_write", true, "hash-abc", "tool", "fire-1", &signer),
        )];
        let facts =
            fold_conversation_security(&events, &[signer.public_key_bytes()], "conv-fire-1")
                .expect("the fold");
        assert!(
            facts.approvals.is_empty(),
            "a requestless response must not appear in the occurrence-folded table"
        );
        assert_eq!(facts.routine_grants.len(), 1);
        let grant = &facts.routine_grants[0];
        assert_eq!(grant.position, 1);
        assert_eq!(grant.tool_name, "fs_write");
        assert_eq!(grant.tool_descriptor_hash, "hash-abc");
        assert_eq!(grant.grant_scope, "tool");
        assert!(grant.approved);
        assert_eq!(grant.signature_status, "verified");
    }

    /// A revocation (`approved = false`) folds the same way as a grant.
    #[test]
    fn a_routine_grant_revocation_folds_as_a_disapproving_record() {
        let signer = ApprovalSigner::from_seed(42);
        let events = vec![(
            1,
            routine_grant_event("fs_write", false, "hash-abc", "tool", "fire-1", &signer),
        )];
        let facts =
            fold_conversation_security(&events, &[signer.public_key_bytes()], "conv-fire-1")
                .expect("the fold");
        assert_eq!(facts.routine_grants.len(), 1);
        assert!(!facts.routine_grants[0].approved);
    }

    /// An untrusted signer's routine-grant record is KEPT, tagged
    /// `invalid`, not dropped — the harness gate checks this verdict before
    /// honoring the grant either way, the same posture grant-replays holds.
    #[test]
    fn an_untrusted_routine_grant_is_kept_tagged_invalid() {
        let trusted = ApprovalSigner::from_seed(43);
        let untrusted = ApprovalSigner::from_seed(44);
        let events = vec![(
            1,
            routine_grant_event("fs_write", true, "hash-abc", "tool", "fire-1", &untrusted),
        )];
        let facts =
            fold_conversation_security(&events, &[trusted.public_key_bytes()], "conv-fire-1")
                .expect("the fold");
        assert_eq!(facts.routine_grants.len(), 1, "the row still surfaces");
        assert_eq!(facts.routine_grants[0].signature_status, "invalid");
    }

    /// An ordinary (non-grant) response with no preceding request is still
    /// dropped by the occurrence-folded table — the routine-grant exception
    /// does not widen into a general "any requestless response folds" rule.
    #[test]
    fn a_non_grant_requestless_response_is_still_dropped() {
        let signer = ApprovalSigner::from_seed(45);
        let events = vec![response(1, "req-1", "shell_exec", true, &signer)];
        let facts = fold_conversation_security(&events, &[signer.public_key_bytes()], "conv-1")
            .expect("the fold");
        assert!(facts.approvals.is_empty());
        assert!(facts.routine_grants.is_empty());
    }
}