car-a2a 0.9.0

//! Translation layer between A2A messages/tasks and CAR IR.
//!
//! Two directions:
//!
//! 1. **A2A → CAR.** Inbound `message/send` carries a `Message` whose
//!    `parts` may include a structured `data` part naming a tool and
//!    parameters. We compile that into an `ActionProposal` of one
//!    `ToolCall` action. Free-text parts pass through as proposal
//!    `context` so a downstream planner can pick them up.
//!
//! 2. **CAR → A2A.** A `ProposalResult` becomes a list of `Artifact`s
//!    — one per `ActionResult`. Each artifact carries the action's
//!    output as a `data` part plus a short text part summarising the
//!    outcome.
//!
//! Lifecycle mapping:
//!
//! | A2A `TaskState` | CAR signal |
//! |-----------------|-----------|
//! | `SUBMITTED` | Task created, proposal queued |
//! | `WORKING`   | `Runtime::execute` in flight |
//! | `COMPLETED` | `ProposalResult::all_succeeded() == true` |
//! | `FAILED`    | At least one `ActionStatus::Failed`/`Rejected` |
//! | `CANCELED`  | Caller invoked `tasks/cancel` |

use car_ir::{Action, ActionProposal, ActionResult, ActionStatus, ActionType, ProposalResult};
use chrono::Utc;
use std::collections::HashMap;

use crate::auth::Identity;
use crate::types::{Artifact, DataPart, Message, Part, Task, TaskState, TaskStatus, TextPart};

#[derive(Debug, thiserror::Error)]
pub enum ProposalBuildError {
    #[error("message contained no actionable content (no tool data part, no text)")]
    Empty,
    #[error("`data.parameters` must be a JSON object")]
    BadParameters,
}

/// Compile an A2A `Message` into a CAR `ActionProposal`.
///
/// Recognised structured form (the easy path — peer agent already
/// knows the tool name):
///
/// ```json
/// {
///   "kind": "data",
///   "data": {
///     "tool": "fs.read",
///     "parameters": { "path": "/tmp/x" }
///   }
/// }
/// ```
///
/// Falls back to `Empty` if neither a `data.tool` nor any text part is
/// present. When only text is present, the caller can still build a
/// proposal externally (e.g. via `car-active-planner`); this function
/// returns an empty proposal whose `context` carries the text so the
/// planner has something to chew on.
pub fn message_to_proposal(message: &Message) -> Result<ActionProposal, ProposalBuildError> {
    message_to_proposal_with_identity(message, None)
}

