macp-runtime 0.8.5

use chrono::Utc;
use macp_runtime::log_store::LogStore;
use macp_runtime::pb::{CommitmentPayload, Envelope, SessionResumePayload, SessionStartPayload};
use macp_runtime::registry::SessionRegistry;
use macp_runtime::runtime::Runtime;
use macp_runtime::session::{Session, SessionState};
use macp_runtime::storage::MemoryBackend;
use prost::Message;
use std::sync::Arc;

fn new_sid() -> String {
    uuid::Uuid::new_v4().as_hyphenated().to_string()
}

fn make_runtime() -> Runtime {
    let storage: Arc<dyn macp_runtime::storage::StorageBackend> = Arc::new(MemoryBackend);
    let registry = Arc::new(SessionRegistry::new());
    let log_store = Arc::new(LogStore::new());
    Runtime::new(storage, registry, log_store)
}

fn session_start(participants: Vec<String>) -> Vec<u8> {
    SessionStartPayload {
        intent: "integration-test".into(),
        participants,
        mode_version: "1.0.0".into(),
        configuration_version: "cfg-1".into(),
        policy_version: String::new(),
        ttl_ms: 60_000,
        context_id: String::new(),
        extensions: std::collections::HashMap::new(),
        roots: vec![],
        max_suspend_ms: 0,
    }
    .encode_to_vec()
}

fn envelope(
    mode: &str,
    message_type: &str,
    message_id: &str,
    session_id: &str,
    sender: &str,
    payload: Vec<u8>,
) -> Envelope {
    Envelope {
        macp_version: "1.0".into(),
        mode: mode.into(),
        message_type: message_type.into(),
        message_id: message_id.into(),
        session_id: session_id.into(),
        sender: sender.into(),
        timestamp_unix_ms: Utc::now().timestamp_millis(),
        payload,
    }
}

fn commitment(action: &str) -> Vec<u8> {
    CommitmentPayload {
        commitment_id: "c1".into(),
        action: action.into(),
        authority_scope: "test".into(),
        reason: "done".into(),
        mode_version: "1.0.0".into(),
        policy_version: "policy.default".into(),
        configuration_version: "cfg-1".into(),
        outcome_positive: true,
        supersedes: None,
    }
    .encode_to_vec()
}

#[tokio::test]
async fn decision_full_lifecycle_through_runtime() {
    use macp_runtime::decision_pb::{ProposalPayload, VotePayload};

    let rt = make_runtime();
    let sid = new_sid();
    let mode = "macp.mode.decision.v1";

    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://orchestrator",
            session_start(vec![
                "agent://orchestrator".into(),
                "agent://a".into(),
                "agent://b".into(),
            ]),
        ),
        None,
    )
    .await
    .unwrap();

    let proposal = ProposalPayload {
        proposal_id: "p1".into(),
        option: "deploy".into(),
        rationale: "ready".into(),
        supporting_data: vec![],
    }
    .encode_to_vec();
    rt.process(
        &envelope(
            mode,
            "Proposal",
            "m2",
            &sid,
            "agent://orchestrator",
            proposal,
        ),
        None,
    )
    .await
    .unwrap();

    let vote = VotePayload {
        proposal_id: "p1".into(),
        vote: "approve".into(),
        reason: "good".into(),
    }
    .encode_to_vec();
    rt.process(&envelope(mode, "Vote", "m3", &sid, "agent://a", vote), None)
        .await
        .unwrap();

    let result = rt
        .process(
            &envelope(
                mode,
                "Commitment",
                "m4",
                &sid,
                "agent://orchestrator",
                commitment("decision.selected"),
            ),
            None,
        )
        .await
        .unwrap();
    assert_eq!(result.session_state, SessionState::Resolved);
}

