daemon/grpc_local_impl/

state_review.rs

// SPDX-License-Identifier: Apache-2.0
//! Local-mode implementation of the [`StateReviewService`] gRPC contract.
//!
//! Reads and writes review-signature attachments. Verifies the client-supplied
//! signature against the deterministic [`signing_payload`] before persisting.

// `::state_review` disambiguates from this module's own name
// (`grpc_local_impl::state_review`), the same way the hosted impl
// disambiguates by being in a sibling module.
use ::state_review::{
    PathSymbol, ReadingOrderPartition, SymbolKind, build_review_payload_partition,
};
use api::heddle::api::v1alpha1::{
    AnchoredDiscussion, GetRepoSignalHealthRequest, GetReviewPayloadRequest,
    GetReviewProgressRequest, GetReviewProgressResponse, ListSignaturesRequest,
    ListSignaturesResponse, MergeRequirement, PathSymbolRef as ProtoPathSymbolRef,
    ReadingOrderPartition as ProtoReadingOrderPartition, RecordCheckAckRequest,
    RecordCheckAckResponse, RecordVerdictRequest, RecordVerdictResponse, RepoSignalHealthReport,
    ReviewPayload, ReviewScope as ProtoReviewScope, ReviewSignature as ProtoReviewSignature,
    ReviewSummary, RiskSignal as ProtoRiskSignal, SignStateRequest, SignStateResponse,
    SignalAnchor as ProtoSignalAnchor, SignalHealthEntry as ProtoSignalHealthEntry, SigningFooter,
    Verdict as ProtoVerdict, state_review_service_server::StateReviewService,
};
use crypto::verify_payload_signature;
use objects::{
    lock::RepositoryLockExt,
    object::{
        Blob, DiffKind, Discussion, DiscussionResolution, DiscussionsBlob, ReviewKind, ReviewScope,
        ReviewSignature, ReviewSignaturesBlob, RiskSignalBlob, State, StateAttachment,
        StateAttachmentBody, StateId, SymbolAnchor, signing_payload,
    },
    store::ObjectStore,
    worktree::diff_blobs,
};
use prost::Message;
use repo::{Repository, StateAttachmentKind};
use tonic::{Request, Response, Status};

use super::{GrpcLocalService, to_status, with_idempotency};

/// Maximum drift (seconds) between the client's `signed_at_unix` and the
/// server's wall clock. Generous enough to absorb NTP skew, narrow enough
/// to bound the window for replay-style attacks.
const SIGN_TIMESTAMP_SKEW_SECS: i64 = 5 * 60;

/// Verdicts use a distinct canonical payload so they cannot be replayed as
/// ordinary positive review signatures.
const VERDICT_SIGNING_PAYLOAD_TAG: &[u8] = b"hd-rev-sig-v2\x00";

/// Collision-safe marker for verdict metadata stored in the existing review
/// signature justification field.
const VERDICT_ENVELOPE_TAG: &str = "\u{1}hd-verdict-v1\u{1}";

/// Local-mode `StateReviewService` implementation.
#[derive(Clone)]
pub struct LocalStateReviewService {
    inner: GrpcLocalService,
}

impl LocalStateReviewService {
    pub fn new(inner: GrpcLocalService) -> Self {
        Self { inner }
    }
}

#[tonic::async_trait]
impl StateReviewService for LocalStateReviewService {
    async fn get_repo_signal_health(
        &self,
        request: Request<GetRepoSignalHealthRequest>,
    ) -> Result<Response<RepoSignalHealthReport>, Status> {
        let report =
            super::get_repo_signal_health(self.inner.repo(), request.into_inner().window_states)
                .map_err(to_status)?;
        Ok(Response::new(RepoSignalHealthReport {
            entries: report
                .entries
                .into_iter()
                .map(|entry| ProtoSignalHealthEntry {
                    module_id: entry.module_id,
                    fire_rate: entry.fire_rate,
                    warn: entry.warn,
                })
                .collect(),
            window_states: report.window_states,
        }))
    }

    async fn get_review_payload(
        &self,
        request: Request<GetReviewPayloadRequest>,
    ) -> Result<Response<ReviewPayload>, Status> {
        let req = request.into_inner();
        let state_id = parse_state_id(&req.state_id)?;
        let repo = self.inner.repo();
        let state = repo
            .store()
            .get_state(&state_id)
            .map_err(to_status)?
            .ok_or_else(|| {
                Status::not_found(format!("state {} not found", state_id.to_string_full()))
            })?;

        // Diff the state's tree against its first parent so the summary
        // counts reflect what actually changed in this state. The
        // signal registry / budgeter will eventually layer on top of
        // this; until then `files_changed` is the most useful single
        // number an agent can use for self-review.
        let diff_summary = compute_state_diff_summary(repo, &state).map_err(to_status)?;

        let summary = ReviewSummary {
            headline: state.intent.clone().unwrap_or_default(),
            files_changed: diff_summary.files_changed,
            added_lines: diff_summary.added_lines,
            removed_lines: diff_summary.removed_lines,
            in_budget_signal_count: 0,
            hidden_signal_count: 0,
        };

        let agent_narrative = if state.attribution.agent.is_some() {
            state.intent.clone().unwrap_or_default()
        } else {
            String::new()
        };

        // Surface fired risk signals if requested. The signal registry will
        // replace this with a proper budget split; until then everything we
        // read is "visible".
        let mut all_signals: Vec<ProtoRiskSignal> = Vec::new();
        if req.include_all_signals
            && let Some(hash) =
                attachment_hash(repo, &state.state_id, StateAttachmentKind::RiskSignals)?
            && let Some(blob) = repo.store().get_blob(&hash).map_err(to_status)?
        {
            let decoded = RiskSignalBlob::decode(blob.content())
                .map_err(|err| Status::internal(format!("decode risk signals: {err}")))?;
            all_signals = decoded
                .signals
                .into_iter()
                .map(|s| risk_signal_to_proto(s, "visible"))
                .collect();
        }

        // Synthesize a structured `diff_summary` signal so the
        // `in_budget_signals` array is non-empty even before the real
        // signal registry is wired up. Anchored on each modified
        // file (capped) so consumers can iterate without losing the
        // summary aggregate. This is a deliberate stable shape: agents
        // already iterating signals get a usable record per file
        // change, and the registry-driven path will simply layer real
        // signals alongside it.
        let mut in_budget_signals: Vec<ProtoRiskSignal> = Vec::new();
        let summary_kind = match (
            diff_summary.added_files,
            diff_summary.modified_files,
            diff_summary.deleted_files,
        ) {
            (a, 0, 0) if a > 0 => "diff_summary.added_only",
            (0, m, 0) if m > 0 => "diff_summary.modified_only",
            (0, 0, d) if d > 0 => "diff_summary.deleted_only",
            (0, 0, 0) => "diff_summary.empty",
            _ => "diff_summary.mixed",
        };
        let summary_reason = format!(
            "{} files changed (+{}/-{}, {} added, {} modified, {} deleted)",
            diff_summary.files_changed,
            diff_summary.added_lines,
            diff_summary.removed_lines,
            diff_summary.added_files,
            diff_summary.modified_files,
            diff_summary.deleted_files,
        );
        // Per-file anchors keep the array reasoning-friendly when
        // many files change, but cap so very large diffs don't bloat
        // the payload. The aggregate summary always rides on the
        // first entry's reason field; the rest carry per-file deltas.
        const MAX_DIFF_SIGNAL_ANCHORS: usize = 32;
        if diff_summary.changed_paths.is_empty() {
            in_budget_signals.push(ProtoRiskSignal {
                kind: summary_kind.to_string(),
                anchor: Some(ProtoSignalAnchor {
                    file: String::new(),
                    symbol: String::new(),
                    start_line: 0,
                    end_line: 0,
                }),
                reason: summary_reason.clone(),
                producer_module: "review_show.diff_summary".to_string(),
                producer_version: 1,
                computed_at: None,
                visibility: "visible".to_string(),
            });
        } else {
            for (idx, path_kind) in diff_summary
                .changed_paths
                .iter()
                .take(MAX_DIFF_SIGNAL_ANCHORS)
                .enumerate()
            {
                let reason = if idx == 0 {
                    summary_reason.clone()
                } else {
                    format!("{} ({})", path_kind.path, path_kind.kind_str())
                };
                in_budget_signals.push(ProtoRiskSignal {
                    kind: summary_kind.to_string(),
                    anchor: Some(ProtoSignalAnchor {
                        file: path_kind.path.clone(),
                        symbol: String::new(),
                        start_line: 0,
                        end_line: 0,
                    }),
                    reason,
                    producer_module: "review_show.diff_summary".to_string(),
                    producer_version: 1,
                    computed_at: None,
                    visibility: "visible".to_string(),
                });
            }
        }

        // Server-side reading-order partition. Same per-symbol
        // extraction logic as the hosted handler: tree-sitter when the
        // `semantic` feature is enabled, path-only fallback otherwise.
        let symbols = changed_files_as_symbols(repo, &state, &diff_summary.changed_paths)
            .map_err(to_status)?;
        let partition = build_review_payload_partition(&symbols);

        // Project the state's `DiscussionsBlob` (when present)
        // into the wire-shape `AnchoredDiscussion` list.
        let discussions =
            match attachment_hash(repo, &state.state_id, StateAttachmentKind::Discussions)? {
                Some(hash) => {
                    let blob = repo
                        .store()
                        .get_blob(&hash)
                        .map_err(to_status)?
                        .ok_or_else(|| {
                            Status::internal(format!(
                                "discussions blob {} referenced by state {} is missing",
                                hash,
                                state.state_id.to_string_full()
                            ))
                        })?;
                    let decoded = DiscussionsBlob::decode(blob.content())
                        .map_err(|err| Status::internal(format!("decode discussions: {err}")))?;
                    decoded
                        .discussions
                        .iter()
                        .map(discussion_to_anchored_proto)
                        .collect()
                }
                None => Vec::<AnchoredDiscussion>::new(),
            };

        let mut summary = summary;
        summary.in_budget_signal_count = in_budget_signals.len() as u32;
        summary.hidden_signal_count =
            all_signals.len().saturating_sub(in_budget_signals.len()) as u32;

        let payload = ReviewPayload {
            state_id: req.state_id.clone(),
            summary: Some(summary),
            agent_narrative,
            partition: Some(partition_to_proto(partition)),
            in_budget_signals,
            all_signals,
            tick_budget: 3,
            discussions,
            // Local mode has no policy registry — merge requirements
            // are surfaced only by the hosted handler.
            merge_requirements: Vec::<MergeRequirement>::new(),
            signing_footer: Some(SigningFooter {
                available_kinds: vec![
                    api::heddle::api::v1alpha1::ReviewKind::Read as i32,
                    api::heddle::api::v1alpha1::ReviewKind::AgentPreview as i32,
                    api::heddle::api::v1alpha1::ReviewKind::AgentCoReview as i32,
                ],
            }),
        };

        Ok(Response::new(payload))
    }