/// Same as [`message_to_proposal`] but also merges the caller's
/// verified identity (when present) onto
/// `proposal.context["a2a_caller_verified"]`. Used by the HTTP
/// dispatcher after [`crate::auth::AuthValidator::validate`]
/// returns an [`Identity`]. Tools and policies can cross-check
/// the *verified* subject against the cooperative `a2a_caller`
/// claims (Parslee-ai/car#187 phase 2).
pub fn message_to_proposal_with_identity(
    message: &Message,
    identity: Option<&Identity>,
) -> Result<ActionProposal, ProposalBuildError> {
    let mut actions = Vec::new();
    let mut context: HashMap<String, serde_json::Value> = HashMap::new();
    let mut text_blobs: Vec<String> = Vec::new();

    for part in &message.parts {
        match part {
            Part::Text(t) => text_blobs.push(t.text.clone()),
            Part::Data(d) => {
                if let Some(obj) = d.data.as_object() {
                    if let Some(tool) = obj.get("tool").and_then(|v| v.as_str()) {
                        let params_value = obj
                            .get("parameters")
                            .cloned()
                            .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new()));
                        let params_obj = match params_value {
                            serde_json::Value::Object(map) => map,
                            _ => return Err(ProposalBuildError::BadParameters),
                        };
                        let parameters: HashMap<String, serde_json::Value> =
                            params_obj.into_iter().collect();
                        actions.push(Action {
                            id: short_id(),
                            action_type: ActionType::ToolCall,
                            tool: Some(tool.to_string()),
                            parameters,
                            preconditions: Vec::new(),
                            expected_effects: HashMap::new(),
                            state_dependencies: Vec::new(),
                            idempotent: false,
                            max_retries: 3,
                            failure_behavior: car_ir::FailureBehavior::Abort,
                            timeout_ms: None,
                            metadata: HashMap::new(),
                        });
                    } else {
                        // Stash anonymous data under a stable key so the
                        // planner can see it.
                        context.insert("a2a_data".into(), d.data.clone());
                    }
                }
            }
            Part::File(f) => {
                context.insert(
                    "a2a_file".into(),
                    serde_json::to_value(&f.file).unwrap_or(serde_json::Value::Null),
                );
            }
        }
    }

    if !text_blobs.is_empty() {
        context.insert(
            "a2a_text".into(),
            serde_json::Value::String(text_blobs.join("\n")),
        );
    }

    if actions.is_empty() && context.is_empty() {
        return Err(ProposalBuildError::Empty);
    }

    let mut ctx_with_origin = context;
    ctx_with_origin.insert(
        "a2a_message_id".into(),
        serde_json::Value::String(message.message_id.clone()),
    );

    // Surface caller-supplied identity fields from `Message.metadata`.
    // A2A v1.0 lets peers carry arbitrary key/value metadata; for
    // multi-tenant deployments the convention is `caller_id`,
    // `org_id`, `project_id`, and `tenant_id`. We don't enforce
    // these here — that's the embedder's job for now (the listener
    // hasn't verified the peer's claim to any of these values
    // beyond whatever auth the transport did) — but surfacing them
    // on `a2a_caller` lets tools and policies read them downstream
    // without each one re-parsing the message. This is phase 1 of
    // Parslee-ai/car#187; phase 2 will plumb the verified bearer
    // subject in here so policies can cross-check.
    let identity_keys = ["caller_id", "org_id", "project_id", "tenant_id"];
    let mut caller_obj = serde_json::Map::new();
    for key in identity_keys {
        if let Some(value) = message.metadata.get(key).cloned() {
            caller_obj.insert(key.into(), value);
        }
    }
    if !caller_obj.is_empty() {
        ctx_with_origin.insert(
            "a2a_caller".into(),
            serde_json::Value::Object(caller_obj),
        );
    }

    // Server-verified identity (phase 2). Distinct from `a2a_caller`
    // which carries the peer's *claims*. When the AuthValidator
    // returned an Identity, we surface its subject + claims here so
    // policies can enforce "claims agree with verified subject."
    if let Some(id) = identity {
        let mut verified = serde_json::Map::new();
        verified.insert(
            "subject".into(),
            serde_json::Value::String(id.subject.clone()),
        );
        if !id.claims.is_empty() {
            verified.insert(
                "claims".into(),
                serde_json::Value::Object(id.claims.iter().map(|(k, v)| (k.clone(), v.clone())).collect()),
            );
        }
        ctx_with_origin.insert(
            "a2a_caller_verified".into(),
            serde_json::Value::Object(verified),
        );
    }

    Ok(ActionProposal {
        id: short_id(),
        source: "a2a".into(),
        actions,
        timestamp: Utc::now(),
        context: ctx_with_origin,
    })
}

/// Build a [`car_engine::RuntimeScope`] from the verified `Identity`
/// (phase 2) and the inbound `Message.metadata` (phase 1) —
/// Parslee-ai/car#187 phase 3 foundation.
///
/// Precedence: verified-subject claims override cooperative-peer
/// hints. Specifically:
/// - `caller_id` ← `identity.subject` if present, else `metadata.caller_id`
/// - `tenant_id` ← `identity.claims["tenant_id"]` if present (and a
///   string), else `metadata.tenant_id`
/// - `claims` ← forwarded verbatim from `identity.claims` (subject is
///   already on `caller_id`; this carries the surrounding bag for
///   audit). Cooperative-peer hints are NOT mixed in — that would
///   blur the verified / unverified boundary.
///
/// Returns the default (all-`None`) scope when no identity surfaces
/// land on either source. The dispatcher's default-deny check
/// inspects `RuntimeScope::is_unscoped` to decide whether to reject.
pub fn scope_from_message_and_identity(
    message: &Message,
    identity: Option<&Identity>,
) -> car_engine::RuntimeScope {
    let mut caller_id: Option<String> = None;
    let mut tenant_id: Option<String> = None;
    let mut claims = std::collections::BTreeMap::new();

    if let Some(id) = identity {
        if !id.subject.is_empty() {
            caller_id = Some(id.subject.clone());
        }
        if let Some(tid) = id
            .claims
            .get("tenant_id")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
        {
            tenant_id = Some(tid.to_string());
        }
        for (k, v) in &id.claims {
            claims.insert(k.clone(), v.clone());
        }
    }

    // Fall back to cooperative-peer hints from Message.metadata when
    // the verified surface didn't supply that field. Verified always
    // wins — peers can claim whatever tenant they want; we only use
    // their hint when no verified value exists.
    if caller_id.is_none() {
        if let Some(v) = message
            .metadata
            .get("caller_id")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
        {
            caller_id = Some(v.to_string());
        }
    }
    if tenant_id.is_none() {
        if let Some(v) = message
            .metadata
            .get("tenant_id")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
        {
            tenant_id = Some(v.to_string());
        }
    }

    car_engine::RuntimeScope {
        caller_id,
        tenant_id,
        claims,
    }
}

