/* Подпись источника, RS-потери, кольцо, циклическое время и реальные uasync-таймеры источника. */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "../src/stream/stream_source.h"
#include "../src/stream/stream_metrics.h"
#include "../src/stream/stream_load.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "test_clock.h"

#define CHECK(expr) do { if (!(expr)) { \
    DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_core failed line=%d: %s", __LINE__, #expr); exit(1); \
} } while (0)

static const struct stream_descriptor descriptor = {
    .group_id = UINT64_C(0x0123456789abcdef), .source_node_id = UINT64_C(0xfedcba9876543210),
    .source_epoch = UINT64_C(0x0102030405060708), .block_size = STREAM_RS_DEFAULT_SIZE, .n = 8, .r = 2
};
static uint8_t private_key[32];
static unsigned cache_added, cache_removed;
static void cache_observer(void *arg, const uint8_t *wire, size_t length, int added) {
    (void)arg; CHECK(length == 1300 && stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8) == descriptor.group_id);
    if (added) cache_added++; else cache_removed++;
}

/* Рост до X, ровно 500мс подтверждения, короткий глитч, смена состава и фактические задержанные интервалы. */
static void test_load(void) {
    struct stream_load load; struct stream_load_snapshot snapshot;
    stream_load_init(&load, 0);
    CHECK(stream_load_add(&load, 10, 10000) == STREAM_OK);
    CHECK(stream_load_remove(&load, 6, 6000, 1) == STREAM_OK);
    CHECK(stream_load_poll(&load, 200, 250, &snapshot) == STREAM_OK && snapshot.growth_rate == 20000 && !snapshot.confirmed);
    CHECK(stream_load_add(&load, 2, 2000) == STREAM_OK && stream_load_remove(&load, 1, 1000, 0) == STREAM_OK);
    CHECK(stream_load_poll(&load, 400, 450, &snapshot) == STREAM_OK && snapshot.growth_rate == 15000);
    CHECK(snapshot.queued.bytes == 5000 && snapshot.arrivals.bytes == snapshot.sent.bytes + snapshot.removed.bytes + snapshot.queued.bytes);
    CHECK(stream_load_poll(&load, 699, 749, &snapshot) == STREAM_OK && !snapshot.confirmed);
    CHECK(stream_load_poll(&load, 700, 750, &snapshot) == STREAM_WAIT && snapshot.confirmed);
    CHECK(stream_load_remove(&load, 5, 5000, 0) == STREAM_OK);
    CHECK(stream_load_poll(&load, 701, 0, &snapshot) == STREAM_WAIT && !snapshot.pressure && !snapshot.confirmed);
    CHECK(stream_load_remove(&load, 1, 1000, 1) == STREAM_ERR_STATE);
    stream_load_init(&load, 1000);
    CHECK(stream_load_add(&load, 2, UINT64_C(1) << 40) == STREAM_OK);
    CHECK(stream_load_poll(&load, 1300, 300, &snapshot) == STREAM_OK && !snapshot.confirmed && snapshot.interval_ms == 300);
    CHECK(stream_load_remove(&load, 2, UINT64_C(1) << 40, 1) == STREAM_OK);
    CHECK(stream_load_poll(&load, 1400, 0, &snapshot) == STREAM_WAIT && !snapshot.confirmed);
    CHECK(stream_load_poll(&load, 1500, 0, &snapshot) == STREAM_OK && snapshot.arrivals.bytes == snapshot.sent.bytes);
    CHECK(stream_load_add(&load, 3, 3900) == STREAM_OK); stream_load_changed(&load);
    CHECK(stream_load_poll(&load, 1700, 300, &snapshot) == STREAM_OK && snapshot.growth_rate == 0 && !snapshot.confirmed);
    CHECK(stream_load_poll(&load, 13000, 300, &snapshot) == STREAM_OK && snapshot.input_rate == 390);
    struct stream_metric_snapshot input = {.lag_ms = UINT64_C(1) << 42, .blocks = UINT64_MAX, .late_blocks = 2, .stops = 3,
        .source_rate_bytes_sec = UINT64_C(1) << 40, .receive_rate_bytes_sec = 10000, .source_rate_valid = 1, .receive_rate_valid = 1, .stopped = 1};
    uint8_t wire[STREAM_METRIC_WIRE_SIZE]; stream_metric_write(wire, &input, 1);
    struct stream_metric_snapshot output; int valid = 0;
    CHECK(stream_metric_read(wire, &output, &valid) == STREAM_OK && valid && output.blocks == UINT64_MAX
          && output.lag_ms == input.lag_ms && output.source_rate_bytes_sec == input.source_rate_bytes_sec && output.stopped);
    wire[0] |= 128; CHECK(stream_metric_read(wire, &output, &valid) == STREAM_ERR_FORMAT);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] load: A/D/X/Q, growth before drops, 500ms boundary, glitches, revisions, interval/window and wide metrics");
}

