use oa_core::{Command, Id, Kind, Layer, Project, RelationKind};
use oa_server::{
    ServerConfig, app,
    store::{Store, StoredProject},
};
use reqwest::{Client, StatusCode};
use serde_json::{Value, json};
use std::{path::Path, sync::Arc};
use tokio::{sync::oneshot, task::JoinHandle};

#[path = "http/communication_limits.rs"]
mod communication_limits;
#[path = "http/data_contracts.rs"]
mod data_contracts;
#[path = "http/port_anchors.rs"]
mod port_anchors;
#[path = "http/semantic_transitions.rs"]
mod semantic_transitions;
#[path = "http/source6_views.rs"]
mod source6_views;
#[path = "http/source7_contracts.rs"]
mod source7_contracts;
#[path = "http/source_projects.rs"]
mod source_projects;

fn v2_missing_ports() -> Value {
    let mut old: Value = serde_json::from_slice(include_bytes!(
        "../../oa-core/tests/fixtures/v2-coastal-rescue.json"
    ))
    .unwrap();
    for relation in old["relations"].as_object_mut().unwrap().values_mut() {
        if let Some(exchange) = relation["body"].get_mut("FunctionalExchange") {
            exchange["source_port"] = Value::Null;
            exchange["target_port"] = Value::Null;
        }
    }
    old
}

struct TestServer {
    base: String,
    client: Client,
    stop: oneshot::Sender<()>,
    task: JoinHandle<()>,
}

impl TestServer {
    async fn start(directory: &Path, token: Option<String>) -> Self {
        let directory = directory.to_owned();
        let store = Arc::new(
            tokio::task::spawn_blocking(move || Store::open(directory))
                .await
                .unwrap()
                .unwrap(),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = format!("http://{}", listener.local_addr().unwrap());
        let (stop, done) = oneshot::channel();
        let router = app(store, token);
        let task = tokio::spawn(async move {
            axum::serve(listener, router)
                .with_graceful_shutdown(async {
                    let _ = done.await;
                })
                .await
                .unwrap();
        });
        Self {
            base,
            client: Client::new(),
            stop,
            task,
        }
    }

    fn url(&self, path: &str) -> String {
        format!("{}/api/v1{path}", self.base)
    }

    async fn create(&self) -> (Id, String) {
        let response = self
            .client
            .post(self.url("/projects"))
            .json(&json!({"name":"Test mission"}))
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::CREATED);
        let etag = response.headers()["etag"].to_str().unwrap().to_owned();
        let summary: Value = response.json().await.unwrap();
        (
            summary["resource_id"].as_str().unwrap().parse().unwrap(),
            etag,
        )
    }

    async fn import(&self, project: &Project) -> (Id, String) {
        let response = self
            .client
            .post(self.url("/projects/import"))
            .json(project)
            .send()
            .await
            .unwrap();
        let status = response.status();
        let etag = response
            .headers()
            .get("etag")
            .map(|value| value.to_str().unwrap().to_owned());
        let body: Value = response.json().await.unwrap();
        assert_eq!(status, StatusCode::CREATED, "{body}");
        (
            body["resource_id"].as_str().unwrap().parse().unwrap(),
            etag.unwrap(),
        )
    }

    async fn get(&self, id: Id) -> StoredProject {
        self.client
            .get(self.url(&format!("/projects/{id}")))
            .send()
            .await
            .unwrap()
            .error_for_status()
            .unwrap()
            .json()
            .await
            .unwrap()
    }

    async fn stop(self) {
        let _ = self.stop.send(());
        drop(self.client);
        tokio::time::timeout(std::time::Duration::from_secs(5), self.task)
            .await
            .unwrap()
            .unwrap();
    }
}

fn graph() -> Project {
    let mut model = Project::empty();
    model.name = "Harbor assistance".into();
    let actor = model
        .add_element(Kind::Actor, Layer::Oa, "Vessel", None)
        .unwrap();
    let first = model
        .add_element(Kind::Activity, Layer::Oa, "Report distress", None)
        .unwrap();
    let second = model
        .add_element(Kind::Activity, Layer::Oa, "Assess distress", None)
        .unwrap();
    model.relate(RelationKind::Allocation, "Reports", actor, first);
    model.relate(RelationKind::Interaction, "Distress signal", first, second);
    model
}