/// Convert a CAR `ProposalResult` into A2A artifacts — one per action
/// outcome.
pub fn action_results_to_artifacts(result: &ProposalResult) -> Vec<Artifact> {
    result
        .results
        .iter()
        .map(|r| action_result_to_artifact(r))
        .collect()
}

/// Extract A2UI envelopes carried inside an A2A artifact.
///
/// CAR treats A2UI as data, so peer agents can return either a direct
/// envelope in a `data` part or wrap it as `{ "a2ui": envelope }`.
/// The renderer side can use this helper to route UI payloads into a
/// surface store while still preserving the original artifact.
pub fn a2ui_envelopes_from_artifact(
    artifact: &Artifact,
) -> Result<Vec<car_a2ui::A2uiEnvelope>, car_a2ui::A2uiError> {
    let value =
        serde_json::to_value(artifact).map_err(|e| car_a2ui::A2uiError::Parse(e.to_string()))?;
    car_a2ui::envelopes_from_value(&value)
}

fn action_result_to_artifact(result: &ActionResult) -> Artifact {
    let summary = match result.status {
        ActionStatus::Succeeded => "succeeded".to_string(),
        ActionStatus::Failed => format!(
            "failed: {}",
            result.error.as_deref().unwrap_or("(no detail)")
        ),
        ActionStatus::Rejected => format!(
            "rejected: {}",
            result.error.as_deref().unwrap_or("(no detail)")
        ),
        ActionStatus::Skipped => "skipped".to_string(),
        ActionStatus::Executing => "executing".to_string(),
        ActionStatus::Validated => "validated".to_string(),
        ActionStatus::Proposed => "proposed".to_string(),
    };

    let mut parts: Vec<Part> = vec![Part::Text(TextPart {
        text: format!("action {}: {}", result.action_id, summary),
        metadata: HashMap::new(),
    })];

    if let Some(out) = &result.output {
        if !car_a2ui::envelopes_from_value(out)
            .unwrap_or_default()
            .is_empty()
        {
            parts.push(Part::Text(TextPart {
                text: "A2UI surface update".into(),
                metadata: HashMap::new(),
            }));
        }
        parts.push(Part::Data(DataPart {
            data: out.clone(),
            metadata: HashMap::new(),
        }));
    }

    let mut metadata = HashMap::new();
    if result.output.as_ref().is_some_and(|out| {
        !car_a2ui::envelopes_from_value(out)
            .unwrap_or_default()
            .is_empty()
    }) {
        metadata.insert("a2ui".into(), serde_json::Value::Bool(true));
        metadata.insert(
            "mimeType".into(),
            serde_json::Value::String(car_a2ui::A2UI_MIME_TYPE.into()),
        );
    }

    Artifact {
        artifact_id: format!("art-{}", result.action_id),
        name: format!("action.{}", result.action_id),
        description: result.error.clone(),
        parts,
        metadata,
    }
}