/* Воспроизводимые данные, не совпадающие с zero-padding и проверочными кодограммами. */
static void pattern(uint8_t *data, size_t length) {
    uint32_t state = 0x12345678;
    for (size_t i = 0; i < length; i++) {
        state ^= state << 13; state ^= state >> 17; state ^= state << 5;
        data[i] = (uint8_t)state;
    }
}

/* Подмена любого класса неизменяемых полей отвергается, view и входной блок сохраняются. */
static void test_signed_blocks(void) {
    struct stream_encoder *encoder = NULL;
    CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
    CHECK(stream_block_size(&descriptor) == 1300 && stream_input_size(&descriptor) == 7080);
    uint8_t input[7080]; pattern(input, sizeof(input));
    const uint8_t *blocks[RS_MAX_BLOCKS];
    uint64_t seq = UINT64_C(0x1122334455667788);
    CHECK(stream_encoder_encode(encoder, seq, 65530, STREAM_BLOCK_TIME_RESET, input, sizeof(input), blocks) == STREAM_OK);
    const uint8_t *public_key = stream_encoder_public_key(encoder);
    for (unsigned i = 0; i < descriptor.n; i++) {
        struct stream_block_view view;
        CHECK(stream_block_verify(&descriptor, public_key, blocks[i], 1300, &view) == STREAM_OK);
        CHECK(view.seq == seq && view.subblock == i && view.timestamp_ms == 65530 && view.data_length == sizeof(input));
        CHECK(view.flags == STREAM_BLOCK_TIME_RESET && view.rs == blocks[i] + STREAM_BLOCK_RS_OFF);
        CHECK(!memcmp(blocks[i], "stream_block", 12));
        CHECK(!memcmp(blocks[i] + STREAM_BLOCK_GROUP_OFF, "\x01\x23\x45\x67\x89\xab\xcd\xef", 8));
        CHECK(!memcmp(blocks[i] + STREAM_BLOCK_SEQ_OFF, "\x11\x22\x33\x44\x55\x66\x77\x88", 8));
        CHECK(sc_ed25519_verify(public_key, blocks[i], 1300 - 64, blocks[i] + 1300 - 64) == SC_OK);
    }
    const unsigned changed[] = {0, 12, 20, 28, 36, 37, 38, 40, 45, 46, 47, 55, 56, 1235, 1299};
    uint8_t wire[STREAM_BLOCK_MAX_SIZE];
    for (unsigned i = 0; i < sizeof(changed) / sizeof(changed[0]); i++) {
        memcpy(wire, blocks[0], 1300); wire[changed[i]] ^= 1;
        struct stream_block_view view; memset(&view, 0xa5, sizeof(view));
        uint8_t before[sizeof(view)]; memcpy(before, &view, sizeof(view));
        CHECK(stream_block_verify(&descriptor, public_key, wire, 1300, &view) < 0);
        CHECK(!memcmp(before, &view, sizeof(view)));
    }
    struct stream_block_view view;
    CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 0, &view) == STREAM_ERR_FORMAT);
    CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 1299, &view) == STREAM_ERR_FORMAT);
    CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 1301, &view) == STREAM_ERR_FORMAT);
    uint8_t wrong_key[32]; memcpy(wrong_key, public_key, 32); wrong_key[0] ^= 1;
    CHECK(stream_block_verify(&descriptor, wrong_key, blocks[0], 1300, &view) == STREAM_ERR_SIGNATURE);
    CHECK(stream_encoder_encode(encoder, UINT64_MAX, 0, 0, input, 1, blocks) == STREAM_ERR_ARGUMENT);
    CHECK(stream_encoder_encode(encoder, 1, 0, 0, input, 0, blocks) == STREAM_ERR_ARGUMENT);
    CHECK(stream_encoder_encode(encoder, 1, 0, 2, input, 1, blocks) == STREAM_ERR_ARGUMENT);
    struct stream_descriptor invalid = descriptor; invalid.n = 0;
    CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT);
    invalid = descriptor; invalid.r = invalid.n;
    CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT);
    invalid = descriptor; invalid.block_size = 9;
    CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT);
    invalid = descriptor; invalid.source_epoch = 0;
    CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT);
    stream_encoder_destroy(encoder);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] signed blocks: canonical bytes, all field classes, wrong key and malformed sizes");
}