#[tokio::test]
async fn edge_route_transactions_source_and_json_share_defaults_etags_and_atomic_rejection() {
    let model = oa_core::language::parse(
        r#"arcadia 3; project routing "Routing" {
        oa_activity a "Report" {}
        oa_activity b "Assess" {}
        interaction signal "Signal" { source @a; target @b; }
        diagram architecture "Architecture" {}
        diagram review "Review" {}
    }"#,
    )
    .unwrap();
    let find = |alias: &str| {
        model
            .identities()
            .into_iter()
            .find(|(_, candidate)| *candidate == alias)
            .unwrap()
            .0
    };
    let diagram = find("architecture");
    let other = find("review");
    let edge = find("signal");
    let element = find("a");
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let (id, etag) = server.import(&model).await;
    let url = server.url(&format!("/projects/{id}/transactions"));
    // The second route intentionally omits both optional fields, exercising
    // the public JSON defaults rather than serializing a pre-hydrated struct.
    let commands = json!([
        {"SetEdgeRoute":{"diagram":diagram,"route":{
            "edge":edge,"waypoints":[{"x":420.0,"y":260.0},{"x":690.0,"y":260.0}],
            "label":{"fraction":0.55,"offset":{"x":0.0,"y":-16.0}}
        }}},
        {"SetEdgeRoute":{"diagram":other,"route":{"edge":edge}}}
    ]);
    let typed: Vec<Command> = serde_json::from_value(commands.clone()).unwrap();
    let expected = oa_core::execute_batch(&model, &typed).unwrap();
    let preview = server
        .client
        .post(&url)
        .header("if-match", &etag)
        .json(&json!({"commands":commands,"dry_run":true}))
        .send()
        .await
        .unwrap();
    assert_eq!(preview.status(), StatusCode::OK);
    let preview: Value = preview.json().await.unwrap();
    assert_eq!(
        serde_json::from_value::<Project>(preview["model"].clone()).unwrap(),
        expected
    );
    assert_eq!(server.get(id).await.model, model);

    let commit = server
        .client
        .post(&url)
        .header("if-match", &etag)
        .json(&json!({"commands":commands}))
        .send()
        .await
        .unwrap();
    assert_eq!(commit.status(), StatusCode::OK);
    let new_etag = commit.headers()["etag"].to_str().unwrap().to_owned();
    let stored = server.get(id).await;
    assert_eq!(stored.revision, 2);
    assert_eq!(stored.model, expected);
    assert!(stored.model.diagrams[&other].routes[0].waypoints.is_empty());
    assert!(stored.model.diagrams[&other].routes[0].label.is_none());
    let source = server
        .client
        .get(server.url(&format!("/projects/{id}/source")))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(source.contains("edge @signal;"));
    assert_eq!(oa_core::language::parse(&source).unwrap(), expected);
    let imported = server
        .client
        .post(server.url("/projects/import"))
        .header("content-type", "text/plain")
        .body(source.clone())
        .send()
        .await
        .unwrap();
    assert_eq!(imported.status(), StatusCode::CREATED);
    let imported: Value = imported.json().await.unwrap();
    assert_eq!(
        server
            .get(imported["resource_id"].as_str().unwrap().parse().unwrap())
            .await
            .model,
        expected
    );

    let reset = json!({"RemoveEdgeRoute":{"diagram":diagram,"edge":edge}});
    for stale_command in [commands[0].clone(), reset.clone()] {
        let response = server
            .client
            .post(&url)
            .header("if-match", &etag)
            .json(&json!({"commands":[stale_command]}))
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::PRECONDITION_FAILED);
    }
    let no_op = server
        .client
        .post(&url)
        .header("if-match", &new_etag)
        .json(&json!({"commands":commands}))
        .send()
        .await
        .unwrap();
    assert_eq!(no_op.status(), StatusCode::OK);
    assert_eq!(no_op.headers()["etag"], new_etag);
    assert_eq!(no_op.json::<Value>().await.unwrap()["changed"], false);

    let invalid_routes = [
        (json!({"edge":element}), StatusCode::UNPROCESSABLE_ENTITY),
        (
            json!({"edge":edge,"waypoints":vec![json!({"x":0,"y":0});129]}),
            StatusCode::UNPROCESSABLE_ENTITY,
        ),
        (
            json!({"edge":edge,"waypoints":[{"x":1000001,"y":0}]}),
            StatusCode::UNPROCESSABLE_ENTITY,
        ),
        (
            json!({"edge":edge,"label":{"fraction":2,"offset":{"x":0,"y":0}}}),
            StatusCode::UNPROCESSABLE_ENTITY,
        ),
        (
            json!({"edge":edge,"label":{"fraction":0.5}}),
            StatusCode::BAD_REQUEST,
        ),
        (json!({"edge":edge,"unknown":true}), StatusCode::BAD_REQUEST),
    ];
    for (route, status) in invalid_routes {
        let response = server
            .client
            .post(&url)
            .header("if-match", &new_etag)
            .json(&json!({"commands":[
                Command::Rename(element, "Must roll back".into()),
                {"SetEdgeRoute":{"diagram":diagram,"route":route}}
            ]}))
            .send()
            .await
            .unwrap();
        assert_eq!(
            response.status(),
            status,
            "{route}: {}",
            response.text().await.unwrap()
        );
    }
    let mut duplicated = expected.clone();
    let route = duplicated.diagrams[&diagram].routes[0].clone();
    duplicated
        .diagrams
        .get_mut(&diagram)
        .unwrap()
        .routes
        .push(route);
    let rejected = server
        .client
        .put(server.url(&format!("/projects/{id}/model")))
        .header("if-match", &new_etag)
        .json(&duplicated)
        .send()
        .await
        .unwrap();
    assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(server.get(id).await.model, stored.model);
    assert_eq!(server.get(id).await.revision, stored.revision);

    // The source write path hydrates the same metadata, and an equal source
    // application must not create a revision or erase an explicit empty route.
    let source_no_op = server
        .client
        .put(server.url(&format!("/projects/{id}/source")))
        .header("if-match", &new_etag)
        .header("content-type", "text/plain")
        .body(source)
        .send()
        .await
        .unwrap();
    assert_eq!(source_no_op.status(), StatusCode::OK);
    assert_eq!(source_no_op.headers()["etag"], new_etag);

    let removed = server
        .client
        .post(&url)
        .header("if-match", &new_etag)
        .json(&json!({"commands":[reset]}))
        .send()
        .await
        .unwrap();
    assert_eq!(removed.status(), StatusCode::OK);
    let removed_etag = removed.headers()["etag"].to_str().unwrap().to_owned();
    let remaining = server.get(id).await;
    assert!(remaining.model.diagrams[&diagram].routes.is_empty());
    assert_eq!(remaining.model.diagrams[&other], expected.diagrams[&other]);
    let removed_again = server
        .client
        .post(&url)
        .header("if-match", &removed_etag)
        .json(&json!({"commands":[reset]}))
        .send()
        .await
        .unwrap();
    assert_eq!(removed_again.status(), StatusCode::OK);
    assert_eq!(removed_again.headers()["etag"], removed_etag);
    assert_eq!(
        removed_again.json::<Value>().await.unwrap()["changed"],
        false
    );
    server.stop().await;
    let reopened = TestServer::start(directory.path(), None).await;
    assert_eq!(reopened.get(id).await.model, remaining.model);
    assert_eq!(reopened.get(id).await.revision, remaining.revision);
    reopened.stop().await;
}

#[test]
fn route_openapi_describes_the_serde_command_shapes_and_nested_defaults() {
    let spec: Value = serde_json::from_str(include_str!("../../../docs/openapi.json")).unwrap();
    let schemas = &spec["components"]["schemas"];
    assert_eq!(
        schemas["Diagram"]["properties"]["routes"]["default"],
        json!([])
    );
    assert!(
        !schemas["Diagram"]["required"]
            .as_array()
            .unwrap()
            .contains(&json!("routes"))
    );
    assert_eq!(schemas["EdgeRoute"]["required"], json!(["edge"]));
    assert_eq!(
        schemas["EdgeRoute"]["properties"]["waypoints"]["maxItems"],
        oa_core::MAX_ROUTE_POINTS
    );
    assert_eq!(
        schemas["EdgeRoute"]["properties"]["waypoints"]["default"],
        json!([])
    );
    assert_eq!(
        schemas["EdgeLabel"]["required"],
        json!(["fraction", "offset"])
    );
    assert_eq!(schemas["EdgeLabel"]["properties"]["fraction"]["minimum"], 0);
    assert_eq!(schemas["EdgeLabel"]["properties"]["fraction"]["maximum"], 1);
    let commands = schemas["Command"]["oneOf"].as_array().unwrap();
    for (name, fields) in [
        ("SetEdgeRoute", json!(["diagram", "route"])),
        ("RemoveEdgeRoute", json!(["diagram", "edge"])),
    ] {
        let schema = commands
            .iter()
            .find(|schema| schema["properties"].get(name).is_some())
            .unwrap();
        assert_eq!(schema["required"], json!([name]));
        assert_eq!(schema["properties"][name]["required"], fields);
        assert_eq!(schema["properties"][name]["additionalProperties"], false);
    }
}

#[tokio::test]
async fn oversized_source_and_json_are_rejected_without_stalling_the_catalog() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let (existing, etag) = server.create().await;
    let before = server.get(existing).await;
    let mut source =
        String::from("arcadia 2; project oversized \"Oversized\" { oa_activity a0 \"Root\" {}\n");
    let mut model = Project::empty();
    for i in 0..12_000u128 {
        if i > 0 {
            source.push_str(&format!(
                "oa_activity a{i} \"Activity {i}\" {{ parent @a{}; }}\n",
                i - 1
            ));
        }
        let id = Id::from_u128(i + 1);
        model.elements.insert(
            id,
            oa_core::Element {
                id,
                alias: format!("a{i}"),
                name: format!("Activity {i}"),
                description: String::new(),
                parent: (i > 0).then(|| Id::from_u128(i)),
                body: oa_core::ElementBody::Oa(oa_core::OaBody::Activity),
            },
        );
    }
    source.push('}');
    let native_json = serde_json::to_vec(&model).unwrap();
    let mut legacy_elements = serde_json::Map::new();
    for (id, e) in &model.elements {
        legacy_elements.insert(id.to_string(), json!({"id":id,"kind":"Activity","name":e.name,"description":"","position":{"x":0,"y":0},"steps":[]}));
    }
    let legacy_json = serde_json::to_vec(&json!({"schema_version":1,"id":model.id,"name":"Oversized","elements":legacy_elements,"relations":{}})).unwrap();
    for (media, body) in [
        ("text/plain", source.as_bytes().to_vec()),
        ("application/json", native_json.clone()),
        ("application/json", legacy_json),
    ] {
        let client = server.client.clone();
        let url = server.url("/projects/import");
        let upload = tokio::spawn(async move {
            client
                .post(url)
                .header("content-type", media)
                .body(body)
                .send()
                .await
                .unwrap()
        });
        // The large request is real transport traffic. This deliberately generous
        // liveness bound catches the former multi-second quadratic validation;
        // exact wall times belong to the separate release measurement record.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        let read = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            server.client.get(server.url("/projects")).send(),
        )
        .await
        .expect("catalog stalled by rejected import")
        .unwrap();
        assert_eq!(read.status(), StatusCode::OK);
        // Completion includes transport/decoding while other tests may saturate
        // the two-core runner. Catalog responsiveness is asserted separately
        // above; this watchdog only prevents a hung request from hanging CI.
        let rejected = tokio::time::timeout(std::time::Duration::from_secs(30), upload)
            .await
            .expect("oversized import did not complete within the test watchdog")
            .unwrap();
        assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
        assert_eq!(
            rejected.json::<Value>().await.unwrap()["code"],
            "model_limit"
        );
        assert_eq!(server.get(existing).await.model, before.model);
    }
    // The same budgets apply to replacement and non-mutating candidate checks;
    // neither endpoint may run expensive structural validation before admission.
    // Their completion watchdog is not a catalog latency measurement: real
    // JSON encoding/decoding can contend with accepted deep-graph preparation.
    for (suffix, media, body) in [
        ("source", "text/plain", source.as_bytes().to_vec()),
        ("model", "application/json", native_json),
    ] {
        let rejected = tokio::time::timeout(
            std::time::Duration::from_secs(30),
            server
                .client
                .put(server.url(&format!("/projects/{existing}/{suffix}")))
                .header("if-match", &etag)
                .header("content-type", media)
                .body(body)
                .send(),
        )
        .await
        .unwrap()
        .unwrap();
        assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
        assert_eq!(
            rejected.json::<Value>().await.unwrap()["code"],
            "model_limit"
        );
    }
    for candidate in [
        json!({"format":"arcadia", "source":source}),
        json!({"format":"json", "model":model}),
    ] {
        let checked = tokio::time::timeout(
            std::time::Duration::from_secs(30),
            server
                .client
                .post(server.url(&format!("/projects/{existing}/validation")))
                .json(&candidate)
                .send(),
        )
        .await
        .unwrap()
        .unwrap();
        assert_eq!(checked.status(), StatusCode::OK);
        let checked: Value = checked.json().await.unwrap();
        assert_eq!(checked["valid"], false);
        assert_eq!(checked["issues"][0]["code"], "model_limit");
    }
    assert_eq!(server.get(existing).await.model, before.model);
    assert_eq!(server.get(existing).await.revision, before.revision);
    let catalog: Value = server
        .client
        .get(server.url("/projects"))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(catalog["items"].as_array().unwrap().len(), 1);
    server.stop().await;
}