/// Map a `ProposalResult` to a terminal A2A `TaskState`.
///
/// Rules (priority order):
///
/// - Empty `results` ⇒ `Completed`. A text-only message yields a
///   proposal with no actions, the runtime executes a no-op, and the
///   honest answer is "acknowledged" — not "failed".
/// - Any action `Skipped` with the engine's `CANCELED_PREFIX` error
///   ⇒ `Canceled`. This is what `Runtime::execute_with_cancel` emits
///   when its token trips at a level boundary; without recognising
///   it here the bridge would misreport cancelled tasks as
///   `Completed`. The store's terminal-state guard masks this for
///   peer-initiated `tasks/cancel` (which flips state to `Canceled`
///   before the executor's late `update_state` arrives), but
///   engine-initiated cancels (cost-budget, timeouts) skip that
///   path — they need this branch to land correctly.
/// - Any `Rejected` action ⇒ `Rejected`.
/// - Any `Failed` action ⇒ `Failed`.
/// - Otherwise ⇒ `Completed`.
pub fn a2a_state_for(result: &ProposalResult) -> TaskState {
    if result.results.is_empty() {
        return TaskState::Completed;
    }
    let mut any_canceled = false;
    let mut any_rejected = false;
    let mut any_failed = false;
    for r in &result.results {
        match r.status {
            ActionStatus::Rejected => any_rejected = true,
            ActionStatus::Failed => any_failed = true,
            ActionStatus::Skipped
                if r.error
                    .as_deref()
                    .is_some_and(|e| e.starts_with(car_engine::CANCELED_PREFIX)) =>
            {
                any_canceled = true;
            }
            _ => {}
        }
    }
    if any_canceled {
        TaskState::Canceled
    } else if any_rejected {
        TaskState::Rejected
    } else if any_failed {
        TaskState::Failed
    } else {
        TaskState::Completed
    }
}

/// Bundle a status update into a complete `Task` payload. Used by the
/// store when emitting state transitions.
pub fn task_with_status(
    task_id: String,
    context_id: String,
    state: TaskState,
    history: Vec<Message>,
    artifacts: Vec<Artifact>,
) -> Task {
    Task {
        id: task_id,
        context_id,
        status: TaskStatus {
            state,
            message: None,
            timestamp: Utc::now(),
        },
        history,
        artifacts,
        metadata: HashMap::new(),
    }
}

