use std::collections::BTreeMap;

use wip_client::{
    CallOutcome, Client, ClientErrorKind, ClientLimits, Completion, DispatchedTransportFailure,
    ObservationState, OutcomeUnknownReason, SecurityContext,
};
use wip_https::http::Response;
use wip_https::{
    Limits, encode_call_operation_response, encode_fetch_interface_response, encode_fetch_response,
    encode_fetch_tree_response, encode_protocol_error_response,
};
use wip_protocol::{
    CallOperationResponse, FetchInterfaceRequest, FetchInterfaceResponse, FetchRequest,
    FetchTreeRequest, FetchTreeResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object,
    ObjectTreeNode, OperationDeclaration, ParameterDeclaration, ProtocolError, ProtocolErrorCode,
    ProtocolInteraction, ReturnDeclaration, TypeExpr, Value,
};

fn wire_limits() -> Limits {
    Limits::new(64 * 1024, 64 * 1024, 64, 64).unwrap()
}

fn client(objects: usize, interfaces: usize, in_flight: usize, history: usize) -> Client {
    Client::new(
        ClientLimits::new(8, objects, interfaces, in_flight, history).unwrap(),
        wire_limits(),
    )
}

fn object(name: &str, reference: Option<&str>, validator: &[u8]) -> Object {
    Object {
        name: name.into(),
        description: Some(format!("object {name}")),
        interfaces: vec!["math".into()],
        r#ref: reference.map(str::to_owned),
        validator: Some(validator.to_vec()),
    }
}

fn descriptor(return_type: TypeExpr) -> InterfaceDescriptor {
    InterfaceDescriptor {
        format: INTERFACE_FORMAT_V1.into(),
        documentation: None,
        types: vec![],
        operations: vec![OperationDeclaration {
            name: "calculate".into(),
            documentation: None,
            parameters: vec![ParameterDeclaration {
                name: "value".into(),
                required: true,
                documentation: None,
                r#type: TypeExpr::Integer,
            }],
            returns: ReturnDeclaration {
                documentation: None,
                r#type: return_type,
            },
        }],
    }
}

fn fetch_response(path: &str, value: &Object) -> Response<Vec<u8>> {
    encode_fetch_response(&FetchRequest { path: path.into() }, value, wire_limits()).unwrap()
}

fn observe_object(client: &mut Client, session: &wip_client::SessionId, path: &str, value: Object) {
    let request = client.prepare_fetch(session, path).unwrap();
    client
        .complete(request.id, fetch_response(path, &value))
        .unwrap();
}

fn observe_interface(
    client: &mut Client,
    session: &wip_client::SessionId,
    value: FetchInterfaceResponse,
) {
    let request = client
        .prepare_interface(session, value.interface.clone())
        .unwrap();
    let response = encode_fetch_interface_response(
        &FetchInterfaceRequest {
            interface: value.interface.clone(),
        },
        &value,
        wire_limits(),
    )
    .unwrap();
    client.complete(request.id, response).unwrap();
}

fn args(value: i64) -> BTreeMap<String, Value> {
    BTreeMap::from([("value".into(), Value::Integer(value))])
}

#[test]
fn sessions_partition_known_space_by_canonical_endpoint_and_security_context() {
    let mut client = client(8, 8, 8, 8);
    let alice = client
        .open_session(
            "https://example.test/wip",
            SecurityContext::new("alice-token"),
        )
        .unwrap();
    let alice_canonical = client
        .open_session(
            "https://example.test/wip/",
            SecurityContext::new("alice-token"),
        )
        .unwrap();
    let bob = client
        .open_session(
            "https://example.test/wip",
            SecurityContext::new("bob-token"),
        )
        .unwrap();
    let other = client
        .open_session(
            "https://other.test/wip",
            SecurityContext::new("alice-token"),
        )
        .unwrap();

    assert_eq!(alice, alice_canonical);
    observe_object(&mut client, &alice, "/item", object("item", None, b"a"));
    assert!(client.object(&alice, "/item").is_some());
    assert!(client.object(&bob, "/item").is_none());
    assert!(client.object(&other, "/item").is_none());
}