/* Wire сохраняет все 64 бита и не меняет выход при усечении; путь не допускает повторного посещения. */
static void test_wire(void) {
    uint8_t wire[STREAM_WIRE_HEADER + 8];
    struct stream_message input = {.descriptor = descriptor, .sender_epoch = UINT64_MAX, .receiver_epoch = UINT64_MAX - 1,
        .handle = UINT64_MAX - 2, .command = STREAM_CMD_SUBSCRIBE, .subblock = 7};
    stream_wire_header(wire, &input); stream_write_be(wire + STREAM_WIRE_HEADER, UINT64_MAX - 3, 8);
    struct stream_message parsed;
    CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_OK);
    CHECK(parsed.sender_epoch == input.sender_epoch && parsed.receiver_epoch == input.receiver_epoch && parsed.handle == input.handle);
    CHECK(parsed.descriptor.group_id == descriptor.group_id && parsed.descriptor.source_epoch == descriptor.source_epoch);
    CHECK(parsed.subblock == 7 && parsed.length == 8 && stream_read_be(parsed.payload, 8) == UINT64_MAX - 3);
    for (size_t i = 0; i < STREAM_WIRE_HEADER; i++) {
        memset(&parsed, 0xa5, sizeof(parsed)); uint8_t before[sizeof(parsed)]; memcpy(before, &parsed, sizeof(parsed));
        CHECK(stream_wire_parse(wire, i, &parsed) == STREAM_ERR_FORMAT && !memcmp(before, &parsed, sizeof(parsed)));
    }
    wire[1]++; CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_ERR_FORMAT); wire[1]--;
    wire[2] = 255; CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_ERR_FORMAT);
    struct stream_path path = {.nodes = {descriptor.source_node_id, 1, 2}, .count = 3}, read;
    uint8_t encoded[1 + STREAM_PATH_MAX * 8]; size_t used = 0, length = stream_path_write(encoded, &path);
    CHECK(stream_path_parse(encoded, length, &read, &used) == STREAM_OK && used == length && read.count == 3 && read.nodes[2] == 2);
    CHECK(stream_path_validate(&read, descriptor.source_node_id, 2, 3) == STREAM_OK);
    CHECK(stream_path_validate(&read, descriptor.source_node_id, 2, 1) == STREAM_ERR_FORMAT);
    read.nodes[2] = 1; CHECK(stream_path_validate(&read, descriptor.source_node_id, 1, 0) == STREAM_ERR_FORMAT);
    for (size_t i = 0; i < length; i++) CHECK(stream_path_parse(encoded, i, &read, &used) == STREAM_ERR_FORMAT);
    encoded[0] = STREAM_PATH_MAX + 1; CHECK(stream_path_parse(encoded, sizeof(encoded), &read, &used) == STREAM_ERR_FORMAT);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream wire: 64-bit fields, truncation, version, path repeats/receiver loop and size");
}