    async fn sign_state(
        &self,
        request: Request<SignStateRequest>,
    ) -> Result<Response<SignStateResponse>, Status> {
        let req = request.into_inner();
        let req_bytes = req.encode_to_vec();
        let client_operation_id = req.client_operation_id.clone();
        let inner = self.inner.clone();

        let response = with_idempotency(
            &self.inner,
            &client_operation_id,
            "state_review.sign_state",
            &req_bytes,
            move || {
                let inner = inner.clone();
                async move { execute_sign_state(&inner, req).await }
            },
        )
        .await?;

        Ok(Response::new(response))
    }

    async fn list_signatures(
        &self,
        request: Request<ListSignaturesRequest>,
    ) -> Result<Response<ListSignaturesResponse>, Status> {
        let req = request.into_inner();
        let state_id = parse_state_id(&req.state_id)?;
        let repo = self.inner.repo();
        let state = repo
            .store()
            .get_state(&state_id)
            .map_err(to_status)?
            .ok_or_else(|| {
                Status::not_found(format!("state {} not found", state_id.to_string_full()))
            })?;

        let signatures =
            match attachment_hash(repo, &state.state_id, StateAttachmentKind::ReviewSignatures)? {
                Some(hash) => {
                    let blob = repo
                        .store()
                        .get_blob(&hash)
                        .map_err(to_status)?
                        .ok_or_else(|| {
                            Status::internal(format!(
                                "review signatures blob {} missing from object store",
                                hash
                            ))
                        })?;
                    let decoded = ReviewSignaturesBlob::decode(blob.content()).map_err(|err| {
                        Status::internal(format!("decode review signatures: {err}"))
                    })?;
                    decoded
                        .signatures
                        .into_iter()
                        .enumerate()
                        .map(|(idx, sig)| {
                            review_signature_to_proto(sig, synthetic_signature_id(idx))
                        })
                        .collect()
                }
                None => Vec::new(),
            };

        Ok(Response::new(ListSignaturesResponse { signatures }))
    }

    async fn record_verdict(
        &self,
        request: Request<RecordVerdictRequest>,
    ) -> Result<Response<RecordVerdictResponse>, Status> {
        let req = request.into_inner();
        let req_bytes = req.encode_to_vec();
        let client_operation_id = req.client_operation_id.clone();
        let inner = self.inner.clone();

        let response = with_idempotency(
            &self.inner,
            &client_operation_id,
            "state_review.record_verdict",
            &req_bytes,
            move || {
                let inner = inner.clone();
                async move { execute_record_verdict(&inner, req).await }
            },
        )
        .await?;

        Ok(Response::new(response))
    }

    async fn record_check_ack(
        &self,
        _request: Request<RecordCheckAckRequest>,
    ) -> Result<Response<RecordCheckAckResponse>, Status> {
        Err(Status::unimplemented(
            "record_check_ack is not available in local mode: local repositories do not persist review check acknowledgements",
        ))
    }

    async fn get_review_progress(
        &self,
        _request: Request<GetReviewProgressRequest>,
    ) -> Result<Response<GetReviewProgressResponse>, Status> {
        Err(Status::unimplemented(
            "get_review_progress is not available in local mode: local repositories do not persist review progress",
        ))
    }
}

/// Body of [`LocalStateReviewService::sign_state`]. Lifted out of the trait
/// method so [`with_idempotency`] can re-execute it inside its closure.
async fn execute_sign_state(
    inner: &GrpcLocalService,
    req: SignStateRequest,
) -> Result<SignStateResponse, Status> {
    // 1. Validate the kind.
    let kind = match api::heddle::api::v1alpha1::ReviewKind::try_from(req.kind)
        .map_err(|_| Status::invalid_argument(format!("unknown review kind tag {}", req.kind)))?
    {
        api::heddle::api::v1alpha1::ReviewKind::Read => ReviewKind::Read,
        api::heddle::api::v1alpha1::ReviewKind::AgentPreview => ReviewKind::AgentPreview,
        api::heddle::api::v1alpha1::ReviewKind::AgentCoReview => ReviewKind::AgentCoReview,
        api::heddle::api::v1alpha1::ReviewKind::Unspecified => {
            return Err(Status::invalid_argument("review kind is required"));
        }
    };

    // 2. Locate the state.
    let state_id = parse_state_id(&req.state_id)?;
    let repo = inner.repo();

    // 3. Translate the scope.
    let scope = match req.scope.as_ref() {
        Some(s) => proto_scope_to_object(s)?,
        None => ReviewScope::WholeChange,
    };

    // 4. Build the ReviewSignature, then verify the client-supplied
    // signature is well-formed and matches the deterministic signing
    // payload. A malformed or forged signature must never reach the
    // persisted blob. Attribute the signature to the local-mode
    // caller (`Repository::get_principal` resolves env vars then
    // `[principal]` in `.heddle/config.toml`), not the state's author
    // — Bob signing Alice's state should record Bob.
    let actor = repo
        .get_principal()
        .map_err(|err| Status::internal(format!("resolve caller principal: {err}")))?;
    if req.justification.starts_with(VERDICT_ENVELOPE_TAG) {
        return Err(Status::invalid_argument(
            "justification must not begin with the reserved verdict-envelope prefix",
        ));
    }
    let justification = if req.justification.is_empty() {
        None
    } else {
        Some(req.justification.clone())
    };

    let now = chrono::Utc::now().timestamp();
    let signed_at = req.signed_at.as_ref().map(|t| t.seconds).unwrap_or(0);
    if signed_at == 0 {
        return Err(Status::invalid_argument(
            "signed_at is required and must match the timestamp the client signed over",
        ));
    }
    if (signed_at - now).abs() > SIGN_TIMESTAMP_SKEW_SECS {
        return Err(Status::invalid_argument(format!(
            "signed_at={signed_at} is too far from server time={now} (max skew {SIGN_TIMESTAMP_SKEW_SECS}s)"
        )));
    }

    let new_sig = ReviewSignature {
        actor,
        kind,
        scope: scope.clone(),
        justification: justification.clone(),
        signed_at,
        algorithm: req.algorithm.clone(),
        public_key: hex::encode(&req.public_key),
        signature: hex::encode(&req.signature),
    };
    new_sig
        .validate()
        .map_err(|err| Status::invalid_argument(format!("invalid review signature: {err}")))?;

    let public_key_bytes = req.public_key.clone();
    let signature_bytes = req.signature.clone();
    let payload = signing_payload(state_id, kind, &scope, signed_at, justification.as_deref());
    verify_payload_signature(
        &payload,
        &req.algorithm,
        &public_key_bytes,
        &signature_bytes,
    )
    .map_err(|err| {
        Status::invalid_argument(format!(
            "review signature failed verification ({}): {err}",
            req.algorithm
        ))
    })?;

    let new_index = append_review_signature(repo, state_id, new_sig)?;

    Ok(SignStateResponse {
        signature_id: synthetic_signature_id(new_index),
        state_id: req.state_id,
    })
}