#[tokio::test]
async fn mission_and_contextual_chain_records_round_trip_and_edit_through_http() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let source = r#"arcadia 2; project release_example "Release modeling" {
        oa_entity team "Rescue team" {}
        oa_mission need "Protect mariners" {}
        oa_capability required "Assist mariners" {}
        oa_capability routine "Routine assistance" {}
        oa_capability extended "Extended assistance" {}
        mission_involvement participant "Mission participant" { source @team; target @need; }
        mission_capability mission_support "Mission support" { source @need; target @required; }
        capability_include included "Required assistance" { source @required; target @routine; }
        capability_extend extra "Conditional assistance" { source @extended; target @required; condition "sea_state > 4"; }
        capability_generalization specialized "Specialized assistance" { source @extended; target @routine; }
        sa_system system_of_interest "Rescue support" {}
        sa_mission response "Support rescue" {}
        trace mission_mapping "Mission realization" { oa @need; sa @response; kind Mission; }
        sa_function assess "Assess conditions" {}
        sa_function notify_operator "Notify operator" {}
        sa_functional_chain routine_chain "Routine chain" { steps [@routine_action]; }
        sa_occurrence routine_action "Routine notification" { parent @routine_chain; body { Action { action @notify_operator; lifeline none; }; }; }
        sa_functional_chain chain "Rescue decision" { steps [@entry, @choice, @routine_path, @extended_path, @merge, @reused, @finish]; }
        sa_occurrence entry "Assess" { parent @chain; body { Action { action @assess; lifeline none; }; }; }
        sa_occurrence choice "Select response" { parent @chain; body { Control { kind Choice; }; }; }
        sa_occurrence routine_path "Routine response" { parent @chain; body { Action { action @notify_operator; lifeline none; }; }; }
        sa_occurrence extended_path "Review conditions" { parent @chain; body { Action { action @assess; lifeline none; }; }; }
        sa_occurrence merge "Continue response" { parent @chain; body { Control { kind Merge; }; }; }
        sa_occurrence reused "Reused notification" { parent @chain; body { SequenceReference { sequence @routine_chain; }; }; }
        sa_occurrence finish "Finish" { parent @chain; body { Action { action @notify_operator; lifeline none; }; }; }
        sequence_link start_choice "Choose" { source @entry; target @choice; }
        sequence_link ordinary "Ordinary path" { source @choice; target @routine_path; guard "sea_state <= 4"; }
        sequence_link exceptional "Extended path" { source @choice; target @extended_path; guard "else"; }
        sequence_link first_merge "Continue" { source @routine_path; target @merge; }
        sequence_link second_merge "Continue" { source @extended_path; target @merge; }
        sequence_link call_routine "Reuse notification" { source @merge; target @reused; }
        sequence_link conclude "Conclude" { source @reused; target @finish; }
    }"#;
    let imported = server
        .client
        .post(server.url("/projects/import"))
        .header("content-type", "text/plain")
        .body(source)
        .send()
        .await
        .unwrap();
    assert_eq!(imported.status(), StatusCode::CREATED);
    let etag = imported.headers()["etag"].to_str().unwrap().to_owned();
    let id: Id = imported.json::<Value>().await.unwrap()["resource_id"]
        .as_str()
        .unwrap()
        .parse()
        .unwrap();
    let baseline = server.get(id).await;
    let mut changed = baseline
        .model
        .relations
        .values()
        .find(|r| r.alias == "ordinary")
        .unwrap()
        .clone();
    if let oa_core::RelationBody::SequenceLink { guard, .. } = &mut changed.body {
        *guard = "sea_state <= 4 && operator == \"Ready\"".into();
    } else {
        panic!("wrong persisted relation body");
    }
    let commit = server
        .client
        .post(server.url(&format!("/projects/{id}/transactions")))
        .header("if-match", etag)
        .json(&json!({"commands":[Command::EditRelation(changed.id, changed.clone())]}))
        .send()
        .await
        .unwrap();
    assert_eq!(commit.status(), StatusCode::OK);
    let current = server.get(id).await;
    assert_eq!(current.revision, 2);
    assert_eq!(current.model.relations[&changed.id], changed);
    let exported = server
        .client
        .get(server.url(&format!("/projects/{id}/source")))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(exported.contains("sequence @routine_chain;"));
    assert_eq!(oa_core::language::parse(&exported).unwrap(), current.model);
    let filtered: Value = server
        .client
        .get(server.url(&format!("/projects/{id}/relations?kind=SequenceLink")))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(filtered["items"].as_array().unwrap().len(), 7);
    server.stop().await;
    let server = TestServer::start(directory.path(), None).await;
    assert_eq!(server.get(id).await.model, current.model);
    server.stop().await;
}