#[test]
fn newer_refresh_rejects_out_of_order_response_for_the_same_subject() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    let old = client.prepare_fetch(&session, "/item").unwrap();
    let new = client.prepare_fetch(&session, "/item").unwrap();

    assert_eq!(
        client
            .complete(
                old.id,
                fetch_response("/item", &object("item", None, b"old"))
            )
            .unwrap(),
        Completion::StaleResponseRejected(old.id)
    );
    assert!(matches!(
        client.object(&session, "/item").unwrap().state,
        ObservationState::Loading(id) if id == new.id
    ));

    client
        .complete(
            new.id,
            fetch_response("/item", &object("item", None, b"new")),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/item").unwrap().validator,
        Some(b"new".to_vec())
    );
}

#[test]
fn normalized_object_cache_rejects_conflicts_and_evicts_to_its_bound() {
    let mut client = client(2, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_object(
        &mut client,
        &session,
        "/a/shared",
        object("shared", Some("shared-ref"), b"v1"),
    );
    observe_object(
        &mut client,
        &session,
        "/b/shared",
        object("shared", Some("shared-ref"), b"v1"),
    );

    let conflict = client.prepare_fetch(&session, "/b/shared").unwrap();
    let mut changed = object("shared", Some("shared-ref"), b"v1");
    changed.description = Some("contradiction".into());
    let error = client
        .complete(conflict.id, fetch_response("/b/shared", &changed))
        .unwrap_err();
    assert_eq!(error.kind, ClientErrorKind::InvalidResponse);

    observe_object(
        &mut client,
        &session,
        "/second",
        object("second", None, b"2"),
    );
    observe_object(&mut client, &session, "/third", object("third", None, b"3"));
    assert!(client.object(&session, "/a/shared").is_none());
    assert!(client.object(&session, "/second").is_some());
    assert!(client.object(&session, "/third").is_some());
}

#[test]
fn normalized_aliases_preserve_newer_request_ownership() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_object(
        &mut client,
        &session,
        "/a/shared",
        object("shared", Some("shared-ref"), b"initial"),
    );
    observe_object(
        &mut client,
        &session,
        "/b/shared",
        object("shared", Some("shared-ref"), b"initial"),
    );
    let request = FetchTreeRequest {
        path: "/a".into(),
        depth: 1,
        limit: 4,
    };
    let older_tree_response = || FetchTreeResponse {
        root: ObjectTreeNode {
            object: object("a", None, b"root"),
            children: vec![ObjectTreeNode {
                object: object("shared", Some("shared-ref"), b"older"),
                children: vec![],
            }],
        },
        truncated: false,
    };

    let older_tree = client.prepare_tree(&session, "/a", 1, 4).unwrap();
    let newer_alias = client.prepare_fetch(&session, "/b/shared").unwrap();
    client
        .complete(
            newer_alias.id,
            fetch_response("/b/shared", &object("shared", Some("shared-ref"), b"newer")),
        )
        .unwrap();
    client
        .complete(
            older_tree.id,
            encode_fetch_tree_response(&request, &older_tree_response(), wire_limits()).unwrap(),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/a/shared").unwrap().validator,
        Some(b"newer".to_vec())
    );
    assert_eq!(
        client.object(&session, "/b/shared").unwrap().validator,
        Some(b"newer".to_vec())
    );

    let older_tree = client.prepare_tree(&session, "/a", 1, 4).unwrap();
    let newer_alias = client.prepare_fetch(&session, "/b/shared").unwrap();
    client
        .complete(
            older_tree.id,
            encode_fetch_tree_response(&request, &older_tree_response(), wire_limits()).unwrap(),
        )
        .unwrap();
    assert!(matches!(
        client.object(&session, "/b/shared").unwrap().state,
        ObservationState::Loading(id) if id == newer_alias.id
    ));
    assert_eq!(
        client.object(&session, "/b/shared").unwrap().validator,
        Some(b"newer".to_vec())
    );
    client
        .complete(
            newer_alias.id,
            fetch_response(
                "/b/shared",
                &object("shared", Some("shared-ref"), b"newest"),
            ),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/a/shared").unwrap().validator,
        Some(b"newest".to_vec())
    );

    let older_tree = client.prepare_tree(&session, "/a", 1, 4).unwrap();
    let newer_alias = client.prepare_fetch(&session, "/b/shared").unwrap();
    client
        .complete(
            newer_alias.id,
            fetch_response(
                "/b/shared",
                &object("shared", Some("shared-ref"), b"newest"),
            ),
        )
        .unwrap();
    let mut contradictory = object("shared", Some("shared-ref"), b"newest");
    contradictory.description = Some("contradictory representation".into());
    let error = client
        .complete(
            older_tree.id,
            encode_fetch_tree_response(
                &request,
                &FetchTreeResponse {
                    root: ObjectTreeNode {
                        object: object("a", None, b"root"),
                        children: vec![ObjectTreeNode {
                            object: contradictory,
                            children: vec![],
                        }],
                    },
                    truncated: false,
                },
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap_err();
    assert_eq!(error.kind, ClientErrorKind::InvalidResponse);
    assert_eq!(
        client.object(&session, "/b/shared").unwrap().validator,
        Some(b"newest".to_vec())
    );

    observe_object(
        &mut client,
        &session,
        "/child",
        object("child", Some("child-ref"), b"stable"),
    );
    let older_tree = client.prepare_tree(&session, "/", 1, 4).unwrap();
    let newer_child = client.prepare_fetch(&session, "/child").unwrap();
    client
        .complete(
            newer_child.id,
            fetch_response("/child", &object("child", Some("child-ref"), b"stable")),
        )
        .unwrap();
    let mut contradictory = object("child", Some("child-ref"), b"stable");
    contradictory.description = Some("same-path contradiction".into());
    let root_request = FetchTreeRequest {
        path: "/".into(),
        depth: 1,
        limit: 4,
    };
    let error = client
        .complete(
            older_tree.id,
            encode_fetch_tree_response(
                &root_request,
                &FetchTreeResponse {
                    root: ObjectTreeNode {
                        object: object("", None, b"root"),
                        children: vec![ObjectTreeNode {
                            object: contradictory,
                            children: vec![],
                        }],
                    },
                    truncated: false,
                },
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap_err();
    assert_eq!(error.kind, ClientErrorKind::InvalidResponse);
    assert_eq!(
        client.object(&session, "/child").unwrap().state,
        ObservationState::Fresh
    );
}

#[test]
fn tree_expansion_is_lazy_truncated_and_refreshes_edges() {
    let mut client = client(16, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    assert!(client.tree(&session, "/").is_none());

    let request = client.prepare_tree(&session, "/", 1, 4).unwrap();
    let logical_request = FetchTreeRequest {
        path: "/".into(),
        depth: 1,
        limit: 4,
    };
    let tree = FetchTreeResponse {
        root: ObjectTreeNode {
            object: object("", None, b"root"),
            children: vec![ObjectTreeNode {
                object: object("items", None, b"items"),
                children: vec![],
            }],
        },
        truncated: true,
    };
    client
        .complete(
            request.id,
            encode_fetch_tree_response(&logical_request, &tree, wire_limits()).unwrap(),
        )
        .unwrap();
    assert_eq!(client.tree(&session, "/").unwrap().children, ["/items"]);
    assert!(client.tree(&session, "/").unwrap().truncated);
    assert!(client.object(&session, "/items").is_some());

    let refresh = client.prepare_tree(&session, "/", 1, 4).unwrap();
    let refreshed = FetchTreeResponse {
        root: ObjectTreeNode {
            object: object("", None, b"root-2"),
            children: vec![],
        },
        truncated: false,
    };
    client
        .complete(
            refresh.id,
            encode_fetch_tree_response(&logical_request, &refreshed, wire_limits()).unwrap(),
        )
        .unwrap();
    assert!(client.tree(&session, "/").unwrap().children.is_empty());
    assert!(!client.tree(&session, "/").unwrap().truncated);
    assert_eq!(
        client.object(&session, "/items").unwrap().state,
        ObservationState::Stale
    );
}

#[test]
fn call_context_fixes_membership_validators_descriptor_and_result_validation() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_object(
        &mut client,
        &session,
        "/item",
        object("item", Some("item-ref"), b"object-v1"),
    );
    observe_interface(
        &mut client,
        &session,
        FetchInterfaceResponse {
            interface: "math".into(),
            descriptor: descriptor(TypeExpr::Integer),
            validator: Some(b"interface-v1".to_vec()),
        },
    );

    let call = client
        .prepare_call(&session, "/item", "math", "calculate", args(7))
        .unwrap();
    assert_eq!(
        client
            .pending_call_context(call.id)
            .unwrap()
            .operation()
            .name,
        "calculate"
    );
    client.mark_dispatched(call.id).unwrap();

    // A newer interface observation cannot change this in-flight call's schema.
    observe_interface(
        &mut client,
        &session,
        FetchInterfaceResponse {
            interface: "math".into(),
            descriptor: descriptor(TypeExpr::String),
            validator: Some(b"interface-v2".to_vec()),
        },
    );

    let history_context = match client
        .complete(
            call.id,
            encode_call_operation_response(
                &wip_protocol::CallOperationRequest {
                    target: wip_protocol::Target {
                        path: "/item".into(),
                        validator: Some(b"object-v1".to_vec()),
                    },
                    interface: wip_protocol::InterfaceTarget {
                        reference: "math".into(),
                        validator: Some(b"interface-v1".to_vec()),
                    },
                    operation: "calculate".into(),
                    arguments: args(7),
                },
                &descriptor(TypeExpr::Integer),
                &CallOperationResponse {
                    result: Value::Integer(14),
                    validator: Some(b"object-v2".to_vec()),
                },
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap()
    {
        Completion::Call(record) => record,
        other => panic!("unexpected completion: {other:?}"),
    };
    assert_eq!(
        history_context.context.descriptor(),
        &descriptor(TypeExpr::Integer)
    );
    assert_eq!(
        history_context.context.object().validator,
        Some(b"object-v1".to_vec())
    );
    assert_eq!(
        history_context.context.interface().validator,
        Some(b"interface-v1".to_vec())
    );
    assert!(matches!(history_context.outcome, CallOutcome::Success(_)));
    assert_eq!(
        client.interface(&session, "math").unwrap().descriptor,
        Some(descriptor(TypeExpr::String))
    );
    assert_eq!(
        client.object(&session, "/item").unwrap().state,
        ObservationState::Stale
    );
}

#[test]
fn older_retrieval_cannot_restore_freshness_after_operation_completion() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_object(
        &mut client,
        &session,
        "/item",
        object("item", Some("shared-ref"), b"object-v1"),
    );
    observe_interface(
        &mut client,
        &session,
        FetchInterfaceResponse {
            interface: "math".into(),
            descriptor: descriptor(TypeExpr::Integer),
            validator: Some(b"interface-v1".to_vec()),
        },
    );

    let older_alias = client.prepare_fetch(&session, "/alias/item").unwrap();
    let call = client
        .prepare_call(&session, "/item", "math", "calculate", args(7))
        .unwrap();
    let request = client
        .pending_call_context(call.id)
        .unwrap()
        .request()
        .clone();
    let fixed_descriptor = client
        .pending_call_context(call.id)
        .unwrap()
        .descriptor()
        .clone();
    client.mark_dispatched(call.id).unwrap();
    client
        .complete(
            call.id,
            encode_call_operation_response(
                &request,
                &fixed_descriptor,
                &CallOperationResponse {
                    result: Value::Integer(14),
                    validator: Some(b"object-v2".to_vec()),
                },
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/item").unwrap().state,
        ObservationState::Stale
    );

    client
        .complete(
            older_alias.id,
            fetch_response(
                "/alias/item",
                &object("item", Some("shared-ref"), b"object-v1"),
            ),
        )
        .unwrap();
    let alias = client.object(&session, "/alias/item").unwrap();
    assert_eq!(alias.state, ObservationState::Stale);
    assert_eq!(alias.validator, Some(b"object-v2".to_vec()));
    assert_eq!(
        client
            .prepare_call(&session, "/alias/item", "math", "calculate", args(1))
            .unwrap_err()
            .kind,
        ClientErrorKind::MissingObservation
    );

    observe_object(
        &mut client,
        &session,
        "/item",
        object("item", Some("shared-ref"), b"object-v2"),
    );
    let older_alias = client.prepare_fetch(&session, "/second/item").unwrap();
    let call = client
        .prepare_call(&session, "/item", "math", "calculate", args(8))
        .unwrap();
    client.mark_dispatched(call.id).unwrap();
    let unknown = ProtocolError {
        code: ProtocolErrorCode::OperationOutcomeUnknown,
        message: "commit uncertain".into(),
    };
    client
        .complete(
            call.id,
            encode_protocol_error_response(
                ProtocolInteraction::CallOperation,
                &unknown,
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();
    client
        .complete(
            older_alias.id,
            fetch_response(
                "/second/item",
                &object("item", Some("shared-ref"), b"object-v2"),
            ),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/second/item").unwrap().state,
        ObservationState::Stale
    );
}

#[test]
fn in_flight_calls_pin_target_identity_until_staleness_is_applied() {
    let mut client = client(1, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_object(
        &mut client,
        &session,
        "/item",
        object("item", Some("shared-ref"), b"object-v1"),
    );
    observe_interface(
        &mut client,
        &session,
        FetchInterfaceResponse {
            interface: "math".into(),
            descriptor: descriptor(TypeExpr::Integer),
            validator: Some(b"interface-v1".to_vec()),
        },
    );
    let call = client
        .prepare_call(&session, "/item", "math", "calculate", args(7))
        .unwrap();
    client.mark_dispatched(call.id).unwrap();

    assert_eq!(
        client.prepare_fetch(&session, "/other").unwrap_err().kind,
        ClientErrorKind::Capacity
    );
    let unknown = ProtocolError {
        code: ProtocolErrorCode::OperationOutcomeUnknown,
        message: "commit uncertain".into(),
    };
    client
        .complete(
            call.id,
            encode_protocol_error_response(
                ProtocolInteraction::CallOperation,
                &unknown,
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/item").unwrap().state,
        ObservationState::Stale
    );

    observe_object(
        &mut client,
        &session,
        "/other",
        object("other", None, b"other"),
    );
    assert!(client.object(&session, "/item").is_none());
    assert!(client.object(&session, "/other").is_some());
}

#[test]
fn validator_mismatch_stales_only_its_observation_and_never_retries() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_object(&mut client, &session, "/item", object("item", None, b"v1"));
    observe_interface(
        &mut client,
        &session,
        FetchInterfaceResponse {
            interface: "math".into(),
            descriptor: descriptor(TypeExpr::Integer),
            validator: Some(b"i1".to_vec()),
        },
    );
    let call = client
        .prepare_call(&session, "/item", "math", "calculate", args(1))
        .unwrap();
    client.mark_dispatched(call.id).unwrap();
    let error = ProtocolError {
        code: ProtocolErrorCode::InterfaceValidatorMismatch,
        message: "stale descriptor".into(),
    };
    let response =
        encode_protocol_error_response(ProtocolInteraction::CallOperation, &error, wire_limits())
            .unwrap();
    client.complete(call.id, response).unwrap();

    assert_eq!(
        client.interface(&session, "math").unwrap().state,
        ObservationState::Stale
    );
    assert_eq!(
        client.object(&session, "/item").unwrap().state,
        ObservationState::Fresh
    );
    assert_eq!(client.in_flight_len(), 0);
    assert_eq!(client.call_history(&session).unwrap().len(), 1);

    // Reobservation changes cache state but does not manufacture a replacement call.
    observe_interface(
        &mut client,
        &session,
        FetchInterfaceResponse {
            interface: "math".into(),
            descriptor: descriptor(TypeExpr::Integer),
            validator: Some(b"i2".to_vec()),
        },
    );
    assert_eq!(client.in_flight_len(), 0);
    assert_eq!(client.call_history(&session).unwrap().len(), 1);
}

#[test]
fn protocol_and_transport_unknown_outcomes_stale_without_retry() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_object(&mut client, &session, "/item", object("item", None, b"v1"));
    observe_interface(
        &mut client,
        &session,
        FetchInterfaceResponse {
            interface: "math".into(),
            descriptor: descriptor(TypeExpr::Integer),
            validator: None,
        },
    );

    let call = client
        .prepare_call(&session, "/item", "math", "calculate", args(1))
        .unwrap();
    client.mark_dispatched(call.id).unwrap();
    let unknown = ProtocolError {
        code: ProtocolErrorCode::OperationOutcomeUnknown,
        message: "commit uncertain".into(),
    };
    let completion = client
        .complete(
            call.id,
            encode_protocol_error_response(
                ProtocolInteraction::CallOperation,
                &unknown,
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();
    assert!(matches!(
        completion,
        Completion::Call(record)
            if matches!(
                record.outcome,
                CallOutcome::Unknown {
                    reason: OutcomeUnknownReason::Protocol,
                    ..
                }
            )
    ));
    assert_eq!(
        client.object(&session, "/item").unwrap().state,
        ObservationState::Stale
    );

    observe_object(&mut client, &session, "/item", object("item", None, b"v2"));
    let second = client
        .prepare_call(&session, "/item", "math", "calculate", args(2))
        .unwrap();
    client.mark_dispatched(second.id).unwrap();
    let completion = client
        .fail_dispatched_call(
            second.id,
            DispatchedTransportFailure::Timeout,
            "deadline elapsed",
        )
        .unwrap();
    assert!(matches!(
        completion,
        Completion::Call(record)
            if matches!(
                record.outcome,
                CallOutcome::Unknown {
                    reason: OutcomeUnknownReason::Timeout,
                    ..
                }
            )
    ));
    assert_eq!(client.in_flight_len(), 0);
    assert_eq!(client.call_history(&session).unwrap().len(), 2);
}

#[test]
fn overlapping_tree_work_preserves_newer_subject_request_ownership() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    let tree_request = FetchTreeRequest {
        path: "/".into(),
        depth: 1,
        limit: 4,
    };
    let tree_value = FetchTreeResponse {
        root: ObjectTreeNode {
            object: object("", None, b"root"),
            children: vec![ObjectTreeNode {
                object: object("child", None, b"tree-child"),
                children: vec![],
            }],
        },
        truncated: false,
    };

    let old_tree = client.prepare_tree(&session, "/", 1, 4).unwrap();
    let newer_fetch = client.prepare_fetch(&session, "/child").unwrap();
    client
        .complete(
            old_tree.id,
            encode_fetch_tree_response(&tree_request, &tree_value, wire_limits()).unwrap(),
        )
        .unwrap();
    assert!(matches!(
        client.object(&session, "/child").unwrap().state,
        ObservationState::Loading(id) if id == newer_fetch.id
    ));
    client
        .complete(
            newer_fetch.id,
            fetch_response("/child", &object("child", None, b"newer-fetch")),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/child").unwrap().validator,
        Some(b"newer-fetch".to_vec())
    );

    let older_tree = client.prepare_tree(&session, "/", 1, 4).unwrap();
    let completed_newer_fetch = client.prepare_fetch(&session, "/child").unwrap();
    client
        .complete(
            completed_newer_fetch.id,
            fetch_response("/child", &object("child", None, b"completed-newer")),
        )
        .unwrap();
    client
        .complete(
            older_tree.id,
            encode_fetch_tree_response(&tree_request, &tree_value, wire_limits()).unwrap(),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/child").unwrap().validator,
        Some(b"completed-newer".to_vec())
    );

    let older_omitting_tree = client.prepare_tree(&session, "/", 1, 4).unwrap();
    let newer_fetch = client.prepare_fetch(&session, "/child").unwrap();
    client
        .complete(
            newer_fetch.id,
            fetch_response("/child", &object("child", None, b"newer-than-omission")),
        )
        .unwrap();
    client
        .complete(
            older_omitting_tree.id,
            encode_fetch_tree_response(
                &tree_request,
                &FetchTreeResponse {
                    root: ObjectTreeNode {
                        object: object("", None, b"older-root"),
                        children: vec![],
                    },
                    truncated: false,
                },
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();
    let child = client.object(&session, "/child").unwrap();
    assert_eq!(child.state, ObservationState::Fresh);
    assert_eq!(child.validator, Some(b"newer-than-omission".to_vec()));

    let old_tree = client.prepare_tree(&session, "/", 1, 4).unwrap();
    let newer_child_tree = client.prepare_tree(&session, "/child", 0, 1).unwrap();
    client
        .complete(
            old_tree.id,
            encode_fetch_tree_response(&tree_request, &tree_value, wire_limits()).unwrap(),
        )
        .unwrap();
    assert!(matches!(
        client.tree(&session, "/child").unwrap().state,
        ObservationState::Loading(id) if id == newer_child_tree.id
    ));
    let child_observation = client.object(&session, "/child").unwrap();
    assert!(matches!(
        child_observation.state,
        ObservationState::Loading(id) if id == newer_child_tree.id
    ));
    assert_eq!(
        child_observation.validator,
        Some(b"newer-than-omission".to_vec())
    );
    let child_request = FetchTreeRequest {
        path: "/child".into(),
        depth: 0,
        limit: 1,
    };
    let child_value = FetchTreeResponse {
        root: ObjectTreeNode {
            object: object("child", None, b"newer-tree"),
            children: vec![],
        },
        truncated: false,
    };
    client
        .complete(
            newer_child_tree.id,
            encode_fetch_tree_response(&child_request, &child_value, wire_limits()).unwrap(),
        )
        .unwrap();
    assert_eq!(
        client.tree(&session, "/child").unwrap().state,
        ObservationState::Fresh
    );

    let older_parent = client.prepare_tree(&session, "/", 1, 4).unwrap();
    let newer_child = client.prepare_tree(&session, "/child", 0, 1).unwrap();
    client
        .complete(
            newer_child.id,
            encode_fetch_tree_response(
                &child_request,
                &FetchTreeResponse {
                    root: ObjectTreeNode {
                        object: object("child", None, b"completed-newer-tree"),
                        children: vec![],
                    },
                    truncated: false,
                },
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();
    client
        .complete(
            older_parent.id,
            encode_fetch_tree_response(&tree_request, &tree_value, wire_limits()).unwrap(),
        )
        .unwrap();
    assert_eq!(
        client.object(&session, "/child").unwrap().validator,
        Some(b"completed-newer-tree".to_vec())
    );
    assert_eq!(
        client.tree(&session, "/child").unwrap().revision,
        newer_child.id
    );
}

#[test]
fn superseded_transport_failure_does_not_poison_newer_refresh() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    let old = client.prepare_fetch(&session, "/item").unwrap();
    let newer = client.prepare_fetch(&session, "/item").unwrap();
    assert_eq!(
        client
            .fail_transport(old.id, "old connection failed")
            .unwrap(),
        Completion::StaleResponseRejected(old.id)
    );
    assert!(matches!(
        client.object(&session, "/item").unwrap().state,
        ObservationState::Loading(id) if id == newer.id
    ));
    client
        .complete(
            newer.id,
            fetch_response("/item", &object("item", None, b"new")),
        )
        .unwrap();

    let old = client.prepare_tree(&session, "/", 0, 1).unwrap();
    let newer_tree = client.prepare_tree(&session, "/", 0, 1).unwrap();
    assert_eq!(
        client
            .fail_transport(old.id, "old tree connection failed")
            .unwrap(),
        Completion::StaleResponseRejected(old.id)
    );
    assert!(matches!(
        client.tree(&session, "/").unwrap().state,
        ObservationState::Loading(id) if id == newer_tree.id
    ));
    client
        .complete(
            newer_tree.id,
            encode_fetch_tree_response(
                &FetchTreeRequest {
                    path: "/".into(),
                    depth: 0,
                    limit: 1,
                },
                &FetchTreeResponse {
                    root: ObjectTreeNode {
                        object: object("", None, b"root"),
                        children: vec![],
                    },
                    truncated: false,
                },
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();

    let old = client.prepare_interface(&session, "math").unwrap();
    let newer = client.prepare_interface(&session, "math").unwrap();
    assert_eq!(
        client
            .fail_transport(old.id, "old connection failed")
            .unwrap(),
        Completion::StaleResponseRejected(old.id)
    );
    assert!(matches!(
        client.interface(&session, "math").unwrap().state,
        ObservationState::Loading(id) if id == newer.id
    ));
    let value = FetchInterfaceResponse {
        interface: "math".into(),
        descriptor: descriptor(TypeExpr::Integer),
        validator: Some(b"current".to_vec()),
    };
    client
        .complete(
            newer.id,
            encode_fetch_interface_response(
                &FetchInterfaceRequest {
                    interface: "math".into(),
                },
                &value,
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();
}

#[test]
fn validator_free_interface_observation_remains_immutable() {
    let mut client = client(8, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_interface(
        &mut client,
        &session,
        FetchInterfaceResponse {
            interface: "math".into(),
            descriptor: descriptor(TypeExpr::Integer),
            validator: None,
        },
    );

    let refresh = client.prepare_interface(&session, "math").unwrap();
    let contradictory = FetchInterfaceResponse {
        interface: "math".into(),
        descriptor: descriptor(TypeExpr::Integer),
        validator: Some(b"became-mutable".to_vec()),
    };
    let error = client
        .complete(
            refresh.id,
            encode_fetch_interface_response(
                &FetchInterfaceRequest {
                    interface: "math".into(),
                },
                &contradictory,
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap_err();
    assert_eq!(error.kind, ClientErrorKind::InvalidResponse);
    assert_eq!(client.interface(&session, "math").unwrap().validator, None);
}

#[test]
fn entry_eviction_never_discards_a_loading_tree_subject() {
    let mut client = client(1, 8, 8, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    observe_object(&mut client, &session, "/", object("", None, b"root"));
    let tree = client.prepare_tree(&session, "/", 0, 1).unwrap();
    assert_eq!(
        client.prepare_fetch(&session, "/other").unwrap_err().kind,
        ClientErrorKind::Capacity
    );
    assert!(matches!(
        client.tree(&session, "/").unwrap().state,
        ObservationState::Loading(id) if id == tree.id
    ));
    client
        .complete(
            tree.id,
            encode_fetch_tree_response(
                &FetchTreeRequest {
                    path: "/".into(),
                    depth: 0,
                    limit: 1,
                },
                &FetchTreeResponse {
                    root: ObjectTreeNode {
                        object: object("", None, b"tree"),
                        children: vec![],
                    },
                    truncated: false,
                },
                wire_limits(),
            )
            .unwrap(),
        )
        .unwrap();
    assert_eq!(
        client.tree(&session, "/").unwrap().state,
        ObservationState::Fresh
    );
}

#[test]
fn failed_entry_and_tree_observation_metadata_is_bounded() {
    let mut client = client(2, 8, 2, 8);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    for path in ["/one", "/two", "/three"] {
        let request = client.prepare_fetch(&session, path).unwrap();
        assert_eq!(
            client
                .fail_transport(request.id, "retrieval failed")
                .unwrap_err()
                .kind,
            ClientErrorKind::Transport
        );
    }
    assert!(client.object(&session, "/one").is_none());
    assert!(client.object(&session, "/two").is_some());
    assert!(client.object(&session, "/three").is_some());

    for path in ["/alpha", "/beta", "/gamma"] {
        let request = client.prepare_tree(&session, path, 0, 1).unwrap();
        assert_eq!(
            client
                .fail_transport(request.id, "tree failed")
                .unwrap_err()
                .kind,
            ClientErrorKind::Transport
        );
    }
    assert!(client.tree(&session, "/alpha").is_none());
    assert!(client.tree(&session, "/gamma").is_some());
}

#[test]
fn in_flight_and_history_are_bounded() {
    let mut client = client(8, 8, 1, 1);
    let session = client
        .open_session("https://example.test/wip", SecurityContext::new("subject"))
        .unwrap();
    let first = client.prepare_fetch(&session, "/one").unwrap();
    let error = client.prepare_fetch(&session, "/two").unwrap_err();
    assert_eq!(error.kind, ClientErrorKind::Capacity);
    client
        .complete(first.id, fetch_response("/one", &object("one", None, b"1")))
        .unwrap();

    // Diagnostic history evicts oldest entries at the configured bound.
    let stale_old = client.prepare_fetch(&session, "/one").unwrap();
    let stale_new = client.prepare_fetch(&session, "/one").unwrap_err();
    assert_eq!(stale_new.kind, ClientErrorKind::Capacity);
    client
        .complete(
            stale_old.id,
            Response::builder()
                .status(200)
                .header("content-type", "application/json")
                .body(b"not json".to_vec())
                .unwrap(),
        )
        .unwrap_err();
    assert_eq!(client.diagnostics(&session).unwrap().len(), 1);
}