async fn execute_record_verdict(
    inner: &GrpcLocalService,
    req: RecordVerdictRequest,
) -> Result<RecordVerdictResponse, Status> {
    let verdict = verdict_str_from_proto(req.verdict)?;
    let state_id = parse_state_id(&req.state_id)?;
    if matches!(verdict, "hold" | "reject") && req.reason.trim().is_empty() {
        return Err(Status::invalid_argument(
            "reason is required for HOLD and REJECT verdicts",
        ));
    }
    let scope = match req.scope.as_ref() {
        Some(scope) => proto_scope_to_object(scope)?,
        None => ReviewScope::WholeChange,
    };
    let signed_at = req.signed_at.as_ref().map(|time| time.seconds).unwrap_or(0);
    if signed_at == 0 {
        return Err(Status::invalid_argument(
            "signed_at is required and must match the timestamp the client signed over",
        ));
    }
    let now = chrono::Utc::now().timestamp();
    if (signed_at - now).abs() > SIGN_TIMESTAMP_SKEW_SECS {
        return Err(Status::invalid_argument(format!(
            "signed_at={signed_at} is too far from server time={now} (max skew {SIGN_TIMESTAMP_SKEW_SECS}s)"
        )));
    }

    let payload = verdict_signing_payload(state_id, verdict, &scope, signed_at, &req.reason);
    verify_payload_signature(&payload, &req.algorithm, &req.public_key, &req.signature).map_err(
        |err| {
            Status::invalid_argument(format!(
                "verdict signature failed verification ({}): {err}",
                req.algorithm
            ))
        },
    )?;

    let repo = inner.repo();
    let signature = ReviewSignature {
        actor: repo
            .get_principal()
            .map_err(|err| Status::internal(format!("resolve caller principal: {err}")))?,
        kind: ReviewKind::Read,
        scope,
        justification: Some(encode_verdict_envelope(verdict, &req.reason)),
        signed_at,
        algorithm: req.algorithm.clone(),
        public_key: hex::encode(&req.public_key),
        signature: hex::encode(&req.signature),
    };
    signature
        .validate()
        .map_err(|err| Status::invalid_argument(format!("invalid review signature: {err}")))?;

    let new_index = append_review_signature(repo, state_id, signature)?;
    Ok(RecordVerdictResponse {
        signature_id: synthetic_signature_id(new_index),
        state_id: req.state_id,
    })
}

/// Append one signed review record while holding the repository write lock.
/// SignState and RecordVerdict share this tail so concurrent local writes do
/// not lose one another's attachment updates.
fn append_review_signature(
    repo: &Repository,
    state_id: StateId,
    signature: ReviewSignature,
) -> Result<usize, Status> {
    let _lock = repo
        .locker()
        .write()
        .map_err(|err| Status::internal(err.to_string()))?;
    repo.store()
        .get_state(&state_id)
        .map_err(to_status)?
        .ok_or_else(|| {
            Status::not_found(format!("state {} not found", state_id.to_string_full()))
        })?;
    let prior = repo
        .latest_state_attachment(&state_id, StateAttachmentKind::ReviewSignatures)
        .map_err(to_status)?;
    let mut blob = match prior.as_ref().map(|attachment| {
        let StateAttachmentBody::ReviewSignatures(hash) = &attachment.body else {
            unreachable!()
        };
        *hash
    }) {
        Some(hash) => {
            let raw = repo
                .store()
                .get_blob(&hash)
                .map_err(to_status)?
                .ok_or_else(|| {
                    Status::internal(format!(
                        "existing review signatures blob {} missing from object store",
                        hash
                    ))
                })?;
            ReviewSignaturesBlob::decode(raw.content())
                .map_err(|err| Status::internal(format!("decode review signatures: {err}")))?
        }
        None => ReviewSignaturesBlob::new(Vec::new()),
    };
    blob.signatures.push(signature);
    let new_index = blob.signatures.len() - 1;

    let bytes = blob
        .encode()
        .map_err(|err| Status::internal(format!("encode review signatures: {err}")))?;
    let content_hash = repo
        .store()
        .put_blob(&Blob::new(bytes))
        .map_err(to_status)?;

    let attachment = StateAttachment {
        state_id,
        body: StateAttachmentBody::ReviewSignatures(content_hash),
        attribution: repo.get_attribution().map_err(to_status)?,
        created_at: chrono::Utc::now(),
        supersedes: prior.map(|attachment| attachment.id()),
    };
    repo.put_state_attachment(&attachment).map_err(to_status)?;
    Ok(new_index)
}

/// `ReviewSignature` doesn't carry an explicit id; we synthesise one from
/// the per-state index so the wire surface has stable signature ids within a
/// single state. (A future schema bump may add an explicit id.)
fn synthetic_signature_id(index: usize) -> String {
    format!("rs-{index}")
}

fn attachment_hash(
    repo: &Repository,
    state_id: &StateId,
    kind: StateAttachmentKind,
) -> Result<Option<objects::object::ContentHash>, Status> {
    let Some(attachment) = repo
        .latest_state_attachment(state_id, kind)
        .map_err(to_status)?
    else {
        return Ok(None);
    };
    let hash = match attachment.body {
        StateAttachmentBody::RiskSignals(hash)
        | StateAttachmentBody::ReviewSignatures(hash)
        | StateAttachmentBody::Discussions(hash) => hash,
        _ => unreachable!(),
    };
    Ok(Some(hash))
}

fn parse_state_id(
    state_id: &Option<api::heddle::api::v1alpha1::StateId>,
) -> Result<StateId, Status> {
    let bytes = state_id
        .as_ref()
        .ok_or_else(|| Status::invalid_argument("state_id is required"))?;
    StateId::try_from_slice(&bytes.value)
        .map_err(|err| Status::invalid_argument(format!("invalid state_id: {err}")))
}

fn api_state_id(state_id: &StateId) -> api::heddle::api::v1alpha1::StateId {
    api::heddle::api::v1alpha1::StateId {
        value: state_id.as_bytes().to_vec(),
    }
}

fn verdict_signing_payload(
    state_id: StateId,
    verdict: &str,
    scope: &ReviewScope,
    signed_at: i64,
    reason: &str,
) -> Vec<u8> {
    let mut payload = Vec::with_capacity(VERDICT_SIGNING_PAYLOAD_TAG.len() + 256);
    payload.extend_from_slice(VERDICT_SIGNING_PAYLOAD_TAG);
    payload.extend_from_slice(state_id.to_string_full().as_bytes());
    payload.push(0);
    payload.extend_from_slice(verdict.as_bytes());
    payload.push(0);
    match scope {
        ReviewScope::WholeChange => {
            payload.extend_from_slice(b"whole_change");
            payload.push(0);
        }
        ReviewScope::Symbols(symbols) => {
            payload.extend_from_slice(b"symbols");
            payload.push(0);
            payload.extend_from_slice(&(symbols.len() as u32).to_le_bytes());
            for symbol in symbols {
                payload.extend_from_slice(symbol.file.as_bytes());
                payload.push(0);
                payload.extend_from_slice(symbol.symbol.as_bytes());
                payload.push(0);
            }
        }
    }
    payload.extend_from_slice(&signed_at.to_le_bytes());
    payload.extend_from_slice(reason.as_bytes());
    payload.push(0);
    payload
}