#[tokio::test]
async fn legacy_v2_scenarios_and_typed_message_edits_use_the_same_http_model() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let legacy_json = include_bytes!("../../oa-core/tests/fixtures/v2-coastal-rescue.json");
    let imported = server
        .client
        .post(server.url("/projects/import"))
        .header("content-type", "application/json")
        .body(legacy_json.to_vec())
        .send()
        .await
        .unwrap();
    assert_eq!(imported.status(), StatusCode::CREATED);
    let legacy_id: Id = imported.json::<Value>().await.unwrap()["resource_id"]
        .as_str()
        .unwrap()
        .parse()
        .unwrap();
    assert_eq!(
        server.get(legacy_id).await.model,
        oa_core::persistence::decode(legacy_json).unwrap()
    );
    let source = r#"arcadia 2; project messages "Synchronous request" {
        sa_system controller "Controller" {}
        sa_function request "Request" {}
        sa_function respond "Respond" {}
        sa_exchange_item payload "Payload" {}
        sa_data_field value_field "Requested value" { parent @payload; }
        functional_exchange request_data "Request data" { source @request; target @respond; items [@payload]; }
        sa_scenario story "Request and response" { viewpoint Functional; steps [@called, @replied]; }
        sa_lifeline requester "Requester" { parent @story; represented @request; }
        sa_lifeline responder "Responder" { parent @story; represented @respond; }
        sa_occurrence called "Call" { parent @story; body { Exchange { exchange @request_data; sender @requester; receiver @responder; message_kind Call; reply_to none; values [{ field @value_field; expression "requested_value"; }]; }; }; }
        sa_occurrence replied "Reply" { parent @story; body { Exchange { exchange @request_data; sender @responder; receiver @requester; receiver @requester; message_kind Reply; reply_to @called; values []; }; }; }
    }"#;
    // Source errors remain atomic even with the richer typed constructors.
    let rejected = server
        .client
        .put(server.url(&format!("/projects/{legacy_id}/source")))
        .header("if-match", server.get(legacy_id).await.etag())
        .header("content-type", "text/plain")
        .body(source)
        .send()
        .await
        .unwrap();
    assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(server.get(legacy_id).await.revision, 1);
    let source = source.replace(
        "receiver @requester; receiver @requester;",
        "receiver @requester;",
    );
    let model = oa_core::language::parse(&source).unwrap();
    let (id, etag) = server.import(&model).await;
    let mut message = model
        .elements
        .values()
        .find(|element| element.alias == "called")
        .unwrap()
        .clone();
    if let Some(oa_core::SharedBody::Occurrence {
        body: oa_core::OccurrenceBody::Exchange { values, .. },
        ..
    }) = message.shared_mut()
    {
        values[0].expression = "requested_value + \"checked\"".into();
    } else {
        panic!("expected call payload");
    }
    let accepted = server
        .client
        .post(server.url(&format!("/projects/{id}/transactions")))
        .header("if-match", etag)
        .json(&json!({"commands":[Command::EditElement(message.id, message.clone())]}))
        .send()
        .await
        .unwrap();
    assert_eq!(accepted.status(), StatusCode::OK);
    let committed = server.get(id).await;
    assert_eq!(committed.model.elements[&message.id], message);
    let exported = server
        .client
        .get(server.url(&format!("/projects/{id}/source")))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(exported.contains("message_kind Call;"));
    assert!(exported.contains("reply_to @called;"));
    assert!(exported.contains("field @value_field;"));
    assert_eq!(
        oa_core::language::parse(&exported).unwrap(),
        committed.model
    );
    server.stop().await;
    let server = TestServer::start(directory.path(), None).await;
    assert_eq!(server.get(id).await.model, committed.model);
    assert_eq!(server.get(legacy_id).await.revision, 1);
    server.stop().await;
}

#[tokio::test]
async fn typed_state_events_batch_reidentification_and_deletion_use_real_http() {
    use oa_core::{EventBody, RelationBody, SharedBody, TimeEventKind, TransitionKind};
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let source = r#"arcadia 2; project recovery "Typed recovery" {
        oa_entity team "Operational team" {}
        oa_activity operational_work "Operational assessment" {}
        oa_event operational_event "Operational change" { body { Change { expression "visibility > 100"; }; }; }
        sa_system controller "Controller" {}
        sa_function assess "Assess recovery" {}
        sa_exchange_item report_data "Report data" {}
        sa_event report "Report arrived" { body { Signal { items [@report_data]; }; }; }
        sa_event deadline "Recovery timeout" { body { Time { kind After; expression "30 s"; }; }; }
        sa_state_machine machine "Recovery" { parent @controller; }
        sa_region region "Main region" { parent @machine; }
        sa_state idle "Idle" { parent @region; }
        sa_state active "Active" { parent @region; }
        state_transition recover "Recover" { source @idle; target @active; triggers [@deadline]; effects [@assess]; trigger ""; guard "operator == \"Ready\""; effect "Record assessment\nНе выполнять автоматически"; }
    }"#;
    let imported = server
        .client
        .post(server.url("/projects/import"))
        .header("content-type", "text/plain")
        .body(source)
        .send()
        .await
        .unwrap();
    assert_eq!(imported.status(), StatusCode::CREATED);
    let first_etag = imported.headers()["etag"].to_str().unwrap().to_owned();
    let id: Id = imported.json::<Value>().await.unwrap()["resource_id"]
        .as_str()
        .unwrap()
        .parse()
        .unwrap();
    let original = server.get(id).await;
    let find = |alias: &str| {
        original
            .model
            .elements
            .values()
            .find(|e| e.alias == alias)
            .unwrap()
            .id
    };
    let deadline = find("deadline");
    let work = find("assess");
    let payload = find("report_data");
    let signal = find("report");
    let mut change = original.model.elements[&deadline].clone();
    change.id = Id::new_v4();
    change.alias = "recovery_changed".into();
    change.name = "Recovery accepted".into();
    *change.shared_mut().unwrap() = SharedBody::Event {
        body: EventBody::Change {
            expression: "pressure > 4 && status == \"Ready\"".into(),
        },
    };
    let mut transition = original
        .model
        .relations
        .values()
        .find(|r| r.alias == "recover")
        .unwrap()
        .clone();
    let RelationBody::StateTransition {
        triggers,
        transition_kind,
        ..
    } = &mut transition.body
    else {
        panic!("expected transition")
    };
    *triggers = vec![deadline, change.id, work];
    *transition_kind = TransitionKind::External;
    let changed_deadline = Id::new_v4();
    let changed_payload = Id::new_v4();
    let commands = vec![
        Command::CreateElement(change.clone()),
        Command::EditRelation(transition.id, transition.clone()),
        Command::Reidentify(deadline, changed_deadline),
        Command::Reidentify(payload, changed_payload),
        Command::Rename(work, "Assess accepted recovery".into()),
    ];
    let url = server.url(&format!("/projects/{id}/transactions"));
    let accepted = server
        .client
        .post(&url)
        .header("if-match", &first_etag)
        .json(&json!({"commands":commands}))
        .send()
        .await
        .unwrap();
    assert_eq!(
        accepted.status(),
        StatusCode::OK,
        "{}",
        accepted.text().await.unwrap()
    );
    let committed = server.get(id).await;
    assert_eq!(committed.revision, 2);
    assert_eq!(
        committed.model,
        oa_core::execute_batch(&original.model, &commands).unwrap()
    );
    assert_eq!(committed.model.elements[&change.id], change);
    assert!(!committed.model.elements.contains_key(&deadline));
    let Some(SharedBody::Event {
        body: EventBody::Time { kind, expression },
    }) = committed.model.elements[&changed_deadline].shared()
    else {
        panic!("expected time event")
    };
    assert_eq!((*kind, expression.as_str()), (TimeEventKind::After, "30 s"));
    let Some(SharedBody::Event {
        body: EventBody::Signal { items },
    }) = committed.model.elements[&signal].shared()
    else {
        panic!("expected signal")
    };
    assert_eq!(items, &[changed_payload]);
    let RelationBody::StateTransition {
        triggers,
        effects,
        guard,
        effect,
        ..
    } = &committed.model.relations[&transition.id].body
    else {
        panic!("expected transition")
    };
    assert_eq!(triggers, &[changed_deadline, change.id, work]);
    assert_eq!(effects, &[work]);
    assert_eq!(guard, "operator == \"Ready\"");
    assert_eq!(effect, "Record assessment\nНе выполнять автоматически");
    let stale = server
        .client
        .post(&url)
        .header("if-match", &first_etag)
        .json(&json!({"commands":[Command::Delete(change.id)]}))
        .send()
        .await
        .unwrap();
    assert_eq!(stale.status(), StatusCode::PRECONDITION_FAILED);
    let mut invalid = committed.model.relations[&transition.id].clone();
    let RelationBody::StateTransition { triggers, .. } = &mut invalid.body else {
        unreachable!()
    };
    *triggers = vec![find("operational_event")];
    let rejected = server.client.post(&url).header("if-match", committed.etag())
        .json(&json!({"commands":[Command::Rename(work, "Must roll back".into()), Command::EditRelation(invalid.id, invalid)]}))
        .send().await.unwrap();
    assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(server.get(id).await.model, committed.model);
    assert_eq!(server.get(id).await.etag(), committed.etag());
    let exported = server
        .client
        .get(server.url(&format!("/projects/{id}/source")))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(exported.contains("triggers [@deadline, @recovery_changed, @assess];"));
    assert!(exported.contains("effects [@assess];"));
    assert_eq!(
        oa_core::language::parse(&exported).unwrap(),
        committed.model
    );
    let events: Value = server
        .client
        .get(server.url(&format!("/projects/{id}/elements?kind=Event")))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(events["items"].as_array().unwrap().len(), 4);
    let deleted = server.client.post(&url).header("if-match", committed.etag())
        .json(&json!({"commands":[Command::Delete(changed_deadline),Command::Delete(change.id),Command::Delete(work),Command::Delete(changed_payload)]}))
        .send().await.unwrap();
    assert_eq!(
        deleted.status(),
        StatusCode::OK,
        "{}",
        deleted.text().await.unwrap()
    );
    let remaining = server.get(id).await;
    assert_eq!(remaining.revision, 3);
    let RelationBody::StateTransition {
        triggers,
        effects,
        trigger,
        guard: retained_guard,
        effect: retained_effect,
        ..
    } = &remaining.model.relations[&transition.id].body
    else {
        panic!("expected transition")
    };
    assert!(triggers.is_empty() && effects.is_empty());
    assert!(
        !trigger.trim().is_empty(),
        "deleting the last trigger must not introduce completion semantics"
    );
    assert_eq!((retained_guard, retained_effect), (guard, effect));
    let Some(SharedBody::Event {
        body: EventBody::Signal { items },
    }) = remaining.model.elements[&signal].shared()
    else {
        panic!("expected signal")
    };
    assert!(items.is_empty());
    assert_eq!(
        oa_core::language::parse(&oa_core::language::format(&remaining.model)).unwrap(),
        remaining.model
    );
    server.stop().await;
    let server = TestServer::start(directory.path(), None).await;
    assert_eq!(server.get(id).await.model, remaining.model);
    assert_eq!(server.get(id).await.etag(), remaining.etag());
    server.stop().await;
}