/* Любая пара потерь при 6+2 восстанавливается из подписанных блоков, отсутствующие блоки не регенерируются. */
static void test_fec_and_padding(void) {
    struct stream_encoder *encoder = NULL;
    CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
    uint8_t input[7080], output[7080]; pattern(input, sizeof(input));
    const uint8_t *blocks[RS_MAX_BLOCKS];
    CHECK(stream_encoder_encode(encoder, 0, 15, 0, input, sizeof(input), blocks) == STREAM_OK);
    for (unsigned first = 0; first < 8; first++) {
        for (unsigned second = first + 1; second < 8; second++) {
            struct stream_buffer *buffer = NULL;
            CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK);
            for (unsigned i = 0; i < 8; i++)
                if (i != first && i != second) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK);
            size_t length = 0;
            memset(output, 0xcc, sizeof(output));
            size_t allocations = u_get_allocated_count();
            CHECK(stream_buffer_decode(buffer, 0, output, sizeof(output), &length) == STREAM_OK);
            CHECK(u_get_allocated_count() == allocations && length == sizeof(input) && !memcmp(input, output, length));
            const uint8_t *missing = NULL; size_t missing_length = 0; uint64_t actual = 100;
            CHECK(stream_buffer_read(buffer, 0, (uint8_t)first, &missing, &missing_length, &actual) == STREAM_GAP);
            CHECK(!missing && missing_length == 0 && actual == 100);
            struct stream_buffer_stats stats; CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK);
            CHECK(stats.blocks == 6 && stats.stored_bytes == 6 * 1300);
            stream_buffer_destroy(buffer);
        }
    }
    struct stream_buffer *buffer = NULL;
    CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK);
    for (unsigned i = 0; i < 5; i++) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK);
    memset(output, 0xcc, sizeof(output)); size_t length = 123;
    CHECK(stream_buffer_decode(buffer, 0, output, sizeof(output), &length) == STREAM_WAIT);
    CHECK(length == 123);
    for (unsigned i = 0; i < sizeof(output); i++) CHECK(output[i] == 0xcc);
    CHECK(stream_buffer_put(buffer, blocks[5], 1300) == STREAM_OK);
    CHECK(stream_buffer_decode(buffer, 0, output, 1, &length) == STREAM_ERR_ARGUMENT);
    stream_buffer_destroy(buffer);
    stream_encoder_destroy(encoder);

    const struct { uint8_t n, r; uint16_t size; } variants[] = {{1, 0, 1189}, {3, 1, 20}, {16, 5, 4096}, {255, 254, 10}};
    for (unsigned v = 0; v < sizeof(variants) / sizeof(variants[0]); v++) {
        struct stream_descriptor d = descriptor; d.n = variants[v].n; d.r = variants[v].r; d.block_size = variants[v].size;
        encoder = NULL; buffer = NULL;
        CHECK(stream_encoder_create(&encoder, &d, private_key) == STREAM_OK);
        size_t size = stream_block_size(&d), wanted = stream_input_size(&d) > 17 ? 17 : stream_input_size(&d);
        CHECK(stream_encoder_encode(encoder, 37, 0, 0, input, wanted, blocks) == STREAM_OK);
        CHECK(stream_buffer_create(&buffer, &d, stream_encoder_public_key(encoder), d.n * size) == STREAM_OK);
        for (unsigned i = d.r; i < d.n; i++) CHECK(stream_buffer_put(buffer, blocks[i], size) == STREAM_OK);
        uint8_t partial[17]; length = 0;
        CHECK(stream_buffer_decode(buffer, 37, partial, wanted, &length) == STREAM_OK);
        CHECK(length == wanted && !memcmp(partial, input, wanted));
        stream_buffer_destroy(buffer); stream_encoder_destroy(encoder);
    }
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] FEC: all 28 double losses, insufficient set, partial data, n=1..255 and r=0");
}

/* Вставлять подписанную пачку в тестовое кольцо до повторного использования буферов encoder. */
static void put_batch(struct stream_encoder *encoder, struct stream_buffer *buffer, uint64_t seq, unsigned mask) {
    uint8_t input[37]; pattern(input, sizeof(input));
    const uint8_t *blocks[RS_MAX_BLOCKS];
    CHECK(stream_encoder_encode(encoder, seq, (uint16_t)seq, 0, input, sizeof(input), blocks) == STREAM_OK);
    for (unsigned i = 0; i < 8; i++) if (mask & (1u << i)) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK);
}