fn verdict_str_from_proto(tag: i32) -> Result<&'static str, Status> {
    match ProtoVerdict::try_from(tag)
        .map_err(|_| Status::invalid_argument(format!("unknown verdict tag {tag}")))?
    {
        ProtoVerdict::Sign => Ok("sign"),
        ProtoVerdict::Hold => Ok("hold"),
        ProtoVerdict::Reject => Ok("reject"),
        ProtoVerdict::Unspecified => Err(Status::invalid_argument("verdict is required")),
    }
}

fn encode_verdict_envelope(verdict: &str, reason: &str) -> String {
    let body = serde_json::json!({ "verdict": verdict, "reason": reason });
    format!("{VERDICT_ENVELOPE_TAG}{body}")
}

fn decode_verdict_envelope(justification: &str) -> Option<(String, String)> {
    let body = justification.strip_prefix(VERDICT_ENVELOPE_TAG)?;
    let value: serde_json::Value = serde_json::from_str(body).ok()?;
    let verdict = value.get("verdict")?.as_str()?.to_string();
    let reason = value
        .get("reason")
        .and_then(|reason| reason.as_str())
        .unwrap_or_default()
        .to_string();
    Some((verdict, reason))
}

fn verdict_proto_from_str(verdict: &str) -> ProtoVerdict {
    match verdict {
        "sign" => ProtoVerdict::Sign,
        "hold" => ProtoVerdict::Hold,
        "reject" => ProtoVerdict::Reject,
        _ => ProtoVerdict::Unspecified,
    }
}

fn proto_scope_to_object(scope: &ProtoReviewScope) -> Result<ReviewScope, Status> {
    use api::heddle::api::v1alpha1::review_scope::Scope;
    match scope.scope.as_ref() {
        None | Some(Scope::WholeChange(_)) => Ok(ReviewScope::WholeChange),
        Some(Scope::Symbols(list)) => {
            if list.symbols.is_empty() {
                return Err(Status::invalid_argument(
                    "symbols scope requires at least one symbol anchor",
                ));
            }
            let symbols = list
                .symbols
                .iter()
                .map(|s| SymbolAnchor::new(s.file.clone(), s.symbol.clone()))
                .collect();
            Ok(ReviewScope::Symbols(symbols))
        }
    }
}

fn object_scope_to_proto(scope: &ReviewScope) -> ProtoReviewScope {
    use api::heddle::api::v1alpha1::review_scope::{Scope, SymbolList, WholeChange};
    let inner = match scope {
        ReviewScope::WholeChange => Scope::WholeChange(WholeChange {}),
        ReviewScope::Symbols(symbols) => Scope::Symbols(SymbolList {
            symbols: symbols
                .iter()
                .map(|s| ProtoPathSymbolRef {
                    file: s.file.clone(),
                    symbol: s.symbol.clone(),
                })
                .collect(),
        }),
    };
    ProtoReviewScope { scope: Some(inner) }
}

fn review_signature_to_proto(sig: ReviewSignature, signature_id: String) -> ProtoReviewSignature {
    let (verdict, reason, justification) = match sig
        .justification
        .as_deref()
        .and_then(decode_verdict_envelope)
    {
        Some((verdict, reason)) => (
            verdict_proto_from_str(&verdict) as i32,
            reason,
            String::new(),
        ),
        None => (
            ProtoVerdict::Unspecified as i32,
            String::new(),
            sig.justification.unwrap_or_default(),
        ),
    };
    ProtoReviewSignature {
        signature_id,
        actor_name: sig.actor.name.clone(),
        actor_email: sig.actor.email.clone(),
        kind: review_kind_to_proto(sig.kind) as i32,
        scope: Some(object_scope_to_proto(&sig.scope)),
        justification,
        signed_at: Some(prost_types::Timestamp {
            seconds: sig.signed_at,
            nanos: 0,
        }),
        algorithm: sig.algorithm,
        public_key: hex::decode(&sig.public_key).unwrap_or_default(),
        signature: hex::decode(&sig.signature).unwrap_or_default(),
        verdict,
        reason,
    }
}

fn review_kind_to_proto(kind: ReviewKind) -> api::heddle::api::v1alpha1::ReviewKind {
    match kind {
        ReviewKind::Read => api::heddle::api::v1alpha1::ReviewKind::Read,
        ReviewKind::AgentPreview => api::heddle::api::v1alpha1::ReviewKind::AgentPreview,
        ReviewKind::AgentCoReview => api::heddle::api::v1alpha1::ReviewKind::AgentCoReview,
    }
}

fn risk_signal_to_proto(sig: objects::object::RiskSignal, visibility: &str) -> ProtoRiskSignal {
    let (start_line, end_line) = sig.anchor.line_range.unwrap_or((0, 0));
    ProtoRiskSignal {
        kind: sig.kind.as_str().to_string(),
        anchor: Some(ProtoSignalAnchor {
            file: sig.anchor.file,
            symbol: sig.anchor.symbol.unwrap_or_default(),
            start_line,
            end_line,
        }),
        reason: sig.reason,
        producer_module: sig.producer.module,
        producer_version: sig.producer.version,
        computed_at: Some(prost_types::Timestamp {
            seconds: sig.computed_at,
            nanos: 0,
        }),
        visibility: visibility.to_string(),
    }
}

// ---------------------------------------------------------------------------
// Symbol extraction + discussion projection (mirrors hosted impl).
// ---------------------------------------------------------------------------

/// Symbol projection for the reading-order partition. Mirrors the hosted-handler
/// implementation: when the `semantic` feature is enabled and the
/// changed path has a tree-sitter parser and a readable new-side blob, emits
/// one [`PathSymbol`] per definition. Otherwise falls back to a single path-only
/// entry (kind = `Other`), which keeps deletes and gitlink pointer changes
/// visible even though they do not carry Heddle blob content.
fn changed_files_as_symbols(
    repo: &Repository,
    state: &State,
    changed_paths: &[ChangedPath],
) -> objects::error::Result<Vec<PathSymbol>> {
    let new_tree = match repo.store().get_tree(&state.tree)? {
        Some(t) => t,
        None => return Ok(Vec::new()),
    };
    let new_files = collect_files(repo, &new_tree, "")?;

    let mut out: Vec<PathSymbol> = Vec::new();
    for path_kind in changed_paths {
        let path = &path_kind.path;
        #[cfg_attr(not(feature = "semantic"), allow(unused_mut))]
        let mut emitted_any = false;
        if let Some(hash) = new_files.get(path) {
            #[cfg(feature = "semantic")]
            {
                if let Some(blob) = repo.store().get_blob(hash)? {
                    emitted_any = extract_file_symbols(path, blob.content(), &mut out);
                }
            }
            #[cfg(not(feature = "semantic"))]
            {
                let _ = hash;
            }
        }
        if !emitted_any {
            out.push(PathSymbol {
                file: path.clone(),
                symbol: path.clone(),
                kind: SymbolKind::Other,
            });
        }
    }
    Ok(out)
}

#[cfg(feature = "semantic")]
fn extract_file_symbols(path: &str, source: &[u8], out: &mut Vec<PathSymbol>) -> bool {
    use ::semantic::symbol_resolver::{Definition, DefinitionKind, extract_definitions};
    let definitions: Vec<Definition> = match extract_definitions(source, std::path::Path::new(path))
    {
        Ok(defs) => defs,
        Err(_) => return false,
    };
    if definitions.is_empty() {
        return false;
    }
    for d in definitions {
        let kind = match d.kind {
            DefinitionKind::Type => SymbolKind::Type,
            DefinitionKind::Trait => SymbolKind::Trait,
            DefinitionKind::Class => SymbolKind::Class,
            DefinitionKind::Interface => SymbolKind::Interface,
            DefinitionKind::TypeAlias => SymbolKind::TypeAlias,
            DefinitionKind::EnumDef => SymbolKind::EnumDef,
            DefinitionKind::ConstDecl => SymbolKind::ConstDecl,
            DefinitionKind::Module => SymbolKind::Module,
            DefinitionKind::Function => SymbolKind::Function,
            DefinitionKind::Other => SymbolKind::Other,
        };
        let symbol = match d.parent_name.as_deref() {
            Some(parent) if !parent.is_empty() => format!("{parent}::{}", d.name),
            _ => d.name,
        };
        out.push(PathSymbol {
            file: path.to_string(),
            symbol,
            kind,
        });
    }
    true
}