#[tokio::test]
async fn proposal_full_lifecycle_through_runtime() {
    use macp_runtime::proposal_pb::{AcceptPayload, ProposalPayload};

    let rt = make_runtime();
    let sid = new_sid();
    let mode = "macp.mode.proposal.v1";

    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://buyer",
            session_start(vec!["agent://buyer".into(), "agent://seller".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    let proposal = ProposalPayload {
        proposal_id: "p1".into(),
        title: "offer".into(),
        summary: "terms".into(),
        details: vec![],
        tags: vec![],
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "Proposal", "m2", &sid, "agent://seller", proposal),
        None,
    )
    .await
    .unwrap();

    let accept = AcceptPayload {
        proposal_id: "p1".into(),
        reason: String::new(),
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "Accept", "m3", &sid, "agent://buyer", accept.clone()),
        None,
    )
    .await
    .unwrap();
    rt.process(
        &envelope(mode, "Accept", "m4", &sid, "agent://seller", accept),
        None,
    )
    .await
    .unwrap();

    let result = rt
        .process(
            &envelope(
                mode,
                "Commitment",
                "m5",
                &sid,
                "agent://buyer",
                commitment("proposal.accepted"),
            ),
            None,
        )
        .await
        .unwrap();
    assert_eq!(result.session_state, SessionState::Resolved);
}

#[tokio::test]
async fn task_full_lifecycle_through_runtime() {
    use macp_runtime::task_pb::{TaskAcceptPayload, TaskCompletePayload, TaskRequestPayload};

    let rt = make_runtime();
    let sid = new_sid();
    let mode = "macp.mode.task.v1";

    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://planner",
            session_start(vec!["agent://planner".into(), "agent://worker".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    let request = TaskRequestPayload {
        task_id: "t1".into(),
        title: "Build widget".into(),
        instructions: "Do it".into(),
        requested_assignee: "agent://worker".into(),
        input: vec![],
        deadline_unix_ms: 0,
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "TaskRequest", "m2", &sid, "agent://planner", request),
        None,
    )
    .await
    .unwrap();

    let accept = TaskAcceptPayload {
        task_id: "t1".into(),
        assignee: "agent://worker".into(),
        reason: "ready".into(),
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "TaskAccept", "m3", &sid, "agent://worker", accept),
        None,
    )
    .await
    .unwrap();

    let complete = TaskCompletePayload {
        task_id: "t1".into(),
        assignee: "agent://worker".into(),
        output: b"result".to_vec(),
        summary: "done".into(),
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "TaskComplete", "m4", &sid, "agent://worker", complete),
        None,
    )
    .await
    .unwrap();

    let result = rt
        .process(
            &envelope(
                mode,
                "Commitment",
                "m5",
                &sid,
                "agent://planner",
                commitment("task.completed"),
            ),
            None,
        )
        .await
        .unwrap();
    assert_eq!(result.session_state, SessionState::Resolved);
}

#[tokio::test]
async fn handoff_full_lifecycle_through_runtime() {
    use macp_runtime::handoff_pb::{HandoffAcceptPayload, HandoffOfferPayload};

    let rt = make_runtime();
    let sid = new_sid();
    let mode = "macp.mode.handoff.v1";

    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://owner",
            session_start(vec!["agent://owner".into(), "agent://target".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    let offer = HandoffOfferPayload {
        handoff_id: "h1".into(),
        target_participant: "agent://target".into(),
        scope: "support".into(),
        reason: "escalate".into(),
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "HandoffOffer", "m2", &sid, "agent://owner", offer),
        None,
    )
    .await
    .unwrap();

    let accept = HandoffAcceptPayload {
        handoff_id: "h1".into(),
        accepted_by: "agent://target".into(),
        reason: "ready".into(),
        implicit: false,
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "HandoffAccept", "m3", &sid, "agent://target", accept),
        None,
    )
    .await
    .unwrap();

    let result = rt
        .process(
            &envelope(
                mode,
                "Commitment",
                "m4",
                &sid,
                "agent://owner",
                commitment("handoff.accepted"),
            ),
            None,
        )
        .await
        .unwrap();
    assert_eq!(result.session_state, SessionState::Resolved);
}