#[tokio::test]
async fn conditional_writes_are_atomic_and_survive_restart() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let (id, etag) = server.create().await;
    let url = server.url(&format!("/projects/{id}"));
    let a = server
        .client
        .patch(&url)
        .header("if-match", &etag)
        .json(&json!({"name":"Writer A"}))
        .send();
    let b = server
        .client
        .patch(&url)
        .header("if-match", &etag)
        .json(&json!({"name":"Writer B"}))
        .send();
    let (a, b) = tokio::join!(a, b);
    let statuses = [a.unwrap().status(), b.unwrap().status()];
    assert_eq!(
        statuses
            .iter()
            .filter(|status| **status == StatusCode::OK)
            .count(),
        1
    );
    assert_eq!(
        statuses
            .iter()
            .filter(|status| **status == StatusCode::PRECONDITION_FAILED)
            .count(),
        1
    );
    let committed = server.get(id).await;
    assert_eq!(committed.revision, 2);
    let stale = server
        .client
        .delete(&url)
        .header("if-match", &etag)
        .send()
        .await
        .unwrap();
    assert_eq!(stale.status(), StatusCode::PRECONDITION_FAILED);
    let conflict: Value = stale.json().await.unwrap();
    assert_eq!(conflict["current_revision"], 2);
    server.stop().await;

    let server = TestServer::start(directory.path(), None).await;
    let restored = server.get(id).await;
    assert_eq!(restored.model, committed.model);
    assert_eq!(restored.revision, 2);
    let cached = server
        .client
        .get(server.url(&format!("/projects/{id}")))
        .header("if-none-match", restored.etag())
        .send()
        .await
        .unwrap();
    assert_eq!(cached.status(), StatusCode::NOT_MODIFIED);
    let deleted = server
        .client
        .delete(server.url(&format!("/projects/{id}")))
        .header("if-match", restored.etag())
        .send()
        .await
        .unwrap();
    assert_eq!(deleted.status(), StatusCode::NO_CONTENT);
    assert!(!directory.path().join(format!("{id}.json")).exists());
    assert_eq!(
        server
            .client
            .get(server.url(&format!("/projects/{id}")))
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::NOT_FOUND
    );
    assert_eq!(
        server
            .client
            .delete(server.url(&format!("/projects/{id}")))
            .header("if-match", restored.etag())
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::NOT_FOUND
    );
    server.stop().await;
}

#[tokio::test]
async fn invalid_candidates_and_bad_headers_leave_snapshot_unchanged() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let (id, etag) = server.import(&graph()).await;
    let original = server.get(id).await;
    let file = directory.path().join(format!("{id}.json"));
    let original_bytes = std::fs::read(&file).unwrap();
    let model_url = server.url(&format!("/projects/{id}/model"));
    let mut invalid = original.model.clone();
    invalid.alias = "project".into();
    let rejected = server
        .client
        .put(&model_url)
        .header("if-match", &etag)
        .json(&invalid)
        .send()
        .await
        .unwrap();
    assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(
        rejected.headers()["content-type"],
        "application/problem+json"
    );
    assert!(
        !rejected.json::<Value>().await.unwrap()["issues"]
            .as_array()
            .unwrap()
            .is_empty()
    );

    let malformed = server
        .client
        .put(server.url(&format!("/projects/{id}/source")))
        .header("if-match", &etag)
        .header("content-type", "text/plain")
        .body("arcadia garbage")
        .send()
        .await
        .unwrap();
    assert_eq!(malformed.status(), StatusCode::UNPROCESSABLE_ENTITY);
    assert!(malformed.json::<Value>().await.unwrap()["issues"][0]["line"].is_number());
    let missing = server
        .client
        .put(&model_url)
        .json(&original.model)
        .send()
        .await
        .unwrap();
    assert_eq!(missing.status(), StatusCode::PRECONDITION_REQUIRED);
    for bad in ["\"", "*", "W/\"old\"", "\"first\",\"second\""] {
        let rejected = server
            .client
            .put(&model_url)
            .header("if-match", bad)
            .json(&original.model)
            .send()
            .await
            .unwrap();
        assert_eq!(rejected.status(), StatusCode::BAD_REQUEST);
    }
    assert_eq!(server.get(id).await.model, original.model);
    assert_eq!(std::fs::read(file).unwrap(), original_bytes);
    server.stop().await;
}