fn collect_files(
    repo: &Repository,
    tree: &objects::object::Tree,
    prefix: &str,
) -> objects::error::Result<std::collections::HashMap<String, objects::object::ContentHash>> {
    let mut out = std::collections::HashMap::new();
    for entry in tree.entries() {
        let path = if prefix.is_empty() {
            entry.name().to_string()
        } else {
            format!("{prefix}/{}", entry.name())
        };
        if entry.is_tree() {
            if let Some(hash) = entry.tree_hash()
                && let Some(subtree) = repo.store().get_tree(&hash)?
            {
                let sub = collect_files(repo, &subtree, &path)?;
                out.extend(sub);
            }
        } else if let Some(hash) = entry.content_hash() {
            out.insert(path, hash);
        }
    }
    Ok(out)
}

fn partition_to_proto(p: ReadingOrderPartition) -> ProtoReadingOrderPartition {
    ProtoReadingOrderPartition {
        structural: p.structural.iter().map(path_symbol_to_proto).collect(),
        consequence: p.consequence.iter().map(path_symbol_to_proto).collect(),
        tests_and_docs: p.tests_and_docs.iter().map(path_symbol_to_proto).collect(),
    }
}

fn path_symbol_to_proto(p: &PathSymbol) -> ProtoPathSymbolRef {
    ProtoPathSymbolRef {
        file: p.file.clone(),
        symbol: p.symbol.clone(),
    }
}

fn discussion_to_anchored_proto(d: &Discussion) -> AnchoredDiscussion {
    AnchoredDiscussion {
        id: d.id.clone(),
        anchor: Some(ProtoPathSymbolRef {
            file: d.anchor.file.clone(),
            symbol: d.anchor.symbol.clone(),
        }),
        opened_against_state: Some(api_state_id(&d.opened_against_state)),
        opened_at: Some(prost_types::Timestamp {
            seconds: d.opened_at,
            nanos: 0,
        }),
        turns: d
            .turns
            .iter()
            .map(|t| api::heddle::api::v1alpha1::DiscussionTurn {
                author_name: t.author.name.clone(),
                author_email: t.author.email.clone(),
                body: t.body.clone(),
                posted_at: Some(prost_types::Timestamp {
                    seconds: t.posted_at,
                    nanos: 0,
                }),
            })
            .collect(),
        resolution: Some(discussion_resolution_to_proto(&d.resolution)),
        body_changed_since_open: d.body_changed_since_open,
        orphaned: d.orphaned,
        visibility: d.visibility.as_str().to_string(),
    }
}

fn discussion_resolution_to_proto(
    resolution: &DiscussionResolution,
) -> api::heddle::api::v1alpha1::DiscussionResolution {
    use api::heddle::api::v1alpha1::discussion_resolution::{
        Dismissed, Open, ResolvedByEdit, ResolvedIntoAnnotation, State,
    };
    let state = match resolution {
        DiscussionResolution::Open => State::Open(Open {}),
        DiscussionResolution::ResolvedIntoAnnotation { annotation_id } => {
            State::IntoAnnotation(ResolvedIntoAnnotation {
                annotation_id: annotation_id.clone(),
            })
        }
        DiscussionResolution::ResolvedByEdit { state_id } => State::ByEdit(ResolvedByEdit {
            state_id: Some(api_state_id(state_id)),
        }),
        DiscussionResolution::Dismissed { reason } => State::Dismissed(Dismissed {
            reason: reason.clone(),
        }),
    };
    api::heddle::api::v1alpha1::DiscussionResolution { state: Some(state) }
}

// ---------------------------------------------------------------------------
// Diff summary helpers (state.tree vs first parent's tree).
// ---------------------------------------------------------------------------

/// File-change kinds we surface in the diff summary signal anchors.
/// Mirrors `objects::object::DiffKind` minus the `Unchanged` variant
/// (we filter those out before constructing this).
#[derive(Debug, Clone)]
struct ChangedPath {
    path: String,
    kind: DiffKind,
}

impl ChangedPath {
    fn kind_str(&self) -> &'static str {
        match self.kind {
            DiffKind::Added => "added",
            DiffKind::Modified => "modified",
            DiffKind::Deleted => "deleted",
            DiffKind::Unchanged => "unchanged",
        }
    }
}

/// Aggregated counts plus a path list, computed by diffing
/// `state.tree` against the first parent's tree (or empty when the
/// state is a root). When `state.parents` is empty every file in the
/// state's tree counts as added, which makes "first capture" reviews
/// non-empty too. The `_state` prefix on `_state` is intentional: the
/// helper currently only reads `state.tree` and `state.parents`.
struct DiffSummary {
    files_changed: u32,
    added_files: u32,
    modified_files: u32,
    deleted_files: u32,
    added_lines: u32,
    removed_lines: u32,
    changed_paths: Vec<ChangedPath>,
}

/// Compute a summary diff for `state` vs its first parent. Errors
/// from the object store propagate; missing trees / blobs are skipped
/// silently (treated as zero-change for that path) so a partially
/// pruned object store never blocks the review surface. The
/// distinction matters: missing-object errors must become zero (the
/// summary is best-effort, callers want a payload they can render),
/// but genuine I/O errors must still propagate so a corrupt store
/// surfaces loudly instead of silently truncating the review.
fn compute_state_diff_summary(
    repo: &Repository,
    state: &State,
) -> objects::error::Result<DiffSummary> {
    use objects::object::Tree;
    let parent_tree_hash = if let Some(parent_id) = state.parents.first() {
        match repo.store().get_state(parent_id)? {
            Some(parent_state) => parent_state.tree,
            None => Tree::new().hash(),
        }
    } else {
        Tree::new().hash()
    };

    // Resolve both tree objects up front so the missing-tree case
    // becomes a synthesized empty changeset rather than an error from
    // the recursive diff. `get_tree` returns `Ok(None)` for missing
    // (not an error), and propagates only on genuine I/O — matching
    // the policy the doc-comment claims.
    let parent_tree_obj = repo.store().get_tree(&parent_tree_hash)?;
    let new_tree_obj = repo.store().get_tree(&state.tree)?;

    // If either tree is missing from the local store the diff is not
    // meaningful — return an empty summary instead of erroring out.
    // This mirrors the "Modified branch tolerates missing blobs" stance
    // for the *tree* level: a partially pruned store should never block
    // review payload retrieval, only render an empty summary.
    let changes = if parent_tree_obj.is_some() && new_tree_obj.is_some() {
        repo.diff_trees(&parent_tree_hash, &state.tree)?
    } else {
        objects::object::FileChangeSet::new()
    };

    // Compute per-file line deltas. We only count `Modified` here for
    // the symmetric add/remove totals; `Added` files contribute every
    // line as an add, and `Deleted` files contribute every line as a
    // remove. Files with non-utf8 contents (e.g. binaries) silently
    // contribute zero — `diff_blobs` already returns an empty vec in
    // that case, and we mirror the same behavior for raw line counts.
    let mut added_lines: u32 = 0;
    let mut removed_lines: u32 = 0;
    let mut changed_paths: Vec<ChangedPath> = Vec::with_capacity(changes.len());

    let parent_files = match parent_tree_obj.as_ref() {
        Some(t) => collect_files(repo, t, "")?,
        None => std::collections::HashMap::new(),
    };
    let new_files = match new_tree_obj.as_ref() {
        Some(t) => collect_files(repo, t, "")?,
        None => std::collections::HashMap::new(),
    };

    let mut added_files: u32 = 0;
    let mut modified_files: u32 = 0;
    let mut deleted_files: u32 = 0;

    for change in changes.iter() {
        match change.kind {
            DiffKind::Added => {
                added_files += 1;
                // Missing blob (`get_blob` returns `Ok(None)`) → file
                // counts but contributes zero lines. Genuine I/O
                // errors still propagate via `?` — same shape as the
                // Modified branch's intent, but here we keep the
                // distinction explicit so a corrupt store surfaces
                // rather than getting silently swallowed.
                if let Some(hash) = new_files.get(&change.path)
                    && let Some(blob) = repo.store().get_blob(hash)?
                {
                    added_lines = added_lines.saturating_add(line_count(blob.content()));
                }
            }
            DiffKind::Deleted => {
                deleted_files += 1;
                if let Some(hash) = parent_files.get(&change.path)
                    && let Some(blob) = repo.store().get_blob(hash)?
                {
                    removed_lines = removed_lines.saturating_add(line_count(blob.content()));
                }
            }
            DiffKind::Modified => {
                modified_files += 1;
                // `get_blob` already returns `Ok(None)` for a missing
                // blob, so `?` here only fires on genuine I/O. Match
                // the Added/Deleted branches' propagation policy
                // explicitly instead of the older `.ok().flatten()`
                // form, which silently swallowed IO errors and
                // conflated them with "missing".
                let old_blob = match parent_files.get(&change.path) {
                    Some(h) => repo.store().get_blob(h)?,
                    None => None,
                };
                let new_blob = match new_files.get(&change.path) {
                    Some(h) => repo.store().get_blob(h)?,
                    None => None,
                };
                if let (Some(old), Some(new)) = (old_blob, new_blob) {
                    for line in diff_blobs(&old, &new) {
                        match line {
                            objects::worktree::DiffLine::Added(_) => {
                                added_lines = added_lines.saturating_add(1);
                            }
                            objects::worktree::DiffLine::Removed(_) => {
                                removed_lines = removed_lines.saturating_add(1);
                            }
                            objects::worktree::DiffLine::Context(_) => {}
                        }
                    }
                }
            }
            DiffKind::Unchanged => continue,
        }
        changed_paths.push(ChangedPath {
            path: change.path.clone(),
            kind: change.kind,
        });
    }

    Ok(DiffSummary {
        files_changed: changed_paths.len() as u32,
        added_files,
        modified_files,
        deleted_files,
        added_lines,
        removed_lines,
        changed_paths,
    })
}