#[tokio::test]
async fn quorum_full_lifecycle_through_runtime() {
    use macp_runtime::quorum_pb::{ApprovalRequestPayload, ApprovePayload};

    let rt = make_runtime();
    let sid = new_sid();
    let mode = "macp.mode.quorum.v1";

    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://coordinator",
            session_start(vec![
                "agent://alice".into(),
                "agent://bob".into(),
                "agent://carol".into(),
            ]),
        ),
        None,
    )
    .await
    .unwrap();

    let request = ApprovalRequestPayload {
        request_id: "r1".into(),
        action: "deploy.production".into(),
        summary: "Deploy v2".into(),
        details: vec![],
        required_approvals: 2,
    }
    .encode_to_vec();
    rt.process(
        &envelope(
            mode,
            "ApprovalRequest",
            "m2",
            &sid,
            "agent://coordinator",
            request,
        ),
        None,
    )
    .await
    .unwrap();

    let approve1 = ApprovePayload {
        request_id: "r1".into(),
        reason: "lgtm".into(),
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "Approve", "m3", &sid, "agent://alice", approve1),
        None,
    )
    .await
    .unwrap();

    let approve2 = ApprovePayload {
        request_id: "r1".into(),
        reason: "ship it".into(),
    }
    .encode_to_vec();
    rt.process(
        &envelope(mode, "Approve", "m4", &sid, "agent://bob", approve2),
        None,
    )
    .await
    .unwrap();

    let result = rt
        .process(
            &envelope(
                mode,
                "Commitment",
                "m5",
                &sid,
                "agent://coordinator",
                commitment("quorum.approved"),
            ),
            None,
        )
        .await
        .unwrap();
    assert_eq!(result.session_state, SessionState::Resolved);
}