/* Общее кольцо хранит реальные блоки, принимает порядок 10/12/11 и вытесняет целую минимальную пачку. */
static void test_ring_and_conflicts(void) {
    struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL;
    CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
    CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 16 * 1300) == STREAM_OK);
    cache_added = cache_removed = 0; stream_buffer_set_observer(buffer, cache_observer, NULL);
    const uint8_t *wire = NULL; size_t length = 0; uint64_t actual = 100;
    CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_WAIT);
    put_batch(encoder, buffer, 10, 255); put_batch(encoder, buffer, 12, 255);
    CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_OK);
    CHECK(stream_buffer_put(buffer, wire, length) == STREAM_DUPLICATE);
    uint8_t bad[1300]; memcpy(bad, wire, length); bad[STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE] ^= 1;
    CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_SIGNATURE);
    CHECK(sc_ed25519_sign(private_key, bad, sizeof(bad) - 64, bad + sizeof(bad) - 64) == SC_OK);
    CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_CONFLICT);
    put_batch(encoder, buffer, 11, 1);
    CHECK(cache_added == 17 && cache_removed == 8);
    uint64_t pending = 0, oldest = 0;
    CHECK(stream_buffer_pending(buffer, 11, 0, &pending, &oldest) == STREAM_OK && pending == 2 && oldest);
    CHECK(stream_buffer_pending(buffer, 12, 0, &pending, &oldest) == STREAM_OK && pending == 1);
    CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_EXPIRED);
    CHECK(stream_buffer_read(buffer, 11, 1, &wire, &length, &actual) == STREAM_GAP);
    CHECK(stream_buffer_read(buffer, 13, 0, &wire, &length, &actual) == STREAM_WAIT);
    CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_OK && actual == 12);
    uint64_t first = 0, last = 0;
    CHECK(stream_buffer_range(buffer, 0, &first, &last) == STREAM_OK && first == 11 && last == 12);
    struct stream_buffer_stats stats; CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK);
    CHECK(stats.blocks == 9 && stats.batches == 2 && stats.stored_bytes == 9 * 1300 && stats.expired_before == 11);
    const uint8_t *blocks[RS_MAX_BLOCKS]; uint8_t input[37]; pattern(input, sizeof(input));
    CHECK(stream_encoder_encode(encoder, 10, 10, 0, input, sizeof(input), blocks) == STREAM_OK);
    CHECK(stream_buffer_put(buffer, blocks[0], 1300) == STREAM_EXPIRED);
    put_batch(encoder, buffer, 13, 255);
    CHECK(stream_buffer_read(buffer, 11, 0, &wire, &length, &actual) == STREAM_EXPIRED);
    put_batch(encoder, buffer, UINT64_MAX - 1, 1);
    CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_OK && actual == UINT64_MAX - 1);
    CHECK(stream_buffer_read(buffer, UINT64_MAX, 7, &wire, &length, &actual) == STREAM_OK && actual == 13);
    CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.stored_bytes <= stats.capacity_bytes);
    stream_buffer_destroy(buffer);

    buffer = NULL;
    CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 16 * 1300) == STREAM_OK);
    for (unsigned seq = 0; seq < 16; seq++) put_batch(encoder, buffer, seq, 1);
    CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.blocks == 16 && stats.batches == 16);
    CHECK(stats.stored_bytes == stats.capacity_bytes); /* Ни один из семи пустых субпотоков не резервирует место. */
    put_batch(encoder, buffer, 16, 1);
    CHECK(stream_buffer_read(buffer, 0, 0, &wire, &length, &actual) == STREAM_EXPIRED);
    CHECK(stream_buffer_range(buffer, 0, &first, &last) == STREAM_OK && first == 1 && last == 16);
    stream_buffer_destroy(buffer); stream_encoder_destroy(encoder);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] ring: shared data budget, sparse batches, out-of-order indices, whole-batch eviction and conflicts");
}