#[tokio::test]
async fn batches_dry_run_noop_and_source_share_the_core() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let (id, etag) = server.import(&graph()).await;
    let original = server.get(id).await;
    let first = original
        .model
        .elements
        .values()
        .find(|element| element.kind() == Kind::Activity)
        .unwrap()
        .id;
    let commands = vec![
        Command::Rename(first, "Updated activity".into()),
        Command::Describe(first, "Reviewed by API".into()),
    ];
    let transaction_url = server.url(&format!("/projects/{id}/transactions"));
    let rejected = server.client.post(&transaction_url).header("if-match", &etag)
        .json(&json!({"commands":[Command::Rename(first, "Must roll back".into()), Command::Delete(Id::new_v4())]})).send().await.unwrap();
    assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(server.get(id).await.model, original.model);
    let excessive = Command::Batch(vec![Command::Rename(first, "Never applied".into()); 200]);
    assert_eq!(
        server
            .client
            .post(&transaction_url)
            .header("if-match", &etag)
            .json(&json!({"commands":[excessive]}))
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::BAD_REQUEST
    );
    let preview = server
        .client
        .post(&transaction_url)
        .header("if-match", &etag)
        .json(&json!({"commands":commands,"dry_run":true}))
        .send()
        .await
        .unwrap();
    assert_eq!(preview.status(), StatusCode::OK);
    let preview: Value = preview.json().await.unwrap();
    assert_eq!(preview["dry_run"], true);
    assert_eq!(server.get(id).await.model, original.model);

    let commit = server
        .client
        .post(&transaction_url)
        .header("if-match", &etag)
        .json(&json!({"commands":commands}))
        .send()
        .await
        .unwrap();
    assert_eq!(commit.status(), StatusCode::OK);
    let new_etag = commit.headers()["etag"].to_str().unwrap().to_owned();
    let stored = server.get(id).await;
    assert_eq!(stored.revision, 2);
    assert_eq!(
        stored.model,
        oa_core::execute_batch(&original.model, &commands).unwrap()
    );
    let no_op = server
        .client
        .post(&transaction_url)
        .header("if-match", &new_etag)
        .json(&json!({"commands":commands}))
        .send()
        .await
        .unwrap();
    assert_eq!(no_op.status(), StatusCode::OK);
    assert_eq!(no_op.json::<Value>().await.unwrap()["changed"], false);

    let source = server
        .client
        .get(server.url(&format!("/projects/{id}/source")))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert_eq!(
        server
            .client
            .get(server.url(&format!("/projects/{id}/source?revision=1")))
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::BAD_REQUEST
    );
    assert_eq!(oa_core::language::parse(&source).unwrap(), stored.model);
    let imported = server
        .client
        .post(server.url("/projects/import"))
        .header("content-type", "text/plain; charset=utf-8")
        .body(source)
        .send()
        .await
        .unwrap();
    assert_eq!(imported.status(), StatusCode::CREATED);
    let imported: Value = imported.json().await.unwrap();
    let copy = server
        .get(imported["resource_id"].as_str().unwrap().parse().unwrap())
        .await;
    assert_ne!(copy.resource_id, stored.resource_id);
    assert_eq!(copy.model, stored.model);
    server.stop().await;
}