/// Count the number of newline-separated lines in a file blob. Binary
/// blobs (non-utf8) count as zero — we deliberately don't byte-count
/// them, since "lines" is meaningless for binary content. A trailing
/// newline does not introduce a phantom empty line.
fn line_count(content: &[u8]) -> u32 {
    let Ok(s) = std::str::from_utf8(content) else {
        return 0;
    };
    if s.is_empty() {
        return 0;
    }
    let trimmed = s.strip_suffix('\n').unwrap_or(s);
    if trimmed.is_empty() {
        return 1;
    }
    (trimmed.matches('\n').count() as u32).saturating_add(1)
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use api::heddle::api::v1alpha1::ReviewScope as ProtoReviewScope;
    use crypto::Signer as _;
    use repo::{Repository, operation_dedup::OperationDedupStore};
    use tempfile::TempDir;

    use super::*;

    fn fresh_service() -> (LocalStateReviewService, Arc<Repository>, TempDir) {
        let temp = TempDir::new().expect("create tempdir");
        // SAFETY: tests run with a controlled environment; setting these
        // env vars steers the default attribution into a predictable place.
        // SAFETY: setting env vars in tests; rust 2024 marks these unsafe.
        unsafe {
            std::env::set_var("HEDDLE_PRINCIPAL_NAME", "Alice Tester");
            std::env::set_var("HEDDLE_PRINCIPAL_EMAIL", "alice@example.com");
        }
        let repo = Repository::init_default(temp.path()).expect("init repo");
        let dedup = OperationDedupStore::open(repo.heddle_dir()).expect("open dedup");
        let repo = Arc::new(repo);
        let svc =
            LocalStateReviewService::new(GrpcLocalService::new(repo.clone(), Arc::new(dedup)));
        (svc, repo, temp)
    }

    fn capture_state(repo: &Repository) -> StateId {
        // Write a tiny file so snapshot has something to capture.
        std::fs::write(repo.root().join("hello.txt"), b"hi").expect("write file");
        let state = repo
            .snapshot(Some("seed".to_string()), None)
            .expect("snapshot");
        state.state_id
    }

    fn sign_request(state_id: &StateId, op_id: &str) -> SignStateRequest {
        let signer = crypto::Ed25519Signer::generate().expect("generate ed25519 key");
        let scope = ReviewScope::WholeChange;
        let signed_at = chrono::Utc::now().timestamp();
        let payload = signing_payload(*state_id, ReviewKind::Read, &scope, signed_at, None);
        let signature = signer.sign(&payload).expect("sign payload");
        use api::heddle::api::v1alpha1::review_scope::{Scope, WholeChange};
        SignStateRequest {
            repo_path: None,
            state_id: Some(api_state_id(state_id)),
            kind: api::heddle::api::v1alpha1::ReviewKind::Read as i32,
            scope: Some(ProtoReviewScope {
                scope: Some(Scope::WholeChange(WholeChange {})),
            }),
            justification: String::new(),
            algorithm: "ed25519".to_string(),
            public_key: signer.public_key().to_vec(),
            signature: signature.clone(),
            signed_at: Some(prost_types::Timestamp {
                seconds: signed_at,
                nanos: 0,
            }),
            client_operation_id: op_id.to_string(),
        }
    }

    fn verdict_request(
        state_id: &StateId,
        verdict: ProtoVerdict,
        reason: &str,
        op_id: &str,
    ) -> RecordVerdictRequest {
        let signer = crypto::Ed25519Signer::generate().expect("generate ed25519 key");
        let scope = ReviewScope::WholeChange;
        let signed_at = chrono::Utc::now().timestamp();
        let verdict_str = verdict_str_from_proto(verdict as i32).expect("explicit verdict");
        let payload = verdict_signing_payload(*state_id, verdict_str, &scope, signed_at, reason);
        let signature = signer.sign(&payload).expect("sign verdict payload");
        use api::heddle::api::v1alpha1::review_scope::{Scope, WholeChange};
        RecordVerdictRequest {
            repo_path: None,
            state_id: Some(api_state_id(state_id)),
            verdict: verdict as i32,
            scope: Some(ProtoReviewScope {
                scope: Some(Scope::WholeChange(WholeChange {})),
            }),
            reason: reason.to_string(),
            algorithm: "ed25519".to_string(),
            public_key: signer.public_key().to_vec(),
            signature,
            signed_at: Some(prost_types::Timestamp {
                seconds: signed_at,
                nanos: 0,
            }),
            client_operation_id: op_id.to_string(),
        }
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn sign_state_persists_to_review_signatures_blob() {
        let (svc, repo, _tmp) = fresh_service();
        let state_id = capture_state(&repo);

        let resp = svc
            .sign_state(Request::new(sign_request(&state_id, "")))
            .await
            .expect("sign_state");
        assert!(!resp.get_ref().signature_id.is_empty());
        assert_eq!(resp.get_ref().state_id, Some(api_state_id(&state_id)));

        let listing = svc
            .list_signatures(Request::new(ListSignaturesRequest {
                repo_path: None,
                state_id: Some(api_state_id(&state_id)),
            }))
            .await
            .expect("list_signatures");
        let sigs = &listing.get_ref().signatures;
        assert_eq!(sigs.len(), 1, "expected one signature, got {sigs:?}");
        assert_eq!(
            sigs[0].kind,
            api::heddle::api::v1alpha1::ReviewKind::Read as i32
        );
        assert_eq!(sigs[0].algorithm, "ed25519");
        assert_eq!(sigs[0].actor_name, "Alice Tester");
        assert_eq!(sigs[0].actor_email, "alice@example.com");
        let scope_case = sigs[0].scope.as_ref().and_then(|s| s.scope.as_ref());
        assert!(matches!(
            scope_case,
            Some(api::heddle::api::v1alpha1::review_scope::Scope::WholeChange(_))
        ));
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn record_verdict_persists_signed_opinion_without_changing_state() {
        let (svc, repo, _tmp) = fresh_service();
        let state_id = capture_state(&repo);
        let op_id = objects::object::OperationId::new().to_string();
        let request = verdict_request(&state_id, ProtoVerdict::Hold, "needs another pass", &op_id);

        let first = svc
            .record_verdict(Request::new(request.clone()))
            .await
            .expect("record verdict");
        let replay = svc
            .record_verdict(Request::new(request))
            .await
            .expect("replay verdict");
        assert_eq!(first.get_ref(), replay.get_ref());
        assert_eq!(first.get_ref().state_id, Some(api_state_id(&state_id)));
        assert_eq!(repo.head().expect("read head"), Some(state_id));

        let listing = svc
            .list_signatures(Request::new(ListSignaturesRequest {
                repo_path: None,
                state_id: Some(api_state_id(&state_id)),
            }))
            .await
            .expect("list signatures");
        let signatures = &listing.get_ref().signatures;
        assert_eq!(signatures.len(), 1, "idempotent replay must append once");
        assert_eq!(signatures[0].verdict, ProtoVerdict::Hold as i32);
        assert_eq!(signatures[0].reason, "needs another pass");
        assert!(signatures[0].justification.is_empty());
        assert_eq!(
            signatures[0].kind,
            api::heddle::api::v1alpha1::ReviewKind::Read as i32
        );
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn record_verdict_rejects_reason_changed_after_signing() {
        let (svc, repo, _tmp) = fresh_service();
        let state_id = capture_state(&repo);
        let mut request = verdict_request(&state_id, ProtoVerdict::Reject, "unsafe", "");
        request.reason = "different reason".to_string();

        let error = svc
            .record_verdict(Request::new(request))
            .await
            .expect_err("changed signed reason must fail verification");
        assert_eq!(error.code(), tonic::Code::InvalidArgument);
        assert!(error.message().contains("failed verification"));
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn full_contract_routes_signal_health_and_stubs_review_progress() {
        let (svc, _repo, _tmp) = fresh_service();
        let health = svc
            .get_repo_signal_health(Request::new(GetRepoSignalHealthRequest {
                repo_path: None,
                window_states: 1,
            }))
            .await
            .expect("local signal health");
        assert_eq!(health.get_ref().window_states, 1);

        let ack_error = svc
            .record_check_ack(Request::new(RecordCheckAckRequest::default()))
            .await
            .expect_err("local check acks are unsupported");
        assert_eq!(ack_error.code(), tonic::Code::Unimplemented);
        assert!(ack_error.message().contains("not available in local mode"));

        let progress_error = svc
            .get_review_progress(Request::new(GetReviewProgressRequest::default()))
            .await
            .expect_err("local review progress is unsupported");
        assert_eq!(progress_error.code(), tonic::Code::Unimplemented);
        assert!(
            progress_error
                .message()
                .contains("not available in local mode")
        );
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn sign_state_idempotent() {
        let (svc, repo, _tmp) = fresh_service();
        let state_id = capture_state(&repo);
        let op_id = objects::object::OperationId::new().to_string();
        // The second call must replay the *same* request body — fresh
        // signatures hash differently, so we build once and clone.
        let req = sign_request(&state_id, &op_id);

        let first = svc
            .sign_state(Request::new(req.clone()))
            .await
            .expect("first sign_state");
        let second = svc
            .sign_state(Request::new(req))
            .await
            .expect("second sign_state");
        assert_eq!(
            first.get_ref().signature_id,
            second.get_ref().signature_id,
            "replayed call must return the same signature_id"
        );

        let listing = svc
            .list_signatures(Request::new(ListSignaturesRequest {
                repo_path: None,
                state_id: Some(api_state_id(&state_id)),
            }))
            .await
            .expect("list_signatures");
        assert_eq!(
            listing.get_ref().signatures.len(),
            1,
            "idempotent replay must not append a duplicate signature"
        );
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn sign_state_rejects_forged_signature() {
        let (svc, repo, _tmp) = fresh_service();
        let state_id = capture_state(&repo);
        let mut req = sign_request(&state_id, "");
        // Flip the last byte of the signature so verification fails.
        let last = req.signature.len() - 1;
        req.signature[last] ^= 0xff;

        let err = svc
            .sign_state(Request::new(req))
            .await
            .expect_err("forged signature must be rejected");
        assert_eq!(err.code(), tonic::Code::InvalidArgument, "{err:?}");
        assert!(
            err.message().contains("failed verification"),
            "unexpected error message: {}",
            err.message()
        );
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn sign_state_rejects_skewed_timestamp() {
        let (svc, repo, _tmp) = fresh_service();
        let state_id = capture_state(&repo);
        let mut req = sign_request(&state_id, "");
        // Timestamp 1 hour in the future is well outside the skew window.
        if let Some(ts) = req.signed_at.as_mut() {
            ts.seconds += 60 * 60;
        }

        let err = svc
            .sign_state(Request::new(req))
            .await
            .expect_err("skewed timestamp must be rejected");
        assert_eq!(err.code(), tonic::Code::InvalidArgument);
        assert!(err.message().contains("too far from server time"));
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn sign_state_attributes_to_caller_not_state_author() {
        // Regression for the codex-flagged bug: sign_state used to
        // attribute the signature to the state's author. Bob signing
        // Alice's state would record Alice. The signature must record
        // the *caller*. In local mode the caller is resolved via
        // `Repository::get_principal()` (env vars then config).
        let (svc, repo, _tmp) = fresh_service();
        // `fresh_service` already set the env to Alice; capture the
        // state under Alice's identity so `state.attribution.principal`
        // is Alice.
        let state_id = capture_state(&repo);

        // Now switch the local user to Bob and sign Alice's state.
        // SAFETY: tests run with a controlled environment.
        unsafe {
            std::env::set_var("HEDDLE_PRINCIPAL_NAME", "Bob Signer");
            std::env::set_var("HEDDLE_PRINCIPAL_EMAIL", "bob@example.com");
        }

        svc.sign_state(Request::new(sign_request(&state_id, "")))
            .await
            .expect("sign_state");

        let listing = svc
            .list_signatures(Request::new(ListSignaturesRequest {
                repo_path: None,
                state_id: Some(api_state_id(&state_id)),
            }))
            .await
            .expect("list_signatures");
        let sigs = &listing.get_ref().signatures;
        assert_eq!(sigs.len(), 1);
        assert_eq!(
            sigs[0].actor_name, "Bob Signer",
            "signature must attribute to the caller (Bob), not the state author (Alice)"
        );
        assert_eq!(sigs[0].actor_email, "bob@example.com");

        // Restore the env so other serial tests see the expected
        // Alice baseline.
        unsafe {
            std::env::set_var("HEDDLE_PRINCIPAL_NAME", "Alice Tester");
            std::env::set_var("HEDDLE_PRINCIPAL_EMAIL", "alice@example.com");
        }
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn sign_state_serializes_concurrent_appends() {
        let (svc, repo, _tmp) = fresh_service();
        let state_id = capture_state(&repo);

        // Two distinct ops so neither replays the other through dedup.
        let op_a = objects::object::OperationId::new().to_string();
        let op_b = objects::object::OperationId::new().to_string();
        let req_a = sign_request(&state_id, &op_a);
        let req_b = sign_request(&state_id, &op_b);

        let svc_a = svc.clone();
        let svc_b = svc.clone();
        let (a, b) = tokio::join!(
            svc_a.sign_state(Request::new(req_a)),
            svc_b.sign_state(Request::new(req_b)),
        );
        a.expect("first sign_state");
        b.expect("second sign_state");

        let listing = svc
            .list_signatures(Request::new(ListSignaturesRequest {
                repo_path: None,
                state_id: Some(api_state_id(&state_id)),
            }))
            .await
            .expect("list_signatures");
        assert_eq!(
            listing.get_ref().signatures.len(),
            2,
            "both concurrent signatures must land — neither should be lost \
             to a stale-blob clobber"
        );
    }

    /// Regression: `get_review_payload` previously returned
    /// `summary.files_changed = 0` and empty `in_budget_signals` for
    /// every state, even when the diff against the parent had real
    /// content. This test snapshots once (root state — every file is
    /// "added"), then snapshots again with a modification, and asserts
    /// both states report a non-empty diff summary plus a populated
    /// `diff_summary` signal anchored on the changed file.
    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn get_review_payload_populates_diff_summary_and_signals() {
        let (svc, repo, _tmp) = fresh_service();

        // First capture: root state, one file added. Every line of
        // `hello.txt` should count as added.
        std::fs::write(repo.root().join("hello.txt"), b"first\nsecond\nthird\n")
            .expect("write hello.txt");
        let first = repo
            .snapshot(Some("first capture".to_string()), None)
            .expect("first snapshot");

        let resp_first = svc
            .get_review_payload(Request::new(GetReviewPayloadRequest {
                repo_path: None,
                state_id: Some(api_state_id(&first.state_id)),
                include_all_signals: false,
            }))
            .await
            .expect("get_review_payload first");
        let payload_first = resp_first.into_inner();
        let summary_first = payload_first.summary.as_ref().expect("summary present");
        assert!(
            summary_first.files_changed >= 1,
            "first state should report at least one file changed (vs empty parent), got {}",
            summary_first.files_changed
        );
        assert!(
            summary_first.added_lines >= 3,
            "first state should report 3+ added lines, got {}",
            summary_first.added_lines
        );
        assert!(
            !payload_first.in_budget_signals.is_empty(),
            "in_budget_signals must include a diff_summary entry"
        );
        let first_signal = &payload_first.in_budget_signals[0];
        assert!(
            first_signal.kind.starts_with("diff_summary."),
            "expected synthetic diff_summary signal kind, got {}",
            first_signal.kind
        );
        assert_eq!(first_signal.producer_module, "review_show.diff_summary");
        assert_eq!(first_signal.visibility, "visible");

        // Second capture: modify the file. Diff vs the first state's
        // tree should report a single modified file with at least one
        // added and one removed line.
        std::fs::write(
            repo.root().join("hello.txt"),
            b"first\nsecond\nthird\nfourth\n",
        )
        .expect("modify hello.txt");
        let second = repo
            .snapshot(Some("second capture".to_string()), None)
            .expect("second snapshot");

        let resp_second = svc
            .get_review_payload(Request::new(GetReviewPayloadRequest {
                repo_path: None,
                state_id: Some(api_state_id(&second.state_id)),
                include_all_signals: false,
            }))
            .await
            .expect("get_review_payload second");
        let payload_second = resp_second.into_inner();
        let summary_second = payload_second.summary.as_ref().expect("summary present");
        assert_eq!(
            summary_second.files_changed, 1,
            "second state should report exactly one modified file"
        );
        assert!(
            summary_second.added_lines >= 1,
            "second state should report at least one added line, got {}",
            summary_second.added_lines
        );
        assert!(
            !payload_second.in_budget_signals.is_empty(),
            "in_budget_signals must include a diff_summary entry"
        );
        let signal = &payload_second.in_budget_signals[0];
        assert_eq!(
            signal
                .anchor
                .as_ref()
                .map(|a| a.file.as_str())
                .unwrap_or(""),
            "hello.txt",
            "diff_summary signal should anchor on the modified file"
        );
        assert!(
            signal.reason.contains("files changed"),
            "first signal reason should carry the aggregate summary, got {}",
            signal.reason
        );
        // The summary's signal-count fields should reflect the visible
        // budget so consumers can short-circuit on empty-vs-populated
        // without re-counting the array.
        assert_eq!(
            summary_second.in_budget_signal_count,
            payload_second.in_budget_signals.len() as u32,
            "in_budget_signal_count must match the array length"
        );
    }

    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn get_review_payload_surfaces_gitlink_target_changes() {
        let (svc, repo, _tmp) = fresh_service();

        let old_target = "0303030303030303030303030303030303030303"
            .parse()
            .expect("old git oid");
        let new_target = "0404040404040404040404040404040404040404"
            .parse()
            .expect("new git oid");
        let old_tree = objects::object::Tree::from_entries(vec![
            objects::object::TreeEntry::gitlink("vendor", old_target).expect("old gitlink"),
        ]);
        let new_tree = objects::object::Tree::from_entries(vec![
            objects::object::TreeEntry::gitlink("vendor", new_target).expect("new gitlink"),
        ]);
        let old_tree_hash = repo.store().put_tree(&old_tree).expect("put old tree");
        let new_tree_hash = repo.store().put_tree(&new_tree).expect("put new tree");
        let base = State::new_snapshot(
            old_tree_hash,
            Vec::new(),
            objects::object::Attribution::human(objects::object::Principal::new(
                "Gitlink Reviewer",
                "gitlink@example.test",
            )),
        );
        let base_id = base.state_id;
        repo.store().put_state(&base).expect("put base state");
        let changed = State::new_snapshot(
            new_tree_hash,
            vec![base_id],
            objects::object::Attribution::human(objects::object::Principal::new(
                "Gitlink Reviewer",
                "gitlink@example.test",
            )),
        );
        let changed_id = changed.state_id;
        repo.store().put_state(&changed).expect("put changed state");

        let resp = svc
            .get_review_payload(Request::new(GetReviewPayloadRequest {
                repo_path: None,
                state_id: Some(api_state_id(&changed_id)),
                include_all_signals: false,
            }))
            .await
            .expect("get_review_payload gitlink change");
        let payload = resp.into_inner();
        let summary = payload.summary.as_ref().expect("summary present");
        assert_eq!(
            summary.files_changed, 1,
            "gitlink pointer change should count as one changed path"
        );
        assert_eq!(summary.added_lines, 0);
        assert_eq!(summary.removed_lines, 0);
        assert_eq!(
            payload
                .in_budget_signals
                .first()
                .and_then(|signal| signal.anchor.as_ref())
                .map(|anchor| anchor.file.as_str()),
            Some("vendor"),
            "diff_summary signal should be anchored on the gitlink path"
        );
        let partition = payload.partition.expect("partition present");
        let surfaced = partition
            .structural
            .iter()
            .chain(partition.consequence.iter())
            .chain(partition.tests_and_docs.iter())
            .any(|symbol| symbol.file == "vendor" && symbol.symbol == "vendor");
        assert!(surfaced, "gitlink change should be path-visible in review");
    }

    /// Regression for codex feedback on PRs #52 (tree fallback) and
    /// #56 (blob fallback): `compute_state_diff_summary` must tolerate
    /// missing trees and blobs by returning an empty/partial summary
    /// rather than blocking the entire review payload. Construct a
    /// state whose `tree` points to a content hash that was never
    /// stored — `get_tree` returns `Ok(None)`, and the function must
    /// fall back to an empty changeset instead of erroring out of
    /// `diff_trees`. (Pre-fix, the Modified branch already tolerated
    /// missing blobs via `.ok().flatten()`, but the Added/Deleted
    /// branches and the top-level diff_trees call did not — surfaces
    /// of pruned object stores all errored out inconsistently.)
    #[tokio::test]
    #[serial_test::serial(process_global)]
    async fn get_review_payload_tolerates_missing_tree() {
        let (svc, repo, _tmp) = fresh_service();
        let state_id = capture_state(&repo);

        // Load the state, swap its tree pointer to a synthetic hash
        // that nothing references, and re-persist. The state object
        // itself survives, but `repo.store().get_tree(state.tree)`
        // will return `Ok(None)` — the missing-tree case.
        let state = repo
            .store()
            .get_state(&state_id)
            .expect("get state")
            .expect("state present");
        let bogus_tree = objects::object::ContentHash::compute(b"definitely-not-in-store-bytes");
        let mut mutated = state.clone();
        mutated.tree = bogus_tree;
        let missing_tree_state_id = mutated.id();
        repo.store().put_state(&mutated).expect("put mutated state");

        // The review payload must still come back — empty summary,
        // but no Status::Internal error from the gRPC layer.
        let resp = svc
            .get_review_payload(Request::new(GetReviewPayloadRequest {
                repo_path: None,
                state_id: Some(api_state_id(&missing_tree_state_id)),
                include_all_signals: false,
            }))
            .await
            .expect("missing tree must not block review payload");
        let payload = resp.into_inner();
        let summary = payload.summary.as_ref().expect("summary present");
        assert_eq!(
            summary.files_changed, 0,
            "missing tree must produce a zero-change summary, got {} files",
            summary.files_changed
        );
        // Synthetic diff_summary signal should still be present (with
        // the `empty` kind) so consumers always see at least one
        // signal — keeps the wire shape stable.
        assert!(
            !payload.in_budget_signals.is_empty(),
            "in_budget_signals should always contain at least the synthetic diff_summary entry"
        );
        let kind = &payload.in_budget_signals[0].kind;
        assert!(
            kind.starts_with("diff_summary."),
            "expected synthetic diff_summary signal, got {kind}"
        );
    }

    /// `line_count` should match git-style line counts — trailing
    /// newline never produces a phantom empty line, but an unterminated
    /// final line still counts.
    #[test]
    fn line_count_matches_git_semantics() {
        assert_eq!(line_count(b""), 0);
        assert_eq!(line_count(b"\n"), 1);
        assert_eq!(line_count(b"hello"), 1);
        assert_eq!(line_count(b"hello\n"), 1);
        assert_eq!(line_count(b"hello\nworld"), 2);
        assert_eq!(line_count(b"hello\nworld\n"), 2);
        assert_eq!(line_count(b"a\nb\nc\n"), 3);
        // Non-utf8 bytes count as zero (treated as binary).
        assert_eq!(line_count(&[0xff, 0xfe, 0xfd]), 0);
    }
}