#include "fake.h"
#include "session_broker.c"
#define printf capture_printf
#include "session_console.c"
#undef printf

static uint8_t payload[SESSION_BROKER_OUTPUT_SIZE + 256];
static void feed(size_t size) {
    assert(size <= sizeof(payload));
    serial_input = payload; serial_remaining = size;
    if (setjmp(task_exit) == 0) task_entry(NULL);
    assert(!mutex && serial_remaining == 0);
}
static session_broker_client_snapshot_t snapshot(session_broker_client_id_t id) {
    session_broker_client_snapshot_t s;
    assert(session_broker_get_client_snapshot(id, &s) == ESP_OK); return s;
}
static session_broker_global_snapshot_t global(void) {
    session_broker_global_snapshot_t s;
    assert(session_broker_get_global_snapshot(&s) == ESP_OK); return s;
}
static size_t drain(session_broker_client_id_t id, size_t size) {
    uint8_t data[sizeof(payload)]; size_t received;
    assert(size <= sizeof(data));
    assert(session_broker_read(id, data, size, &received) == ESP_OK); return received;
}
static session_broker_client_id_t connect_type(session_broker_client_type_t type) {
    session_broker_client_id_t id;
    assert(session_broker_connect(type, "SECRET-NAME-\033[2J", &id) == ESP_OK); return id;
}
static int command(const char *operation, session_broker_client_id_t id, const char *size) {
    char id_text[16]; snprintf(id_text, sizeof(id_text), "%u", id);
    char *argv[] = {"broker", (char *)operation, id_text, (char *)size};
    console_used = 0; console_output[0] = 0;
    return registered_command(id ? (size ? 4 : 3) : 2, argv);
}
static void disconnect_all(void) {
    session_broker_client_snapshot_t clients[SESSION_BROKER_MAX_CLIENTS];
    size_t n = session_broker_list_clients(clients, SESSION_BROKER_MAX_CLIENTS);
    for (size_t i = 0; i < n; ++i) assert(session_broker_disconnect(clients[i].id) == ESP_OK);
}
static void comparison(unsigned browsers) {
    session_broker_client_id_t ids[4];
    ids[0] = connect_type(SESSION_BROKER_CLIENT_USB);
    ids[1] = connect_type(SESSION_BROKER_CLIENT_SSH);
    for (unsigned i = 0; i < browsers; ++i) ids[2+i] = connect_type(SESSION_BROKER_CLIENT_WEB);
    assert(session_broker_clear_counters() == ESP_OK);
    size_t total = 0;
    while (total < 71292) {
        size_t n = 256;
        if (n > 71292 - total) n = 71292 - total;
        /* Deterministic slow-browser window, not a claim about real scheduling. */
        if (browsers == 2 && total < 23412 && n > 23412 - total) n = 23412 - total;
        feed(n); total += n;
        for (unsigned i = 0; i < 2+browsers; ++i) {
            if (browsers == 2 && i == 3 && total < 23412) continue;
            drain(ids[i], sizeof(payload));
        }
    }
    session_broker_global_snapshot_t g = global();
    uint64_t expected = 71292U * (2+browsers);
    uint64_t drops = browsers == 2 ? 19316 : 0;
    assert(g.counters.uart_rx_bytes == 71292 && g.counters.disconnections == 0);
    assert(g.counters.output_queued_bytes == expected-drops);
    assert(g.counters.output_read_bytes == expected-drops && g.counters.output_dropped_bytes == drops);
    for (unsigned i = 0; i < 2+browsers; ++i) {
        session_broker_client_snapshot_t s = snapshot(ids[i]);
        assert(s.counters.output_dropped_bytes == (i == 3 ? drops : 0));
        assert(s.counters.output_high_water_bytes == (i == 3 ? 4096 : 256));
    }
    printf("PASS synthetic %u-browser comparison: UART=71292 copies=%" PRIu64 " queued=%" PRIu64 " read=%" PRIu64 " dropped=%" PRIu64 " disconnect=0\n",
           browsers, expected, g.counters.output_queued_bytes, g.counters.output_read_bytes, drops);
    disconnect_all();
}
int main(void) {
    for (size_t i = 0; i < sizeof(payload); ++i) payload[i] = (uint8_t)i;
    assert(session_broker_get_global_snapshot(&(session_broker_global_snapshot_t){0}) == ESP_ERR_INVALID_STATE);
    assert(session_broker_init() == ESP_OK);
    assert(session_console_register_commands() == ESP_OK);
    size_t initial_allocations = allocations;
    session_broker_client_id_t fast = connect_type(SESSION_BROKER_CLIENT_USB);
    session_broker_client_id_t slow = connect_type(SESSION_BROKER_CLIENT_WEB);
    assert(session_broker_request_writer(fast) == ESP_OK);
    for (unsigned i = 0; i < 17; ++i) { feed(256); assert(drain(fast, 256) == 256); }
    session_broker_client_snapshot_t s = snapshot(slow);
    assert(s.output_bytes_pending == 4096 && s.counters.output_high_water_bytes == 4096);
    assert(s.counters.output_queued_bytes == 4096 && s.counters.output_dropped_bytes == 256);
    s = snapshot(fast);
    assert(s.counters.output_read_bytes == 4352 && s.counters.output_high_water_bytes == 256);
    assert(s.counters.output_dropped_bytes == 0 && s.is_writer);
    puts("PASS isolated fanout overflow and exact 4096-byte high-water");

    assert(drain(slow, 4000) == 4000);
    assert(snapshot(slow).counters.output_high_water_bytes == 4096);
    session_broker_global_snapshot_t before = global();
    assert(session_broker_clear_client_counters(slow) == ESP_OK);
    s = snapshot(slow);
    assert(s.output_bytes_pending == 96 && s.counters.output_high_water_bytes == 96);
    assert(s.counters.output_queued_bytes == 0 && s.counters.output_read_bytes == 0 && s.counters.output_dropped_bytes == 0);
    assert(global().counters.output_queued_bytes == before.counters.output_queued_bytes);
    feed(256); assert(snapshot(slow).counters.output_high_water_bytes == 352);
    assert(session_broker_clear_counters() == ESP_OK);
    s = snapshot(slow);
    assert(s.output_bytes_pending == 352 && s.counters.output_high_water_bytes == 352);
    assert(global().writer_id == fast && global().latest_event_sequence == before.latest_event_sequence);
    assert(global().counters.uart_rx_bytes == 0 && s.counters.uart_rx_bytes == 0);
    assert(drain(slow, 352) == 352);
    assert(snapshot(slow).counters.output_read_bytes == 352);
    feed(sizeof(payload));
    assert(snapshot(slow).counters.output_dropped_bytes == 256);
    puts("PASS per-client/global clears seed pending; subsequent read/drop/HWM accounting");

    before = global();
    assert(session_broker_disconnect(slow) == ESP_OK);
    assert(global().counters.output_dropped_bytes == before.counters.output_dropped_bytes + 4096);
    assert(session_broker_get_client_snapshot(slow, &s) == ESP_ERR_NOT_FOUND);
    session_broker_client_id_t replacement = connect_type(SESSION_BROKER_CLIENT_WEB);
    assert(replacement != slow && (replacement & 7) == (slow & 7));
    s = snapshot(replacement);
    assert(s.output_bytes_pending == 0 && s.counters.output_high_water_bytes == 0);
    assert(s.counters.output_queued_bytes == 0 && s.counters.output_read_bytes == 0 && s.counters.output_dropped_bytes == 0);
    assert(session_broker_clear_client_counters(slow) == ESP_ERR_NOT_FOUND);
    puts("PASS disconnect discard retained globally and generation-safe reuse resets diagnostics");

    while (session_broker_list_clients(NULL, 0) < SESSION_BROKER_MAX_CLIENTS)
        connect_type(SESSION_BROKER_CLIENT_INTERNAL);
    struct { session_broker_client_snapshot_t row; uint64_t guard; } bounded = {.guard = UINT64_MAX};
    assert(session_broker_list_clients(&bounded.row, 1) == 1 && bounded.guard == UINT64_MAX);
    before = global();
    assert(command("counters", 0, NULL) == 0);
    assert(strstr(console_output, "pending HWM") && !strstr(console_output, "SECRET-NAME") && !strchr(console_output, '\033'));
    unsigned lines = 0;
    for (const char *p = strstr(console_output, "ID         type"); *p; ++p) lines += *p == '\n';
    assert(lines == 1 + SESSION_BROKER_MAX_CLIENTS && console_used < 4096);
    assert(global().counters.output_read_bytes == before.counters.output_read_bytes);
    puts("PASS bounded eight-row metadata-only console counters; snapshots do not consume output");

    feed(1024);
    assert(command("read", replacement, "513") == 1);
    assert(snapshot(replacement).output_bytes_pending == 1024);
    assert(command("read", replacement, "0") == 1);
    assert(command("read", replacement, NULL) == 0);
    assert(strstr(console_output, "read 512 bytes: 00010203") && !strchr(console_output, '\033'));
    assert(snapshot(replacement).output_bytes_pending == 512);
    assert(command("read", replacement, "1") == 0);
    assert(snapshot(replacement).output_bytes_pending == 511);
    puts("PASS console read bounds and binary-safe hexadecimal rendering");
    disconnect_all();
    assert(command("counters", 0, NULL) == 0);
    comparison(1); comparison(2);
    assert(allocations == initial_allocations);
    puts("PASS no post-init broker allocations; 7 diagnostic groups passed");
    session_broker_management_snapshot_t management;
    assert(session_broker_get_management_snapshot(NULL) == ESP_ERR_INVALID_ARG);
    mutex = 1; assert(session_broker_get_management_snapshot(&management) == ESP_ERR_TIMEOUT); mutex = 0;
    session_broker_client_id_t usb = connect_type(SESSION_BROKER_CLIENT_USB);
    session_broker_client_id_t ssh = connect_type(SESSION_BROKER_CLIENT_SSH);
    session_broker_client_id_t web = connect_type(SESSION_BROKER_CLIENT_WEB);
    assert(session_broker_request_writer(usb) == ESP_OK);
    feed(128);
    assert(session_broker_get_management_snapshot(&management) == ESP_OK);
    assert(management.count == 3 && management.writer_id == usb && management.clients[0].pending == 128);
    uint32_t generation = management.generation;
    before = global();
    assert(session_broker_assign_writer_current(0, generation) == ESP_ERR_INVALID_ARG);
    assert(session_broker_assign_writer_current(ssh, 0) == ESP_ERR_INVALID_ARG);
    assert(session_broker_get_management_snapshot(&management) == ESP_OK && management.generation == generation);
    assert(global().latest_event_sequence == before.latest_event_sequence && snapshot(usb).output_bytes_pending == 128);
    assert(session_broker_assign_writer_current(ssh, generation) == ESP_OK);
    assert(global().writer_id == ssh && !snapshot(usb).is_writer && snapshot(ssh).is_writer);
    before = global();
    assert(session_broker_assign_writer_current(web, generation) == ESP_ERR_INVALID_STATE);
    assert(global().writer_id == ssh && global().latest_event_sequence == before.latest_event_sequence);
    assert(session_broker_get_management_snapshot(&management) == ESP_OK); generation = management.generation;
    assert(session_broker_release_writer(ssh) == ESP_OK && session_broker_request_writer(ssh) == ESP_OK);
    assert(session_broker_assign_writer_current(web, generation) == ESP_ERR_INVALID_STATE);
    assert(session_broker_get_management_snapshot(&management) == ESP_OK); generation = management.generation;
    assert(session_broker_disconnect(web) == ESP_OK);
    session_broker_client_id_t reused = connect_type(SESSION_BROKER_CLIENT_WEB);
    assert(reused != web && (reused & 7) == (web & 7));
    before = global();
    assert(session_broker_assign_writer_current(web, generation) == ESP_ERR_NOT_FOUND);
    assert(global().writer_id == ssh && global().latest_event_sequence == before.latest_event_sequence);
    assert(session_broker_clear_counters() == ESP_OK);
    assert(session_broker_get_management_snapshot(&management) == ESP_OK && management.generation == generation);
    assert(session_broker_assign_writer_current(reused, generation) == ESP_OK);
    assert(session_broker_get_management_snapshot(&management) == ESP_OK); generation = management.generation;
    assert(session_broker_force_release_writer(reused) == ESP_OK);
    assert(session_broker_assign_writer_current(usb, generation) == ESP_ERR_INVALID_STATE);
    assert(session_broker_request_writer(usb) == ESP_OK);
    assert(session_broker_get_management_snapshot(&management) == ESP_OK); generation = management.generation;
    assert(session_broker_disconnect(usb) == ESP_OK);
    assert(session_broker_assign_writer_current(ssh, generation) == ESP_ERR_INVALID_STATE);
    puts("PASS Broker management: atomic nonconsuming snapshot, one writer, USB/SSH/Web interleavings, stale target/reuse/ABA and counter-clear fencing");
    s_writer_generation = UINT32_MAX - 1;
    assert(session_broker_request_writer(ssh) == ESP_OK && s_writer_generation == UINT32_MAX);
    assert(session_broker_assign_writer_current(reused, UINT32_MAX) == ESP_ERR_INVALID_STATE);
    assert(session_broker_force_writer(reused) == ESP_OK && s_writer_generation == UINT32_MAX);
    disconnect_all();
    for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) s_slots[i].generation = SESSION_BROKER_MAX_GENERATION;
    s_slots[7].generation--;
    session_broker_client_id_t last = connect_type(SESSION_BROKER_CLIENT_USB);
    assert(last == UINT32_MAX);
    assert(session_broker_disconnect(last) == ESP_OK);
    assert(session_broker_connect(SESSION_BROKER_CLIENT_USB, "exhausted", &last) == ESP_ERR_NO_MEM);
    assert(!global().connected_clients && !global().writer_id);
    puts("PASS Broker wrap: saturated confirmation rejects, ordinary recovery remains; all 29-bit client generations retire without reuse");
    cleanup_allocations();
    return 0;
}