#[tokio::test]
async fn graph_filters_pagination_and_bounded_neighborhoods() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let model = graph();
    let (id, _) = server.import(&model).await;
    let page = server
        .client
        .get(server.url(&format!(
            "/projects/{id}/elements?kind=Activity&layer=Oa&limit=1"
        )))
        .send()
        .await
        .unwrap();
    assert_eq!(page.status(), StatusCode::OK);
    let page: Value = page.json().await.unwrap();
    assert_eq!(page["items"].as_array().unwrap().len(), 1);
    let after = page["next_after"].as_str().unwrap();
    let next: Value = server
        .client
        .get(server.url(&format!(
            "/projects/{id}/elements?kind=Activity&layer=Oa&limit=1&after={after}&revision=1"
        )))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_ne!(page["items"][0]["id"], next["items"][0]["id"]);
    let actor = model
        .elements
        .values()
        .find(|element| element.kind() == Kind::Actor)
        .unwrap()
        .id;
    let url = server.url(&format!("/projects/{id}/query"));
    let limited: Value = server
        .client
        .post(&url)
        .json(&json!({"roots":[actor],"max_depth":4,"max_objects":2}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(limited["truncated"], true);
    assert_eq!(limited["elements"].as_array().unwrap().len(), 1);
    assert!(limited["relations"].as_array().unwrap().is_empty());
    let complete: Value = server
        .client
        .post(&url)
        .json(&json!({"roots":[actor],"max_depth":4,"max_objects":100}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(complete["truncated"], false);
    assert_eq!(complete["elements"].as_array().unwrap().len(), 3);
    assert_eq!(complete["relations"].as_array().unwrap().len(), 2);
    let root_relation = *model.relations.keys().next().unwrap();
    let relation_root: Value = server
        .client
        .post(&url)
        .json(&json!({"roots":[root_relation],"max_depth":0,"max_objects":100}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(relation_root["relations"].as_array().unwrap().len(), 1);
    assert_eq!(relation_root["elements"].as_array().unwrap().len(), 2);
    assert_eq!(
        server
            .client
            .get(server.url(&format!("/projects/{id}/elements?revision=2")))
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::CONFLICT
    );
    assert_eq!(
        server
            .client
            .post(url)
            .json(&json!({"roots":[actor],"max_depth":99}))
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::BAD_REQUEST
    );
    server.stop().await;
}

#[tokio::test]
async fn auth_origins_media_and_body_limits_are_enforced_over_http() {
    let directory = tempfile::tempdir().unwrap();
    let token = "test-only-0123456789abcdef0123456789abcdef";
    let server = TestServer::start(directory.path(), Some(token.into())).await;
    let url = server.url("/projects");
    assert_eq!(
        server.client.get(&url).send().await.unwrap().status(),
        StatusCode::UNAUTHORIZED
    );
    assert_eq!(
        server
            .client
            .get(&url)
            .bearer_auth("wrong-token")
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::UNAUTHORIZED
    );
    assert_eq!(
        server
            .client
            .get(&url)
            .bearer_auth(token)
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::OK
    );
    assert_eq!(
        server
            .client
            .get(format!("{}/health", server.base))
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::OK
    );
    let openapi: Value = server
        .client
        .get(server.url("/openapi.json"))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(openapi["openapi"], "3.1.0");
    // Public automation vocabulary must expose every accepted engineering kind.
    let schemas = &openapi["components"]["schemas"];
    for kind in Kind::ALL {
        assert!(
            schemas["Kind"]["enum"]
                .as_array()
                .unwrap()
                .contains(&serde_json::to_value(kind).unwrap())
        );
    }
    for kind in RelationKind::ALL {
        let name = serde_json::to_value(kind).unwrap();
        assert!(
            schemas["RelationKind"]["enum"]
                .as_array()
                .unwrap()
                .contains(&name)
        );
        assert!(
            schemas["RelationBody"]["oneOf"]
                .as_array()
                .unwrap()
                .iter()
                .any(|variant| variant["properties"].get(name.as_str().unwrap()).is_some())
        );
    }
    assert_eq!(
        server
            .client
            .post(&url)
            .bearer_auth(token)
            .header("origin", "https://unrelated.example")
            .json(&json!({"name":"bad"}))
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::FORBIDDEN
    );
    assert_eq!(
        server
            .client
            .post(&url)
            .bearer_auth(token)
            .header("content-type", "text/plain")
            .body("{}")
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::UNSUPPORTED_MEDIA_TYPE
    );
    // Send the oversized Content-Length without uploading its rejected payload.
    // Continuing an 8 MiB upload after an early 413 can produce a Windows TCP reset.
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut socket = tokio::net::TcpStream::connect(server.base.trim_start_matches("http://"))
        .await
        .unwrap();
    socket.write_all(format!("POST /api/v1/projects HTTP/1.1\r\nHost: localhost\r\nAuthorization: Bearer {token}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", 8 * 1024 * 1024 + 1).as_bytes()).await.unwrap();
    let mut reply = String::new();
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        socket.read_to_string(&mut reply),
    )
    .await
    .unwrap()
    .unwrap();
    assert!(reply.starts_with("HTTP/1.1 413"), "{reply}");
    assert!(reply.contains("application/problem+json"), "{reply}");
    assert_eq!(
        server
            .client
            .get(server.url("/missing"))
            .bearer_auth(token)
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::NOT_FOUND
    );
    server.stop().await;
}

#[tokio::test]
async fn failed_atomic_replacement_preserves_the_accepted_model() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let (id, etag) = server.create().await;
    let before = server.get(id).await;
    let envelope = directory.path().join(format!("{id}.json"));
    let bytes = std::fs::read(&envelope).unwrap();
    // Deliberately make the atomic rename target unusable in this isolated store.
    std::fs::remove_file(&envelope).unwrap();
    std::fs::create_dir(&envelope).unwrap();
    let failed = server
        .client
        .patch(server.url(&format!("/projects/{id}")))
        .header("if-match", etag)
        .json(&json!({"name":"Must not be adopted"}))
        .send()
        .await
        .unwrap();
    assert_eq!(failed.status(), StatusCode::SERVICE_UNAVAILABLE);
    let after = server.get(id).await;
    assert_eq!(after.revision, before.revision);
    assert_eq!(after.model, before.model);
    assert_eq!(after.source_project, before.source_project);
    std::fs::remove_dir(&envelope).unwrap();
    std::fs::write(&envelope, bytes).unwrap();
    server.stop().await;
}

#[test]
fn storage_rejects_second_owners_corruption_and_identity_mismatch() {
    let directory = tempfile::tempdir().unwrap();
    let first = Store::open(directory.path()).unwrap();
    assert!(Store::open(directory.path()).is_err());
    drop(first);
    let filename = Id::new_v4();
    let path = directory.path().join(format!("{filename}.json"));
    std::fs::write(&path, "{broken").unwrap();
    assert!(Store::open(directory.path()).is_err());
    let model = Project::empty();
    let mismatched = StoredProject {
        migration: None,
        resource_id: Id::new_v4(),
        revision: 1,
        source_project: oa_core::source_project::format_bundle(&model),
        model,
    };
    std::fs::write(&path, serde_json::to_vec(&mismatched).unwrap()).unwrap();
    assert!(Store::open(directory.path()).is_err());
    let model = Project::empty();
    let invalid = StoredProject {
        migration: None,
        resource_id: filename,
        revision: 0,
        source_project: oa_core::source_project::format_bundle(&model),
        model,
    };
    std::fs::write(&path, serde_json::to_vec(&invalid).unwrap()).unwrap();
    assert!(Store::open(directory.path()).is_err());
}

#[test]
fn non_loopback_requires_a_token() {
    let mut config = ServerConfig {
        bind: "0.0.0.0:8787".parse().unwrap(),
        data_dir: "unused".into(),
        bearer_token: None,
    };
    assert!(config.validate().is_err());
    config.bearer_token = Some("test-only-0123456789abcdef0123456789abcdef".into());
    assert!(config.validate().is_ok());
    config.bearer_token = None;
    config.bind = "[::1]:8787".parse().unwrap();
    assert!(config.validate().is_ok());
}

#[tokio::test]
async fn transition_preview_apply_and_artifact_links_preserve_revision_boundaries() {
    use oa_core::transitions::{TransitionDirection, TransitionRequest};
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let model = graph();
    let source = model
        .elements
        .values()
        .find(|element| element.kind() == Kind::Activity)
        .unwrap()
        .id;
    let (id, etag) = server.import(&model).await;
    let request = TransitionRequest::new(TransitionDirection::OaToSa, vec![source]);
    let preview_url = server.url(&format!("/projects/{id}/transitions/preview"));
    let preview = server
        .client
        .post(&preview_url)
        .header("if-match", &etag)
        .json(&request)
        .send()
        .await
        .unwrap();
    let status = preview.status();
    let plan: Value = preview.json().await.unwrap();
    assert_eq!(status, StatusCode::OK, "{plan}");
    assert!(
        plan["plan"]["conflicts"].as_array().unwrap().is_empty(),
        "{plan}"
    );
    assert_eq!(server.get(id).await.model, model);
    let apply_url = server.url(&format!("/projects/{id}/transitions/apply"));
    let applied = server
        .client
        .post(&apply_url)
        .header("if-match", &etag)
        .json(&plan)
        .send()
        .await
        .unwrap();
    let status = applied.status();
    let result: Value = applied.json().await.unwrap();
    assert_eq!(status, StatusCode::OK, "{result}");
    let current = server.get(id).await;
    assert_eq!(current.revision, 2);
    assert!(
        current
            .model
            .elements
            .values()
            .any(|element| element.kind() == Kind::Function)
    );
    assert_eq!(current.model.traces.len(), 1);
    assert_eq!(
        server
            .client
            .post(&apply_url)
            .header("if-match", &etag)
            .json(&plan)
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::PRECONDITION_FAILED
    );
    assert_eq!(
        server
            .client
            .post(&apply_url)
            .header("if-match", current.etag())
            .json(&plan)
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::PRECONDITION_FAILED
    );

    let artifact = oa_core::ArtifactLink {
        id: Id::new_v4(),
        alias: "evidence_report".into(),
        name: "Evidence report".into(),
        description: "Produced by an external tool".into(),
        owner: source,
        relation: oa_core::ArtifactRelation::Verifies,
        uri: "https://example.invalid/report".into(),
        revision: Some("external:revision-42".into()),
        fragment: Some("section-7".into()),
    };
    let committed = server
        .client
        .post(server.url(&format!("/projects/{id}/transactions")))
        .header("if-match", current.etag())
        .json(&json!({"commands":[Command::CreateArtifact(artifact.clone())]}))
        .send()
        .await
        .unwrap();
    assert_eq!(committed.status(), StatusCode::OK);
    let records: Value = server
        .client
        .get(server.url(&format!("/projects/{id}/artifact-links?object={source}")))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(
        records["items"][0],
        serde_json::to_value(&artifact).unwrap()
    );
    let traces: Value = server
        .client
        .get(server.url(&format!("/projects/{id}/traces?object={source}")))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(traces["items"].as_array().unwrap().len(), 1);
    server.stop().await;
    let server = TestServer::start(directory.path(), None).await;
    assert_eq!(
        server.get(id).await.model.artifact_links[&artifact.id],
        artifact
    );
    server.stop().await;
}

#[tokio::test]
async fn fixed_store_capacity_is_enforced_on_http_and_restart() {
    let directory = tempfile::tempdir().unwrap();
    let mut last = None;
    for _ in 0..oa_server::store::MAX_PROJECTS {
        let model = Project::empty();
        let envelope = StoredProject {
            migration: None,
            resource_id: Id::new_v4(),
            revision: 1,
            source_project: oa_core::source_project::format_bundle(&model),
            model,
        };
        std::fs::write(
            directory
                .path()
                .join(format!("{}.json", envelope.resource_id)),
            serde_json::to_vec(&envelope).unwrap(),
        )
        .unwrap();
        last = Some(envelope);
    }
    let server = TestServer::start(directory.path(), None).await;
    let rejected = server
        .client
        .post(server.url("/projects"))
        .json(&json!({"name":"One too many"}))
        .send()
        .await
        .unwrap();
    assert_eq!(rejected.status(), StatusCode::INSUFFICIENT_STORAGE);
    let last = last.unwrap();
    assert_eq!(
        server
            .client
            .delete(server.url(&format!("/projects/{}", last.resource_id)))
            .header("if-match", last.etag())
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::NO_CONTENT
    );
    server.create().await;
    server.stop().await;
    std::fs::write(
        directory.path().join(format!("{}.json", last.resource_id)),
        serde_json::to_vec(&last).unwrap(),
    )
    .unwrap();
    assert!(Store::open(directory.path()).is_err());
}

#[tokio::test]
async fn chunked_upload_limits_and_loopback_host_check_use_the_real_transport() {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    assert_eq!(
        server
            .client
            .get(server.url("/projects"))
            .header("host", "untrusted.example")
            .send()
            .await
            .unwrap()
            .status(),
        StatusCode::FORBIDDEN
    );
    let socket = tokio::net::TcpStream::connect(server.base.trim_start_matches("http://"))
        .await
        .unwrap();
    let (mut reader, mut writer) = socket.into_split();
    let upload = tokio::spawn(async move {
        writer.write_all(b"POST /api/v1/projects HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n").await?;
        let chunk = vec![b' '; 65_536];
        for _ in 0..128 {
            writer.write_all(b"10000\r\n").await?;
            writer.write_all(&chunk).await?;
            writer.write_all(b"\r\n").await?;
        }
        writer.write_all(b"1\r\nx\r\n0\r\n\r\n").await
    });
    let mut reply = Vec::new();
    let read = tokio::time::timeout(
        std::time::Duration::from_secs(10),
        reader.read_to_end(&mut reply),
    )
    .await;
    assert!(
        read.is_ok(),
        "Server did not reject oversized chunked body promptly."
    );
    // TCP can reset after a rejection while a writer is still sending its tail.
    // Any delivered response still must carry the explicit body-limit status.
    let _ = upload.await.unwrap();
    let reply = String::from_utf8_lossy(&reply);
    assert!(reply.starts_with("HTTP/1.1 413"), "{reply}");
    assert!(reply.contains("application/problem+json"), "{reply}");
    server.stop().await;
}

#[tokio::test]
async fn legacy_json_import_and_model_reidentification_keep_the_service_address() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let model_id = Id::new_v4();
    let legacy = json!({"schema_version":1,"id":model_id,"name":"Migrated mission","elements":{},"relations":{}});
    let imported = server
        .client
        .post(server.url("/projects/import"))
        .json(&legacy)
        .send()
        .await
        .unwrap();
    let status = imported.status();
    let value: Value = imported.json().await.unwrap();
    assert_eq!(status, StatusCode::CREATED, "{value}");
    let resource_id: Id = value["resource_id"].as_str().unwrap().parse().unwrap();
    let current = server.get(resource_id).await;
    assert_eq!(current.model.id, model_id);
    assert_eq!(current.model.schema_version, oa_core::SCHEMA_VERSION);
    let new_model_id = Id::new_v4();
    let changed = server
        .client
        .post(server.url(&format!("/projects/{resource_id}/transactions")))
        .header("if-match", current.etag())
        .json(&json!({"commands":[Command::Reidentify(model_id,new_model_id)]}))
        .send()
        .await
        .unwrap();
    assert_eq!(changed.status(), StatusCode::OK);
    let after = server.get(resource_id).await;
    assert_eq!(after.resource_id, resource_id);
    assert_eq!(after.model.id, new_model_id);
    assert!(
        directory
            .path()
            .join(format!("{resource_id}.json"))
            .is_file()
    );
    assert!(
        !directory
            .path()
            .join(format!("{new_model_id}.json"))
            .exists()
    );
    server.stop().await;
}

#[tokio::test]
async fn current_migration_reports_and_invalid_typed_transaction_preserve_http_state() {
    let directory = tempfile::tempdir().unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let old = v2_missing_ports();
    let expected =
        oa_core::persistence::decode_with_report(&serde_json::to_vec(&old).unwrap()).unwrap();
    let imported = server
        .client
        .post(server.url("/projects/import"))
        .json(&old)
        .send()
        .await
        .unwrap();
    assert_eq!(imported.status(), StatusCode::CREATED);
    let etag = imported.headers()["etag"].to_str().unwrap().to_owned();
    let summary: Value = imported.json().await.unwrap();
    assert_eq!(summary["schema_version"], oa_core::SCHEMA_VERSION);
    assert_eq!(
        summary["migration"],
        serde_json::to_value(&expected.migration).unwrap()
    );
    let id: Id = summary["resource_id"].as_str().unwrap().parse().unwrap();
    let stored = server.get(id).await;
    assert_eq!(stored.model, expected.project);
    assert_eq!(stored.migration, expected.migration);
    let filename = directory.path().join(format!("{id}.json"));
    let before = std::fs::read(&filename).unwrap();
    let envelope: Value = serde_json::from_slice(&before).unwrap();
    assert!(envelope.get("migration").is_none());
    assert_eq!(envelope["model"]["schema_version"], oa_core::SCHEMA_VERSION);
    let validation = server
        .client
        .post(server.url(&format!("/projects/{id}/validation")))
        .json(&json!({"format":"json", "model":old}))
        .send()
        .await
        .unwrap();
    assert_eq!(validation.status(), StatusCode::OK);
    let validation: Value = validation.json().await.unwrap();
    assert_eq!(validation["valid"], true);
    assert_eq!(validation["migration"], summary["migration"]);
    let mut invalid = stored
        .model
        .relations
        .values()
        .find(|r| r.kind() == RelationKind::FunctionalExchange)
        .unwrap()
        .clone();
    if let oa_core::RelationBody::FunctionalExchange { source_port, .. } = &mut invalid.body {
        *source_port = Id::nil();
    }
    let rejected = server
        .client
        .post(server.url(&format!("/projects/{id}/transactions")))
        .header("if-match", &etag)
        .json(&json!({"commands":[Command::EditRelation(invalid.id, invalid)]}))
        .send()
        .await
        .unwrap();
    assert_eq!(rejected.status(), StatusCode::UNPROCESSABLE_ENTITY);
    assert_eq!(server.get(id).await.revision, 1);
    assert_eq!(std::fs::read(&filename).unwrap(), before);
    let source = server
        .client
        .get(server.url(&format!("/projects/{id}/source")))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(source.starts_with("arcadia 7;"));
    assert_eq!(oa_core::language::parse(&source).unwrap(), stored.model);
    let no_op = server
        .client
        .put(server.url(&format!("/projects/{id}/source")))
        .header("content-type", "text/plain")
        .header("if-match", &etag)
        .body(source)
        .send()
        .await
        .unwrap();
    assert_eq!(no_op.status(), StatusCode::OK);
    assert_eq!(no_op.headers()["etag"].to_str().unwrap(), etag);
    assert_eq!(std::fs::read(&filename).unwrap(), before);
    server.stop().await;
    let restarted = TestServer::start(directory.path(), None).await;
    let reopened = restarted.get(id).await;
    assert_eq!(reopened.model, stored.model);
    assert_eq!(reopened.revision, 1);
    assert!(reopened.migration.is_none());
    let no_op_legacy = restarted
        .client
        .put(restarted.url(&format!("/projects/{id}/model")))
        .header("if-match", &etag)
        .json(&old)
        .send()
        .await
        .unwrap();
    assert_eq!(no_op_legacy.status(), StatusCode::OK);
    assert_eq!(no_op_legacy.headers()["etag"].to_str().unwrap(), etag);
    let no_op_legacy: Value = no_op_legacy.json().await.unwrap();
    assert_eq!(no_op_legacy["changed"], false);
    assert_eq!(no_op_legacy["migration"], summary["migration"]);
    assert_eq!(std::fs::read(&filename).unwrap(), before);
    restarted.stop().await;
}

#[tokio::test]
async fn reopening_v2_envelope_reports_migration_without_saving_or_advancing_revision() {
    let directory = tempfile::tempdir().unwrap();
    let id = Id::new_v4();
    let filename = directory.path().join(format!("{id}.json"));
    let old = v2_missing_ports();
    let original = serde_json::to_vec(&json!({"resource_id":id,"revision":7,"model":old})).unwrap();
    std::fs::write(&filename, &original).unwrap();
    let server = TestServer::start(directory.path(), None).await;
    let migrated = server.get(id).await;
    assert_eq!(migrated.resource_id, id);
    assert_eq!(migrated.revision, 7);
    assert_eq!(migrated.model.schema_version, oa_core::SCHEMA_VERSION);
    assert_eq!(
        migrated
            .migration
            .as_ref()
            .unwrap()
            .generated_endpoints
            .len(),
        2
    );
    assert_eq!(std::fs::read(&filename).unwrap(), original);
    let rename = server
        .client
        .patch(server.url(&format!("/projects/{id}")))
        .header("if-match", migrated.etag())
        .json(&json!({"name":"Reviewed migrated model"}))
        .send()
        .await
        .unwrap();
    assert_eq!(rename.status(), StatusCode::OK);
    let current = server.get(id).await;
    assert_eq!(current.revision, 8);
    assert_eq!(current.model.name, "Reviewed migrated model");
    let saved: Value = serde_json::from_slice(&std::fs::read(&filename).unwrap()).unwrap();
    assert_eq!(saved["revision"], 8);
    assert_eq!(saved["model"]["schema_version"], oa_core::SCHEMA_VERSION);
    assert!(saved.get("migration").is_none());
    server.stop().await;
}