/* Подлинное время источника разворачивается через ноль; постоянный lag не считается полной остановкой. */
static void test_relative_metrics(void) {
    struct stream_metrics m; struct stream_metric_snapshot snapshot;
    stream_metrics_init(&m);
    CHECK(stream_metrics_poll(&m, 0, &snapshot) == STREAM_WAIT);
    CHECK(stream_metrics_observe(&m, 100, 65520, 0, 1300, 1000) == STREAM_OK);
    CHECK(stream_metrics_observe(&m, 101, 4, 0, 1300, 1020) == STREAM_OK);
    CHECK(stream_metrics_poll(&m, 1020, &snapshot) == STREAM_OK);
    CHECK(snapshot.lag_ms == 0 && snapshot.source_rate_bytes_sec == 65000 && snapshot.expected_interval_ms == 20);
    CHECK(snapshot.receive_rate_bytes_sec == 65000 && snapshot.blocks == 2 && !snapshot.stopped);
    CHECK(stream_metrics_observe(&m, 102, 24, 0, 1300, 1320) == STREAM_OK);
    CHECK(stream_metrics_observe(&m, 103, 44, 0, 1300, 1620) == STREAM_OK);
    CHECK(stream_metrics_poll(&m, 1620, &snapshot) == STREAM_OK);
    CHECK(snapshot.lag_ms == 560 && snapshot.late_blocks == 1 && snapshot.stops == 0 && !snapshot.stopped);
    uint64_t seq = 103, now = 1620; uint16_t timestamp = 44;
    for (unsigned i = 0; i < 30; i++) {
        seq++; now++; timestamp = (uint16_t)(timestamp + 20);
        CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now) == STREAM_OK);
    }
    CHECK(stream_metrics_poll(&m, now, &snapshot) == STREAM_OK && snapshot.lag_ms == 0);
    CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now + 300) == STREAM_DUPLICATE);
    CHECK(stream_metrics_poll(&m, now + 521, &snapshot) == STREAM_OK);
    CHECK(snapshot.stopped && snapshot.stops == 1 && snapshot.lag_ms == 501);
    CHECK(stream_metrics_poll(&m, now + 800, &snapshot) == STREAM_OK && snapshot.stops == 1);
    seq++; timestamp = (uint16_t)(timestamp + 20); now += 800;
    CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now) == STREAM_OK);
    CHECK(stream_metrics_poll(&m, now, &snapshot) == STREAM_OK && !snapshot.stopped && snapshot.stops == 1);
    CHECK(stream_metrics_observe(&m, seq + 1, timestamp, 0, 1300, now - 1) == STREAM_ERR_STATE);
    stream_metrics_end(&m);
    CHECK(stream_metrics_poll(&m, now + 70000, &snapshot) == STREAM_OK);
    CHECK(!snapshot.blocks && !snapshot.late_blocks && !snapshot.stops && !snapshot.stopped && !snapshot.source_rate_valid);
    stream_metrics_init(&m);
    CHECK(stream_metrics_observe(&m, 0, 0, 0, 1300, 0) == STREAM_OK);
    CHECK(stream_metrics_observe(&m, 1, 0, 0, 1300, 0) == STREAM_OK);
    CHECK(stream_metrics_poll(&m, 0, &snapshot) == STREAM_OK && !snapshot.source_rate_valid);
    CHECK(stream_metrics_observe(&m, 2, 1, 0, 1300, 1) == STREAM_OK);
    CHECK(stream_metrics_poll(&m, 1, &snapshot) == STREAM_OK && snapshot.source_rate_bytes_sec == 2600000);
    CHECK(stream_metrics_observe(&m, 3, 9, STREAM_BLOCK_TIME_RESET, 1300, 2) == STREAM_OK);
    CHECK(stream_metrics_poll(&m, 2, &snapshot) == STREAM_OK && snapshot.lag_ms == 0);
    CHECK(stream_metrics_observe(&m, 5, 10, 0, 1300, 1000) == STREAM_OK);
    CHECK(stream_metrics_poll(&m, 1000, &snapshot) == STREAM_OK && snapshot.lag_ms == 0);
    CHECK(stream_metrics_observe(&m, 6, 11, 0, 1300, 71000) == STREAM_OK);
    CHECK(stream_metrics_poll(&m, 71000, &snapshot) == STREAM_OK && snapshot.lag_ms == 0 && !snapshot.stopped);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] metrics: timestamp wrap, zero deltas, lag/catchup, complete stall, one episode and window expiry");
}

struct source_events {
    unsigned count;
    uint64_t seq;
    uint64_t time_tb;
    uint16_t timestamp_ms;
    uint8_t flags;
};

/* Callback видит уже всю пачку, а не частично опубликованные подписи. */
static void source_ready(struct stream_source *source, uint64_t seq, void *arg) {
    struct source_events *events = arg;
    events->count++; events->seq = seq; events->time_tb = get_time_tb();
    struct stream_buffer_stats stats;
    CHECK(stream_buffer_get_stats(stream_source_buffer(source), &stats) == STREAM_OK);
    CHECK(stats.blocks == events->count * 8 && stats.highest_seq == seq);
    const uint8_t *wire = NULL; size_t length = 0; uint64_t actual = UINT64_MAX;
    CHECK(stream_buffer_read(stream_source_buffer(source), seq, 0, &wire, &length, &actual) == STREAM_OK);
    CHECK(actual == seq && length == STREAM_BLOCK_DEFAULT_SIZE);
    events->timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2);
    events->flags = wire[STREAM_BLOCK_FLAGS_OFF];
    for (unsigned sub = 1; sub < 8; sub++) {
        CHECK(stream_buffer_read(stream_source_buffer(source), seq, (uint8_t)sub, &wire, &length, &actual) == STREAM_OK);
        CHECK(stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2) == events->timestamp_ms);
        CHECK(wire[STREAM_BLOCK_FLAGS_OFF] == events->flags);
    }
}