#[tokio::test]
async fn multi_round_full_lifecycle_through_runtime() {
    let rt = make_runtime();
    let sid = new_sid();
    let mode = "ext.multi_round.v1";

    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://coordinator",
            session_start(vec!["agent://alice".into(), "agent://bob".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    rt.process(
        &envelope(
            mode,
            "Contribute",
            "m2",
            &sid,
            "agent://alice",
            br#"{"value":"option_a"}"#.to_vec(),
        ),
        None,
    )
    .await
    .unwrap();

    rt.process(
        &envelope(
            mode,
            "Contribute",
            "m3",
            &sid,
            "agent://bob",
            br#"{"value":"option_b"}"#.to_vec(),
        ),
        None,
    )
    .await
    .unwrap();

    rt.process(
        &envelope(
            mode,
            "Contribute",
            "m4",
            &sid,
            "agent://bob",
            br#"{"value":"option_a"}"#.to_vec(),
        ),
        None,
    )
    .await
    .unwrap();

    // Session still Open — convergence tracked, requires Commitment
    let session = rt.get_session_checked(&sid).await.unwrap();
    assert_eq!(session.state, SessionState::Open);

    let result = rt
        .process(
            &envelope(
                mode,
                "Commitment",
                "m5",
                &sid,
                "agent://coordinator",
                commitment("multi_round.converged"),
            ),
            None,
        )
        .await
        .unwrap();
    assert_eq!(result.session_state, SessionState::Resolved);
}

/// One clock per internal log entry (Phase 11a / follow-on 10).
///
/// `suspend_session` and `resume_session` each read the wall clock exactly
/// once and hand that same instant to both the `Session` mutation and
/// `make_internal_entry`. That makes the live `accumulated_suspended_ms` and
/// the value replay re-derives from the two entry timestamps identical *by
/// construction* — which matters because since Phase 10 that value gates an
/// accept/reject decision, so a sub-millisecond divergence could make a
/// live-`Resolved` session fail replay outright.
///
/// Honest note on what this test can and cannot do: with the clock injected
/// the equality holds deterministically, and before the fix it would only have
/// broken when a millisecond tick happened to land between the two `Utc::now()`
/// reads. So this pins the invariant rather than differentially proving the old
/// bug; the injected-clock signature is the real guarantee.
#[tokio::test]
async fn suspend_resume_entries_share_the_session_mutation_clock() {
    let storage: Arc<dyn macp_runtime::storage::StorageBackend> = Arc::new(MemoryBackend);
    let registry = Arc::new(SessionRegistry::new());
    let log_store = Arc::new(LogStore::new());
    let rt = Runtime::new(storage, registry, Arc::clone(&log_store));

    let mode = "macp.mode.decision.v1";
    let sid = new_sid();
    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://coordinator",
            session_start(vec!["agent://coordinator".into(), "agent://alice".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    rt.suspend_session(&sid, "pause", "agent://coordinator")
        .await
        .unwrap();
    // Let a millisecond boundary pass so the banked duration is non-trivial;
    // the assertion below is an exact equality either way.
    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
    rt.resume_session(&sid, "carry on", "agent://coordinator")
        .await
        .unwrap();

    let entries = log_store.get_log(&sid).await.unwrap();
    let stamp_of = |message_type: &str| -> i64 {
        let matching: Vec<&macp_runtime::log_store::LogEntry> = entries
            .iter()
            .filter(|e| e.message_type == message_type)
            .collect();
        assert_eq!(
            matching.len(),
            1,
            "expected exactly one {message_type} entry, got {}",
            matching.len()
        );
        matching[0].received_at_ms
    };
    let suspended_at = stamp_of("SessionSuspend");
    let resumed_at = stamp_of("SessionResume");

    let session = rt.get_session_checked(&sid).await.unwrap();
    assert_eq!(session.state, SessionState::Open);
    assert_eq!(
        session.accumulated_suspended_ms,
        resumed_at - suspended_at,
        "live accumulated_suspended_ms must equal the span between the two \
         internal log entries, or replay reconstructs a different value"
    );
    assert!(
        session.accumulated_suspended_ms > 0,
        "the suspension must have measured something, else the equality is vacuous"
    );

    // Phase 11b acceptance criterion 4 — live/replay agreement on the new
    // `suspension_intervals` vec. The live session records the pair in
    // `Session::resume`; replay reconstructs it by driving the same method
    // from the two internal entries' recorded timestamps, so the two must be
    // identical, not merely consistent.
    assert_eq!(
        session.suspension_intervals,
        vec![(suspended_at, resumed_at)],
        "the live session's completed-pause record must match the internal \
         log entries it was derived from"
    );
    let replay_registry = macp_runtime::mode_registry::ModeRegistry::build_default(Arc::new(
        macp_runtime::policy::DefaultPolicyEvaluator,
    ));
    let replayed = macp_runtime::replay::replay_session(&sid, &entries, &replay_registry, None)
        .expect("replay must succeed");
    assert_eq!(
        replayed.suspension_intervals, session.suspension_intervals,
        "replay must reconstruct the identical suspension intervals"
    );
    assert_eq!(
        replayed.accumulated_suspended_ms,
        session.accumulated_suspended_ms
    );
}

/// RFC-MACP-0006 §3.2:117 — `SessionSuspend`/`SessionResume` internal entries
/// MUST NOT consume accepted ordinals. Client-visible ordinals stay
/// contiguous across a suspend/resume cycle, and stay stable across a
/// simulated runtime restart too (§3.2:123 — "An ordinal MUST be stable for
/// the life of the session — across runtime restarts").
///
/// Every negative assertion here (the suspend/resume pair inserts no gap) is
/// paired with a positive one (the exact expected ids and ordinals), so the
/// test cannot pass by the feature having disappeared rather than by the
/// invariant holding — see Phase 1 criterion 6 for the mutation proof that
/// this test actually reds if `make_internal_entry` is changed to
/// `EntryKind::Incoming`.
#[tokio::test]
async fn suspend_resume_does_not_consume_accepted_ordinals() {
    let storage: Arc<dyn macp_runtime::storage::StorageBackend> = Arc::new(MemoryBackend);
    let registry = Arc::new(SessionRegistry::new());
    let log_store = Arc::new(LogStore::new());
    let rt = Runtime::new(storage, registry, Arc::clone(&log_store));

    let mode = "macp.mode.decision.v1";
    let sid = new_sid();
    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://coordinator",
            session_start(vec!["agent://coordinator".into(), "agent://alice".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    let proposal_before = macp_runtime::decision_pb::ProposalPayload {
        proposal_id: "p1".into(),
        option: "deploy".into(),
        rationale: "before suspend".into(),
        supporting_data: vec![],
    }
    .encode_to_vec();
    rt.process(
        &envelope(
            mode,
            "Proposal",
            "m2",
            &sid,
            "agent://coordinator",
            proposal_before,
        ),
        None,
    )
    .await
    .unwrap();

    rt.suspend_session(&sid, "pause", "agent://coordinator")
        .await
        .unwrap();
    rt.resume_session(&sid, "carry on", "agent://coordinator")
        .await
        .unwrap();

    let proposal_after = macp_runtime::decision_pb::ProposalPayload {
        proposal_id: "p2".into(),
        option: "hold".into(),
        rationale: "after resume".into(),
        supporting_data: vec![],
    }
    .encode_to_vec();
    rt.process(
        &envelope(
            mode,
            "Proposal",
            "m3",
            &sid,
            "agent://coordinator",
            proposal_after,
        ),
        None,
    )
    .await
    .unwrap();

    let ordinals = rt.log_store.get_incoming_after(&sid, 0).await.unwrap();
    // Projected to (ordinal, message_id): `LogEntry` has no `PartialEq`
    // (only `EntryKind` does), so a direct `assert_eq!` on the returned
    // `Vec<(u64, LogEntry)>` does not compile.
    let projection: Vec<(u64, String)> = ordinals
        .iter()
        .map(|(ordinal, entry)| (*ordinal, entry.message_id.clone()))
        .collect();
    assert_eq!(
        projection,
        vec![
            (1, "m1".to_string()),
            (2, "m2".to_string()),
            (3, "m3".to_string())
        ],
        "SessionSuspend/SessionResume must not consume an accepted ordinal or \
         appear in the ordinal-consuming sequence (RFC-MACP-0006 §3.2:117)"
    );

    let raw_log = log_store.get_log(&sid).await.unwrap();
    assert_eq!(
        raw_log.len(),
        5,
        "the durable log holds the 3 client envelopes plus the SessionSuspend \
         and SessionResume internal annotations — proving the entries were \
         written and excluded, not merely absent"
    );

    // Criterion 2a: the same projection survives a simulated restart.
    // Deliberately NOT checked via `replay_session` on this same `LogStore` —
    // `replay_session` rebuilds a `Session` and never touches the log, so
    // that comparison would be vacuous. Build a fresh store and replay the
    // collected entries into it the way startup does
    // (`create_session_log` then `append` per entry, `src/main.rs:371-374`).
    let restarted_store = LogStore::new();
    restarted_store.create_session_log(&sid).await;
    for entry in &raw_log {
        restarted_store.append(&sid, entry.clone()).await;
    }
    let restarted_ordinals = restarted_store.get_incoming_after(&sid, 0).await.unwrap();
    let restarted_projection: Vec<(u64, String)> = restarted_ordinals
        .iter()
        .map(|(ordinal, entry)| (*ordinal, entry.message_id.clone()))
        .collect();
    assert_eq!(
        restarted_projection, projection,
        "an ordinal must be stable across a runtime restart (RFC-MACP-0006 §3.2:123)"
    );
}

/// RFC-MACP-0006 §3.2:117 — `SessionCancel` MUST NOT consume an accepted
/// ordinal either. Runs on its own session, separately from the
/// suspend/resume test above, because `cancel_session` is terminal and
/// triggers `maybe_compact_log`, whose behavior forks on the storage backend:
///
/// - On `MemoryBackend` (used here), `replace_log` resolves to the trait's
///   default impl, which returns `Err(Unsupported)`, so compaction fails and
///   `force_insert_checkpoint` appends a `Checkpoint` entry with
///   `compacted_incoming_ordinals: 0` instead — the log grows by **two**
///   non-ordinal-consuming entries (`SessionCancel` + `Checkpoint`), and the
///   before/after ordinal projection is unaffected.
/// - On `FileBackend`, compaction succeeds and collapses the log to a single
///   checkpoint with a nonzero `compacted_incoming_ordinals` base, so
///   `get_incoming_after(&sid, 0)` would instead return `Err(base)` — a
///   different, also-correct shape this test does not exercise (see
///   `tests/handoff_implicit_accept_live.rs`'s two harnesses for that case).
///
/// Comparing before against after is stronger than an absolute count here,
/// and is immune to the force-inserted checkpoint (its
/// `compacted_incoming_ordinals: 0` cannot reset a base set by an earlier
/// real compaction — there is none in this test — because the base is a
/// `.max()` over checkpoints).
#[tokio::test]
async fn cancel_session_does_not_consume_accepted_ordinals() {
    let storage: Arc<dyn macp_runtime::storage::StorageBackend> = Arc::new(MemoryBackend);
    let registry = Arc::new(SessionRegistry::new());
    let log_store = Arc::new(LogStore::new());
    let rt = Runtime::new(storage, registry, Arc::clone(&log_store));

    let mode = "macp.mode.decision.v1";
    let sid = new_sid();
    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://coordinator",
            session_start(vec!["agent://coordinator".into(), "agent://alice".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    let before = rt.log_store.get_incoming_after(&sid, 0).await.unwrap();
    let before_projection: Vec<(u64, String)> = before
        .iter()
        .map(|(ordinal, entry)| (*ordinal, entry.message_id.clone()))
        .collect();
    let raw_len_before = log_store.get_log(&sid).await.unwrap().len();

    rt.cancel_session(&sid, "done", "agent://coordinator")
        .await
        .unwrap();

    let after = rt.log_store.get_incoming_after(&sid, 0).await.unwrap();
    let after_projection: Vec<(u64, String)> = after
        .iter()
        .map(|(ordinal, entry)| (*ordinal, entry.message_id.clone()))
        .collect();
    assert_eq!(
        after_projection, before_projection,
        "SessionCancel must not consume an accepted ordinal (RFC-MACP-0006 §3.2:117)"
    );
    assert!(
        !before_projection.is_empty(),
        "the projection must contain the SessionStart entry, else the equality above is vacuous"
    );

    let raw_len_after = log_store.get_log(&sid).await.unwrap().len();
    assert_eq!(
        raw_len_after,
        raw_len_before + 2,
        "on MemoryBackend, cancel appends SessionCancel plus a force-inserted \
         Checkpoint (compaction has no replace_log override on this backend) \
         — both non-ordinal-consuming, proving the entries were written and \
         excluded rather than never written"
    );
}

/// RFC-MACP-0001 §7.5 / RFC-MACP-0003 §2: `SessionResumePayload.banked_ms`
/// records the remaining TTL banked at suspend (`deadline − t_s`), not the
/// pause's duration (`t_r − t_s`). The fixture's suspension is short (tens of
/// ms) against a 60s TTL suspended almost immediately after start, so the two
/// quantities differ by more than an order of magnitude — a `banked_ms` that
/// still recorded the old (wrong) quantity could not pass the exact-equality
/// assertion below by coincidence.
#[tokio::test]
async fn resume_banks_the_remaining_ttl_at_suspend_not_the_pause_duration() {
    let storage: Arc<dyn macp_runtime::storage::StorageBackend> = Arc::new(MemoryBackend);
    let registry = Arc::new(SessionRegistry::new());
    let log_store = Arc::new(LogStore::new());
    let rt = Runtime::new(storage, registry, Arc::clone(&log_store));

    let mode = "macp.mode.decision.v1";
    let sid = new_sid();
    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://coordinator",
            session_start(vec!["agent://coordinator".into(), "agent://alice".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    // `Session::suspend` never touches `ttl_expiry`, so this pre-suspend
    // snapshot is exactly the spec's `deadline` — the value the formula banks
    // from.
    let ttl_expiry_before = rt.get_session_checked(&sid).await.unwrap().ttl_expiry;

    rt.suspend_session(&sid, "pause", "agent://coordinator")
        .await
        .unwrap();
    let suspended_at = log_store
        .get_log(&sid)
        .await
        .unwrap()
        .last()
        .unwrap()
        .received_at_ms;

    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    rt.resume_session(&sid, "carry on", "agent://coordinator")
        .await
        .unwrap();

    let resume_entry = log_store
        .get_log(&sid)
        .await
        .unwrap()
        .last()
        .unwrap()
        .clone();
    let resumed_at = resume_entry.received_at_ms;
    let payload = SessionResumePayload::decode(&*resume_entry.raw_payload).unwrap();

    let expected_banked_ms = ttl_expiry_before.saturating_sub(suspended_at).max(0);
    assert_eq!(
        payload.banked_ms, expected_banked_ms,
        "banked_ms must be the remaining TTL at suspend (RFC-MACP-0001 §7.5, \
         RFC-MACP-0003 §2), not the pause's duration"
    );

    let pause_duration = resumed_at - suspended_at;
    assert!(
        payload.banked_ms > pause_duration * 10,
        "fixture must distinguish the two quantities by more than an order of \
         magnitude: banked_ms={}, pause_duration={}",
        payload.banked_ms,
        pause_duration
    );

    // Criterion 4: the deadline arithmetic itself is untouched by this phase
    // — only the recorded field's quantity changed, not what the session's
    // actual TTL banks.
    let session_after = rt.get_session_checked(&sid).await.unwrap();
    assert_eq!(
        session_after.ttl_expiry,
        ttl_expiry_before + (resumed_at - suspended_at),
        "the TTL deadline must still bank exactly the pause's duration, \
         unaffected by the banked_ms field's corrected quantity"
    );
}

/// RFC-MACP-0003 §2's per-event formula makes `banked_ms` non-monotonic
/// across suspend/resume cycles by design: each cycle's value is the
/// remaining TTL as of *that* cycle's own suspend, which shrinks as the
/// session ages. A later reader must not "fix" this into a monotonic
/// quantity.
///
/// The clock advance between the first resume and the second suspend is
/// load-bearing, not cosmetic: since a second suspend cannot precede the
/// first resume, banked_2 <= banked_1 always, but the inequality is strict
/// only when the second suspend happens strictly after the first resume.
/// Without this sleep, a fast run could tie the two events at the same
/// millisecond and a strict `<` assertion would flake.
#[tokio::test]
async fn banked_ms_shrinks_across_two_suspend_resume_cycles() {
    let storage: Arc<dyn macp_runtime::storage::StorageBackend> = Arc::new(MemoryBackend);
    let registry = Arc::new(SessionRegistry::new());
    let log_store = Arc::new(LogStore::new());
    let rt = Runtime::new(storage, registry, Arc::clone(&log_store));

    let mode = "macp.mode.decision.v1";
    let sid = new_sid();
    rt.process(
        &envelope(
            mode,
            "SessionStart",
            "m1",
            &sid,
            "agent://coordinator",
            session_start(vec!["agent://coordinator".into(), "agent://alice".into()]),
        ),
        None,
    )
    .await
    .unwrap();

    rt.suspend_session(&sid, "pause 1", "agent://coordinator")
        .await
        .unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
    rt.resume_session(&sid, "carry on 1", "agent://coordinator")
        .await
        .unwrap();
    let banked_ms_1 = SessionResumePayload::decode(
        &*log_store
            .get_log(&sid)
            .await
            .unwrap()
            .last()
            .unwrap()
            .raw_payload,
    )
    .unwrap()
    .banked_ms;

    tokio::time::sleep(std::time::Duration::from_millis(5)).await;

    rt.suspend_session(&sid, "pause 2", "agent://coordinator")
        .await
        .unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
    rt.resume_session(&sid, "carry on 2", "agent://coordinator")
        .await
        .unwrap();
    let banked_ms_2 = SessionResumePayload::decode(
        &*log_store
            .get_log(&sid)
            .await
            .unwrap()
            .last()
            .unwrap()
            .raw_payload,
    )
    .unwrap()
    .banked_ms;

    assert!(
        banked_ms_2 < banked_ms_1,
        "banked_ms must strictly shrink across cycles once the clock has \
         genuinely advanced between them: banked_ms_1={banked_ms_1}, \
         banked_ms_2={banked_ms_2}"
    );
}

/// Edge case: a session suspended exactly at its deadline banks zero
/// remaining TTL — legal, and distinguishable from "field absent" only by
/// context.
#[tokio::test]
async fn banked_ms_is_zero_when_suspended_exactly_at_the_deadline() {
    let storage: Arc<dyn macp_runtime::storage::StorageBackend> = Arc::new(MemoryBackend);
    let registry = Arc::new(SessionRegistry::new());
    let log_store = Arc::new(LogStore::new());
    let sid = new_sid();
    let mode = "macp.mode.decision.v1";

    let deadline = 1_700_000_060_000;
    let mut session = Session::builder(sid.clone(), mode, "agent://coordinator").build();
    session.state = SessionState::Suspended;
    session.ttl_expiry = deadline;
    session.suspended_at_ms = Some(deadline);
    registry
        .insert_recovered_session(sid.clone(), session)
        .await;

    let rt = Runtime::new(storage, Arc::clone(&registry), Arc::clone(&log_store));
    let _ = rt
        .resume_session(&sid, "edge case", "agent://coordinator")
        .await;

    let entries = log_store.get_log(&sid).await.unwrap();
    let resume_entry = entries
        .iter()
        .find(|e| e.message_type == "SessionResume")
        .expect("SessionResume entry recorded even if resume itself later force-expires");
    let payload = SessionResumePayload::decode(&*resume_entry.raw_payload).unwrap();
    assert_eq!(
        payload.banked_ms, 0,
        "deadline - suspend_time == 0 when suspended exactly at the deadline"
    );
}

/// Edge case: a builder-constructed session with `suspended_at_ms` set past
/// `ttl_expiry` — unreachable through the ordinary suspend path, since
/// `suspend_session` runs `maybe_expire_session` first, but constructible by
/// a library consumer via `Session::builder` — must clamp `banked_ms` to
/// zero rather than go negative or panic.
#[tokio::test]
async fn banked_ms_clamps_to_zero_when_suspended_at_is_past_the_deadline() {
    let storage: Arc<dyn macp_runtime::storage::StorageBackend> = Arc::new(MemoryBackend);
    let registry = Arc::new(SessionRegistry::new());
    let log_store = Arc::new(LogStore::new());
    let sid = new_sid();
    let mode = "macp.mode.decision.v1";

    let deadline = 1_700_000_000_000;
    let mut session = Session::builder(sid.clone(), mode, "agent://coordinator").build();
    session.state = SessionState::Suspended;
    session.ttl_expiry = deadline;
    session.suspended_at_ms = Some(deadline + 5_000); // inconsistent: past the deadline
    registry
        .insert_recovered_session(sid.clone(), session)
        .await;

    let rt = Runtime::new(storage, Arc::clone(&registry), Arc::clone(&log_store));
    // Deliberately ignore the Result: `suspended_at_ms` here (a fixed point
    // in 2023) is far in the past relative to the real wall clock
    // `resume_session` reads for the *unrelated* MAX_SUSPEND_MS cap check
    // inside `Session::resume`, which will force-expire this session —
    // exactly the "still records banked_ms" case documented on
    // `resume_session`. The value under test is the appended entry's
    // payload, not this call's Result.
    let _ = rt
        .resume_session(&sid, "edge case", "agent://coordinator")
        .await;

    let entries = log_store.get_log(&sid).await.unwrap();
    let resume_entry = entries
        .iter()
        .find(|e| e.message_type == "SessionResume")
        .expect("SessionResume entry recorded even though resume force-expires");
    let payload = SessionResumePayload::decode(&*resume_entry.raw_payload).unwrap();
    assert_eq!(
        payload.banked_ms, 0,
        "ttl_expiry.saturating_sub(suspended_at) must clamp to 0, not go \
         negative or panic, when suspended_at_ms is past ttl_expiry"
    );
}