/// Short UUID used to mint task ids, message ids, etc. Exposed so the
/// dispatcher and other modules don't each maintain their own copy.
pub fn short_id() -> String {
    uuid::Uuid::new_v4().simple().to_string()[..12].to_string()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{DataPart, Message, MessageRole, Part, TextPart};
    use serde_json::{json, Value};

    #[test]
    fn structured_data_part_compiles_to_tool_call() {
        let msg = Message {
            message_id: "m1".into(),
            role: MessageRole::User,
            parts: vec![Part::Data(DataPart {
                data: json!({
                    "tool": "fs.read",
                    "parameters": { "path": "/tmp/x" }
                }),
                metadata: HashMap::new(),
            })],
            task_id: None,
            context_id: None,
            metadata: HashMap::new(),
        };
        let proposal = message_to_proposal(&msg).expect("compile");
        assert_eq!(proposal.actions.len(), 1);
        let action = &proposal.actions[0];
        assert_eq!(action.action_type, ActionType::ToolCall);
        assert_eq!(action.tool.as_deref(), Some("fs.read"));
        assert_eq!(action.parameters.get("path"), Some(&json!("/tmp/x")));
        assert_eq!(proposal.source, "a2a");
        assert_eq!(proposal.context.get("a2a_message_id"), Some(&json!("m1")));
    }

    #[test]
    fn text_only_message_yields_context_proposal() {
        let msg = Message {
            message_id: "m2".into(),
            role: MessageRole::User,
            parts: vec![Part::Text(TextPart {
                text: "summarise this".into(),
                metadata: HashMap::new(),
            })],
            task_id: None,
            context_id: None,
            metadata: HashMap::new(),
        };
        let proposal = message_to_proposal(&msg).expect("compile");
        assert!(proposal.actions.is_empty());
        assert_eq!(
            proposal.context.get("a2a_text"),
            Some(&json!("summarise this"))
        );
    }

    #[test]
    fn caller_identity_fields_surface_on_proposal_context() {
        // Phase 1 of Parslee-ai/car#187: cooperative-caller identity
        // fields (caller_id / org_id / project_id / tenant_id) carried
        // on Message.metadata are copied to proposal.context["a2a_caller"]
        // so tools and policies can read them without re-parsing the
        // message. No verification yet — see phase 2.
        let mut metadata = HashMap::new();
        metadata.insert("caller_id".to_string(), json!("neo@example.com"));
        metadata.insert("org_id".to_string(), json!("acme-corp"));
        metadata.insert("project_id".to_string(), json!("flagship-repo"));
        metadata.insert("unrelated_key".to_string(), json!("ignored"));

        let msg = Message {
            message_id: "m-caller".into(),
            role: MessageRole::User,
            parts: vec![Part::Data(DataPart {
                data: json!({ "tool": "noop", "parameters": {} }),
                metadata: HashMap::new(),
            })],
            task_id: None,
            context_id: None,
            metadata,
        };
        let proposal = message_to_proposal(&msg).expect("compile");
        let caller = proposal
            .context
            .get("a2a_caller")
            .expect("a2a_caller in context")
            .as_object()
            .expect("a2a_caller is an object");
        assert_eq!(caller.get("caller_id"), Some(&json!("neo@example.com")));
        assert_eq!(caller.get("org_id"), Some(&json!("acme-corp")));
        assert_eq!(caller.get("project_id"), Some(&json!("flagship-repo")));
        assert!(
            !caller.contains_key("unrelated_key"),
            "only the allow-list of identity keys should appear under a2a_caller"
        );
    }

    #[test]
    fn verified_identity_surfaces_as_a2a_caller_verified() {
        // Phase 2 (Parslee-ai/car#187): when the HTTP layer
        // resolves a verified Identity from AuthValidator, it gets
        // copied onto proposal.context["a2a_caller_verified"] —
        // distinct from the cooperative `a2a_caller` claims (which
        // are still unverified at this phase). Policies can
        // cross-check the two surfaces.
        let mut claims = HashMap::new();
        claims.insert("scope".into(), json!("repos:read"));
        let identity = crate::auth::Identity {
            subject: "alice@example.com".into(),
            claims,
        };
        // Peer claims they're Bob; the server has verified Alice.
        // The proposal context shows both, distinguishable — a
        // policy can reject a mismatch.
        let mut metadata = HashMap::new();
        metadata.insert("caller_id".into(), json!("bob@example.com"));
        let msg = Message {
            message_id: "m-verified".into(),
            role: MessageRole::User,
            parts: vec![Part::Data(DataPart {
                data: json!({ "tool": "noop", "parameters": {} }),
                metadata: HashMap::new(),
            })],
            task_id: None,
            context_id: None,
            metadata,
        };
        let proposal =
            message_to_proposal_with_identity(&msg, Some(&identity)).expect("compile");

        let verified = proposal
            .context
            .get("a2a_caller_verified")
            .expect("a2a_caller_verified in context")
            .as_object()
            .expect("a2a_caller_verified is an object");
        assert_eq!(verified.get("subject"), Some(&json!("alice@example.com")));
        let verified_claims = verified
            .get("claims")
            .expect("claims sub-object")
            .as_object()
            .expect("claims is object");
        assert_eq!(verified_claims.get("scope"), Some(&json!("repos:read")));

        // Cooperative claim is still surfaced as-is — separate
        // field, distinguishable from the verified one.
        let claimed = proposal
            .context
            .get("a2a_caller")
            .expect("a2a_caller in context")
            .as_object()
            .expect("a2a_caller is an object");
        assert_eq!(claimed.get("caller_id"), Some(&json!("bob@example.com")));
    }

    #[test]
    fn missing_identity_omits_a2a_caller_verified_field() {
        // When no Identity is supplied (NoAuth, or auth that
        // returned Ok(None)), the verified surface is absent —
        // distinguishable from "supplied but with empty subject".
        let msg = Message {
            message_id: "m-anon".into(),
            role: MessageRole::User,
            parts: vec![Part::Data(DataPart {
                data: json!({ "tool": "noop", "parameters": {} }),
                metadata: HashMap::new(),
            })],
            task_id: None,
            context_id: None,
            metadata: HashMap::new(),
        };
        let proposal = message_to_proposal_with_identity(&msg, None).expect("compile");
        assert!(
            !proposal.context.contains_key("a2a_caller_verified"),
            "a2a_caller_verified absent when no identity supplied"
        );
    }

    #[test]
    fn no_caller_metadata_means_no_a2a_caller_field() {
        // When the peer didn't supply identity fields, the
        // `a2a_caller` key is absent — distinguishable from
        // "supplied but empty" so tools can detect missing identity
        // and apply default-deny if they want.
        let msg = Message {
            message_id: "m-no-caller".into(),
            role: MessageRole::User,
            parts: vec![Part::Data(DataPart {
                data: json!({ "tool": "noop", "parameters": {} }),
                metadata: HashMap::new(),
            })],
            task_id: None,
            context_id: None,
            metadata: HashMap::new(),
        };
        let proposal = message_to_proposal(&msg).expect("compile");
        assert!(
            !proposal.context.contains_key("a2a_caller"),
            "a2a_caller absent when no identity metadata supplied"
        );
    }

    #[test]
    fn empty_message_errors() {
        let msg = Message {
            message_id: "m3".into(),
            role: MessageRole::User,
            parts: vec![],
            task_id: None,
            context_id: None,
            metadata: HashMap::new(),
        };
        assert!(matches!(
            message_to_proposal(&msg),
            Err(ProposalBuildError::Empty)
        ));
    }

    #[test]
    fn proposal_result_round_trips_to_artifacts() {
        let pr = ProposalResult {
            proposal_id: "p1".into(),
            results: vec![
                ActionResult {
                    action_id: "a1".into(),
                    status: ActionStatus::Succeeded,
                    output: Some(json!({"result": 42})),
                    error: None,
                    state_changes: HashMap::new(),
                    duration_ms: Some(12.0),
                    timestamp: Utc::now(),
                },
                ActionResult {
                    action_id: "a2".into(),
                    status: ActionStatus::Failed,
                    output: None,
                    error: Some("boom".into()),
                    state_changes: HashMap::new(),
                    duration_ms: Some(3.0),
                    timestamp: Utc::now(),
                },
            ],
            cost: car_ir::CostSummary::default(),
        };
        let arts = action_results_to_artifacts(&pr);
        assert_eq!(arts.len(), 2);
        assert_eq!(arts[0].name, "action.a1");
        assert_eq!(arts[1].description.as_deref(), Some("boom"));
        assert_eq!(a2a_state_for(&pr), TaskState::Failed);
    }

    #[test]
    fn empty_proposal_result_is_completed() {
        let pr = ProposalResult {
            proposal_id: "p-empty".into(),
            results: vec![],
            cost: car_ir::CostSummary::default(),
        };
        assert_eq!(a2a_state_for(&pr), TaskState::Completed);
    }

    #[test]
    fn skipped_with_canceled_prefix_maps_to_canceled_state() {
        // Engine emits Skipped("canceled: ...") when its cancellation
        // token trips at a level boundary. The bridge must recognise
        // this prefix and return TaskState::Canceled — otherwise
        // engine-initiated cancels (cost budget, future timeouts)
        // silently report as Completed.
        let pr = ProposalResult {
            proposal_id: "p-cancel".into(),
            results: vec![
                ActionResult {
                    action_id: "a1".into(),
                    status: ActionStatus::Succeeded,
                    output: None,
                    error: None,
                    state_changes: HashMap::new(),
                    duration_ms: None,
                    timestamp: Utc::now(),
                },
                ActionResult {
                    action_id: "a2".into(),
                    status: ActionStatus::Skipped,
                    output: None,
                    error: Some(format!(
                        "{}{}",
                        car_engine::CANCELED_PREFIX,
                        "cancellation requested by caller"
                    )),
                    state_changes: HashMap::new(),
                    duration_ms: None,
                    timestamp: Utc::now(),
                },
            ],
            cost: car_ir::CostSummary::default(),
        };
        assert_eq!(a2a_state_for(&pr), TaskState::Canceled);
    }

    #[test]
    fn rejected_action_yields_rejected_state() {
        let pr = ProposalResult {
            proposal_id: "p-rej".into(),
            results: vec![ActionResult {
                action_id: "a1".into(),
                status: ActionStatus::Rejected,
                output: None,
                error: Some("policy denied".into()),
                state_changes: HashMap::new(),
                duration_ms: None,
                timestamp: Utc::now(),
            }],
            cost: car_ir::CostSummary::default(),
        };
        assert_eq!(a2a_state_for(&pr), TaskState::Rejected);
    }

    // Tests for scope_from_message_and_identity (Parslee-ai/car#187
    // phase 3). The helper resolves caller_id / tenant_id with a
    // precedence rule: verified Identity wins; cooperative-peer
    // hints from Message.metadata fall through only when the
    // verified surface didn't supply that field.

    fn msg_with_metadata(metadata: HashMap<String, Value>) -> Message {
        Message {
            message_id: "m-scope".into(),
            role: MessageRole::User,
            parts: vec![Part::Data(DataPart {
                data: json!({ "tool": "noop", "parameters": {} }),
                metadata: HashMap::new(),
            })],
            task_id: None,
            context_id: None,
            metadata,
        }
    }

    fn identity_with(subject: &str, claims: &[(&str, Value)]) -> Identity {
        let mut map = HashMap::new();
        for (k, v) in claims {
            map.insert((*k).to_string(), v.clone());
        }
        Identity {
            subject: subject.to_string(),
            claims: map,
        }
    }

    #[test]
    fn scope_unscoped_when_no_identity_and_no_caller_metadata() {
        let msg = msg_with_metadata(HashMap::new());
        let scope = scope_from_message_and_identity(&msg, None);
        assert!(scope.is_unscoped(), "no identity surface → unscoped");
    }

    #[test]
    fn scope_falls_back_to_cooperative_peer_hints_when_no_identity() {
        let mut md = HashMap::new();
        md.insert("caller_id".into(), Value::String("alice".into()));
        md.insert("tenant_id".into(), Value::String("acme".into()));
        let msg = msg_with_metadata(md);

        let scope = scope_from_message_and_identity(&msg, None);
        assert_eq!(scope.caller_id.as_deref(), Some("alice"));
        assert_eq!(scope.tenant_id.as_deref(), Some("acme"));
        assert!(scope.claims.is_empty(), "no verified claims forwarded");
        assert!(!scope.is_unscoped());
    }

    #[test]
    fn scope_verified_identity_wins_over_peer_caller_id() {
        // Peer claims to be "alice" but the AuthValidator says the
        // verified subject is "bob". Verified wins — the peer can
        // claim whatever they want, we only trust the validator.
        let mut md = HashMap::new();
        md.insert("caller_id".into(), Value::String("alice".into()));
        let msg = msg_with_metadata(md);

        let identity = identity_with("bob", &[]);
        let scope = scope_from_message_and_identity(&msg, Some(&identity));
        assert_eq!(scope.caller_id.as_deref(), Some("bob"));
    }

    #[test]
    fn scope_verified_tenant_claim_wins_over_peer_tenant_id() {
        // Same precedence rule on tenant_id.
        let mut md = HashMap::new();
        md.insert("tenant_id".into(), Value::String("acme".into()));
        let msg = msg_with_metadata(md);

        let identity = identity_with(
            "bob",
            &[("tenant_id", Value::String("verified-corp".into()))],
        );
        let scope = scope_from_message_and_identity(&msg, Some(&identity));
        assert_eq!(scope.tenant_id.as_deref(), Some("verified-corp"));
    }

    #[test]
    fn scope_peer_tenant_used_when_verified_claims_missing_tenant() {
        // Validator returned an Identity with no tenant_id claim;
        // the cooperative peer hint fills in for that one field.
        let mut md = HashMap::new();
        md.insert("tenant_id".into(), Value::String("acme".into()));
        let msg = msg_with_metadata(md);

        let identity = identity_with("bob", &[]);
        let scope = scope_from_message_and_identity(&msg, Some(&identity));
        assert_eq!(scope.caller_id.as_deref(), Some("bob"));
        assert_eq!(scope.tenant_id.as_deref(), Some("acme"));
    }

    #[test]
    fn scope_forwards_verified_claims_verbatim() {
        let identity = identity_with(
            "bob",
            &[
                ("iss", Value::String("https://idp.example".into())),
                ("aud", Value::String("car-a2a".into())),
                ("roles", Value::Array(vec![Value::String("admin".into())])),
            ],
        );
        let scope = scope_from_message_and_identity(&msg_with_metadata(HashMap::new()), Some(&identity));
        assert_eq!(scope.claims.len(), 3);
        assert_eq!(
            scope.claims.get("iss").and_then(|v| v.as_str()),
            Some("https://idp.example")
        );
    }
}