/* Настоящий uasync timer: не сбрасывает до 100 мс, полный блок отменяет timer, destroy отменяет остаток. */
static void test_source_timer(void) {
    test_clock_init(50);
    struct UASYNC *ua = uasync_create(); CHECK(ua != NULL);
    struct stream_source *source = NULL; struct source_events events = {0};
    CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK);
    uint8_t input[7080], output[7080]; pattern(input, sizeof(input));
    CHECK(stream_source_input_capacity(source) == sizeof(input));
    size_t accepted = 999;
    CHECK(stream_source_write(source, NULL, 0, &accepted) == STREAM_OK && accepted == 0);
    uint64_t start = get_time_tb();
    CHECK(stream_source_write(source, input, 17, &accepted) == STREAM_OK && accepted == 17);
    test_clock_run_for(ua, 500);
    CHECK(events.count == 0);
    test_clock_next_timer(ua);
    CHECK(events.count == 1 && events.seq == 0 && events.time_tb - start >= STREAM_SOURCE_FLUSH_TB);
    CHECK((uint16_t)(events.timestamp_ms - (uint16_t)(start / 10)) >= 100 && !events.flags);
    size_t length = 0;
    CHECK(stream_buffer_decode(stream_source_buffer(source), 0, output, 17, &length) == STREAM_OK);
    CHECK(length == 17 && !memcmp(input, output, length));
    CHECK(stream_source_write(source, input, sizeof(input), &accepted) == STREAM_OK && accepted == sizeof(input));
    CHECK(events.count == 2 && events.seq == 1);
    test_clock_run_for(ua, 1100);
    CHECK(events.count == 2); /* Нет лишней пустой пачки. */
    CHECK(stream_source_write(source, input, 13, &accepted) == STREAM_OK && accepted == 13);
    CHECK(stream_source_write(source, input + 13, sizeof(input) - 13, &accepted) == STREAM_OK);
    CHECK(events.count == 3 && events.seq == 2);
    test_clock_run_for(ua, 1100);
    CHECK(events.count == 3); /* Timer неполной пачки отменён заполнением. */
    CHECK(stream_buffer_decode(stream_source_buffer(source), 2, output, sizeof(output), &length) == STREAM_OK);
    CHECK(length == sizeof(input) && !memcmp(input, output, length));
    test_clock_run_for(ua, STREAM_TIMESTAMP_PERIOD_MS * 10);
    CHECK(stream_source_write(source, input, 23, &accepted) == STREAM_OK);
    CHECK(stream_source_finish(source) == STREAM_OK && events.count == 4);
    CHECK(events.flags == STREAM_BLOCK_TIME_RESET); /* Источник явно пометил неоднозначное число оборотов. */
    CHECK(stream_source_finish(source) == STREAM_OK && events.count == 4);
    CHECK(stream_source_write(source, input, 1, &accepted) == STREAM_ERR_STATE && accepted == 0);
    stream_source_destroy(source);
    source = NULL; events.count = 0;
    CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK);
    CHECK(stream_source_write(source, input, 17, &accepted) == STREAM_OK);
    stream_source_destroy(source);
    test_clock_run_for(ua, 1100);
    CHECK(events.count == 0);
    uasync_destroy(ua, 1); test_clock_finish();
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] source: 100ms timer, immediate full batch, padding, finish and cancellation on destroy");
}

int main(void) {
    debug_config_init();
    debug_set_level(DEBUG_LEVEL_ERROR);
    debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
    debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_DISABLED);
    CHECK(get_category_by_name("stream") == DEBUG_CATEGORY_STREAM);
    for (unsigned i = 0; i < sizeof(private_key); i++) private_key[i] = (uint8_t)(i * 37 + 11);
    size_t allocations = u_get_allocated_count();
    test_signed_blocks(); CHECK(u_get_allocated_count() == allocations);
    test_wire(); CHECK(u_get_allocated_count() == allocations);
    test_load(); CHECK(u_get_allocated_count() == allocations);
    test_fec_and_padding(); CHECK(u_get_allocated_count() == allocations);
    test_ring_and_conflicts(); CHECK(u_get_allocated_count() == allocations);
    test_relative_metrics(); CHECK(u_get_allocated_count() == allocations);
    test_source_timer(); CHECK(u_get_allocated_count() == allocations);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream core: all checks, no tracked allocations remain");
    return 0;
}