firmware/test/test_stream_api/test_main.cpp

#include "MeshTypes.h"
#include "SerialConsole.h"
#include "TestUtil.h"
#include "UptimeClock.h"
#include "configuration.h"
#include "gps/RTC.h"
#include "mesh-pb-constants.h"
#include "mesh/MeshService.h"
#include "mesh/NodeDB.h"
#include "mesh/StreamAPI.h"
#include "mesh/StreamFrameWriter.h"
#include <algorithm>
#include <cstdarg>
#include <cstdint>
#include <ctime>
#include <deque>
#include <limits>
#include <unity.h>
#include <vector>

/// Output-only stream whose write quotas deterministically simulate backpressure.
class ScriptedStream : public Stream
{
  public:
    /// Report that no input bytes are queued.
    int available() override { return 0; }
    /// Return end-of-input for the output-only stream.
    int read() override { return -1; }
    /// Return end-of-input without consuming data.
    int peek() override { return -1; }
    /// Report the configured output capacity.
    int availableForWrite() override { return availableCapacity; }

    /// Route single-byte writes through the quota-aware buffer writer.
    size_t write(uint8_t value) override { return write(&value, 1); }

    /// Accept at most the next scripted quota and capture accepted bytes.
    size_t write(const uint8_t *buffer, size_t size) override
    {
        requestedLengths.push_back(size);
        size_t quota = size;
        if (!writeQuotas.empty()) {
            quota = writeQuotas.front();
            writeQuotas.pop_front();
        }
        size_t accepted = std::min(quota, size);
        output.insert(output.end(), buffer, buffer + accepted);
        return accepted;
    }

    /// Record flush calls without changing captured output.
    void flush() override { flushCount++; }

    /// Set the maximum bytes accepted by the next write call.
    void queueWrite(size_t quota) { writeQuotas.push_back(quota); }

    int availableCapacity = std::numeric_limits<int>::max();
    unsigned flushCount = 0;
    std::deque<size_t> writeQuotas;
    std::vector<size_t> requestedLengths;
    std::vector<uint8_t> output;
};

/// Print sink that records bytes emitted by the real SerialConsole.
class RecordingPrint : public Print
{
  public:
    /// Capture one output byte.
    size_t write(uint8_t value) override
    {
        output.push_back(value);
        return 1;
    }

    std::vector<uint8_t> output;
};

/// Installs a MeshService for a test and restores the previous global service.
class ScopedMeshService
{
  public:
    /// Install the scoped service.
    ScopedMeshService() : previous(service) { service = &instance; }
    /// Restore the prior service after StreamAPI fixtures are destroyed.
    ~ScopedMeshService() { service = previous; }

  private:
    MeshService instance;
    MeshService *previous;
};

/// Exposes generic StreamAPI hooks and records frame-write behavior.
class StreamAPITestShim : public StreamAPI
{
  public:
    /// Construct the shim over a scripted stream.
    explicit StreamAPITestShim(Stream *stream) : StreamAPI(stream) {}

    /// Keep connection-timeout handling inactive during tests.
    bool checkIsConnected() override { return true; }

    /// Invoke the generic transport implementation rather than this shim's capture hook.
    bool writeBaseFrame(uint8_t *buf, size_t len, bool bestEffort = false) { return StreamAPI::writeFrame(buf, len, bestEffort); }

    bool finishReady = true;
    bool allowWrite = true;
    unsigned finishCalls = 0;
    unsigned frameWriteCalls = 0;
    unsigned failureCalls = 0;
    size_t failedFrameLen = 0;
    size_t failedWrittenLen = 0;
    std::vector<uint8_t> capturedPayload;

  protected:
    /// Record the pending-frame gate and return its configured state.
    bool finishPendingFrame() override
    {
        finishCalls++;
        return finishReady;
    }

    /// Apply the configured generic write-readiness result.
    bool canWriteFrame(size_t) override { return allowWrite; }

    /// Capture generic short-write failure metadata.
    void onFrameWriteFailed(size_t frameLen, size_t writtenLen) override
    {
        failureCalls++;
        failedFrameLen = frameLen;
        failedWrittenLen = writtenLen;
    }

    /// Capture one encoded PhoneAPI payload without writing it.
    bool writeFrame(uint8_t *buf, size_t len, bool bestEffort) override
    {
        (void)bestEffort;
        frameWriteCalls++;
        capturedPayload.assign(buf + 4, buf + 4 + len);
        return false;
    }
};

/// Minimal PhoneAPI transport for config-stream tests.
class PhoneAPITestShim : public PhoneAPI
{
  protected:
    bool checkIsConnected() override { return true; }
};

/// Exposes framed-log hooks and records best-effort writes.
class LogHookStreamAPI : public StreamAPI
{
  public:
    /// Construct the log shim over a scripted stream.
    explicit LogHookStreamAPI(Stream *stream) : StreamAPI(stream) {}

    /// Keep connection-timeout handling inactive during tests.
    bool checkIsConnected() override { return true; }

    /// Encode a formatted log through StreamAPI's production log path.
    void emitTestLog(const char *format, ...)
    {
        va_list args;
        va_start(args, format);
        emitLogRecord(meshtastic_LogRecord_Level_INFO, "test", format, args);
        va_end(args);
    }

    bool allowLogEncoding = false;
    unsigned frameWriteCalls = 0;
    bool lastBestEffort = false;

  protected:
    /// Apply the configured log-encoding gate.
    bool canEncodeLogRecord() override { return allowLogEncoding; }

    /// Record whether the encoded log was marked best-effort.
    bool writeFrame(uint8_t *, size_t, bool bestEffort) override
    {
        frameWriteCalls++;
        lastBestEffort = bestEffort;
        return true;
    }
};

/// Assert byte-for-byte equality between expected and captured stream output.
static void assertBytesEqual(const std::vector<uint8_t> &expected, const std::vector<uint8_t> &actual)
{
    TEST_ASSERT_EQUAL_UINT(expected.size(), actual.size());
    TEST_ASSERT_EQUAL_UINT8_ARRAY(expected.data(), actual.data(), expected.size());
}

/// Verify retries append only the unwritten tail and reproduce the frame exactly once.
void test_frame_writer_continues_only_unwritten_tail()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> frame = {0x94, 0xc3, 0x00, 0x06, 1, 2, 3, 4, 5, 6};
    stream.queueWrite(3);
    stream.queueWrite(2);
    stream.queueWrite(frame.size());

    TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), false));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_FALSE(writer.finishPendingFrame(stream));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_TRUE(writer.finishPendingFrame(stream));
    TEST_ASSERT_TRUE(writer.isIdle());

    std::vector<size_t> expectedRequests = {10, 7, 5};
    TEST_ASSERT_EQUAL_UINT(expectedRequests.size(), stream.requestedLengths.size());
    TEST_ASSERT_EQUAL_UINT64_ARRAY(expectedRequests.data(), stream.requestedLengths.data(), expectedRequests.size());
    assertBytesEqual(frame, stream.output);
    TEST_ASSERT_EQUAL_UINT(0, stream.flushCount);
}

/// Verify a replacement session receives a complete old frame before its new frame.
void test_frame_writer_completes_retained_tail_before_new_session_frame()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> oldFrame = {0x94, 0xc3, 0x00, 0x03, 0xa1, 0xa2, 0xa3};
    std::vector<uint8_t> newFrame = {0x94, 0xc3, 0x00, 0x02, 0xb1, 0xb2};
    stream.queueWrite(3);
    stream.queueWrite(oldFrame.size());
    stream.queueWrite(newFrame.size());

    TEST_ASSERT_FALSE(writer.writeFrame(stream, oldFrame.data(), oldFrame.size(), false));
    TEST_ASSERT_FALSE(writer.isIdle());

    // A replacement client starts without discarding the accepted old prefix.
    TEST_ASSERT_TRUE(writer.writeFrame(stream, newFrame.data(), newFrame.size(), false));
    TEST_ASSERT_TRUE(writer.isIdle());

    std::vector<uint8_t> expected = oldFrame;
    expected.insert(expected.end(), newFrame.begin(), newFrame.end());
    assertBytesEqual(expected, stream.output);
}

/// Verify a required main frame remains ordered behind a partial log frame.
void test_frame_writer_defers_main_behind_partial_log()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> logFrame = {0x94, 0xc3, 0x00, 0x02, 0xa1, 0xa2};
    std::vector<uint8_t> mainFrame = {0x94, 0xc3, 0x00, 0x03, 0xb1, 0xb2, 0xb3};
    stream.queueWrite(2);
    stream.queueWrite(0);
    stream.queueWrite(logFrame.size());
    stream.queueWrite(mainFrame.size());

    TEST_ASSERT_FALSE(writer.writeFrame(stream, logFrame.data(), logFrame.size(), true));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_FALSE(writer.writeFrame(stream, mainFrame.data(), mainFrame.size(), false));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_FALSE(writer.finishPendingFrame(stream));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_TRUE(writer.finishPendingFrame(stream));
    TEST_ASSERT_TRUE(writer.isIdle());

    std::vector<uint8_t> expected = logFrame;
    expected.insert(expected.end(), mainFrame.begin(), mainFrame.end());
    assertBytesEqual(expected, stream.output);
    TEST_ASSERT_EQUAL_UINT(4, stream.requestedLengths.size());
    TEST_ASSERT_EQUAL_UINT(0, stream.flushCount);
}

/// Verify best-effort output starts only when the complete frame fits.
void test_frame_writer_rejects_best_effort_without_full_capacity()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> frame = {0x94, 0xc3, 0x00, 0x02, 1, 2};
    stream.availableCapacity = frame.size() - 1;

    TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), true));
    TEST_ASSERT_TRUE(writer.isIdle());
    TEST_ASSERT_EQUAL_UINT(0, stream.requestedLengths.size());

    stream.availableCapacity = frame.size();
    TEST_ASSERT_TRUE(writer.writeFrame(stream, frame.data(), frame.size(), true));
    TEST_ASSERT_TRUE(writer.isIdle());
    TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size());
    assertBytesEqual(frame, stream.output);
}

/// Verify each zero-progress continuation makes one bounded write attempt.
void test_frame_writer_zero_progress_is_one_bounded_attempt()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> frame = {0x94, 0xc3, 0x00, 0x02, 1, 2};
    stream.queueWrite(1);
    stream.queueWrite(0);
    stream.queueWrite(0);
    stream.queueWrite(frame.size());

    TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), false));
    TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size());
    TEST_ASSERT_FALSE(writer.finishPendingFrame(stream));
    TEST_ASSERT_EQUAL_UINT(2, stream.requestedLengths.size());
    TEST_ASSERT_FALSE(writer.finishPendingFrame(stream));
    TEST_ASSERT_EQUAL_UINT(3, stream.requestedLengths.size());
    TEST_ASSERT_TRUE(writer.finishPendingFrame(stream));
    TEST_ASSERT_EQUAL_UINT(4, stream.requestedLengths.size());
    assertBytesEqual(frame, stream.output);
}

/// Verify generic StreamAPI framing and successful-write flush behavior.
void test_stream_api_full_write_frames_and_flushes()
{
    ScopedMeshService scopedService;
    ScriptedStream stream;
    StreamAPITestShim api(&stream);
    uint8_t frame[7] = {0, 0, 0, 0, 0x11, 0x22, 0x33};

    TEST_ASSERT_TRUE(api.writeBaseFrame(frame, 3));

    std::vector<uint8_t> expected = {0x94, 0xc3, 0x00, 0x03, 0x11, 0x22, 0x33};
    assertBytesEqual(expected, stream.output);
    TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size());
    TEST_ASSERT_EQUAL_UINT(1, stream.flushCount);
    TEST_ASSERT_EQUAL_UINT(0, api.failureCalls);
}

/// Verify generic transports report short writes without flushing or retrying.
void test_stream_api_short_write_reports_failure_without_flush()
{
    ScopedMeshService scopedService;
    ScriptedStream stream;
    StreamAPITestShim api(&stream);
    uint8_t frame[7] = {0, 0, 0, 0, 0x11, 0x22, 0x33};
    stream.queueWrite(5);

    TEST_ASSERT_FALSE(api.writeBaseFrame(frame, 3));

    TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size());
    TEST_ASSERT_EQUAL_UINT(0, stream.flushCount);
    TEST_ASSERT_EQUAL_UINT(1, api.failureCalls);
    TEST_ASSERT_EQUAL_UINT(7, api.failedFrameLen);
    TEST_ASSERT_EQUAL_UINT(5, api.failedWrittenLen);
}

/// Verify retained output blocks PhoneAPI from dequeuing the next payload.
void test_stream_api_finishes_pending_before_advancing_phone_api()
{
    ScopedMeshService scopedService;
    ScriptedStream stream;
    StreamAPITestShim api(&stream);
    api.sendConfigComplete();
    api.sendNotification(meshtastic_LogRecord_Level_WARNING, 42, "still queued");
    api.finishReady = false;

    api.runOncePart(nullptr, 0);
    TEST_ASSERT_EQUAL_UINT(1, api.finishCalls);
    TEST_ASSERT_EQUAL_UINT(0, api.frameWriteCalls);

    api.finishReady = true;
    api.runOncePart(nullptr, 0);
    TEST_ASSERT_EQUAL_UINT(2, api.finishCalls);
    TEST_ASSERT_EQUAL_UINT(1, api.frameWriteCalls);

    meshtastic_FromRadio decoded = meshtastic_FromRadio_init_zero;
    TEST_ASSERT_TRUE(
        pb_decode_from_bytes(api.capturedPayload.data(), api.capturedPayload.size(), &meshtastic_FromRadio_msg, &decoded));
    TEST_ASSERT_EQUAL_UINT(meshtastic_FromRadio_clientNotification_tag, decoded.which_payload_variant);
    TEST_ASSERT_EQUAL_UINT32(42, decoded.clientNotification.reply_id);
}

/// Verify framed logs honor the encoding gate and use best-effort writes.
void test_stream_api_gates_logs_and_marks_them_best_effort()
{
    ScopedMeshService scopedService;
    ScriptedStream stream;
    LogHookStreamAPI api(&stream);

    api.emitTestLog("blocked %u", 1U);
    TEST_ASSERT_EQUAL_UINT(0, api.frameWriteCalls);

    api.allowLogEncoding = true;
    api.emitTestLog("allowed %u", 2U);
    TEST_ASSERT_EQUAL_UINT(1, api.frameWriteCalls);
    TEST_ASSERT_TRUE(api.lastBestEffort);
}

/// Verify the real SerialConsole emits no unframed bytes in protobuf mode.
void test_serial_console_suppresses_raw_output_in_protobuf_mode()
{
    RecordingPrint sink;
    const bool oldHasLora = config.has_lora;
    const bool oldHasSecurity = config.has_security;
    const bool oldSerialEnabled = config.security.serial_enabled;
    const bool oldDebugLogApiEnabled = config.security.debug_log_api_enabled;

    config.has_lora = true;
    config.has_security = true;
    config.security.serial_enabled = true;
    config.security.debug_log_api_enabled = false;
    console->setDestination(&sink);

    console->write('A');
    const bool rawBeforeProtobuf = sink.output.size() == 1 && sink.output[0] == 'A';
    sink.output.clear();

    const uint8_t emptyToRadio = 0;
    console->handleToRadio(&emptyToRadio, 0);
    console->write('B');
    console->write('\n');
    console->log(MESHTASTIC_LOG_LEVEL_ERROR, "must stay framed");
    const bool emptyAfterProtobuf = sink.output.empty();

    console->setDestination(&Serial);
    config.has_lora = oldHasLora;
    config.has_security = oldHasSecurity;
    config.security.serial_enabled = oldSerialEnabled;
    config.security.debug_log_api_enabled = oldDebugLogApiEnabled;

    TEST_ASSERT_TRUE(rawBeforeProtobuf);
    TEST_ASSERT_TRUE(emptyAfterProtobuf);
}

// Build a phone->radio ADMIN_APP packet carrying `admin`, with an arbitrary wire `from`.
static meshtastic_MeshPacket makeAdminPacket(NodeNum from, const meshtastic_AdminMessage &admin)
{
    meshtastic_MeshPacket p = meshtastic_MeshPacket_init_zero;
    p.from = from;
    p.which_payload_variant = meshtastic_MeshPacket_decoded_tag;
    p.decoded.portnum = meshtastic_PortNum_ADMIN_APP;
    p.decoded.payload.size =
        pb_encode_to_bytes(p.decoded.payload.bytes, sizeof(p.decoded.payload.bytes), &meshtastic_AdminMessage_msg, &admin);
    return p;
}

// The lockdown admin gate must decide on the connection's authorization, not the wire `from`. A
// client that sets from != 0 previously skipped the gate, so an unauthorized connection could run
// admin. classifyLocalAdminPacket ignores `from`, so the same spoofed packet is still dropped.
static void test_lockdown_admin_gate_ignores_wire_from(void)
{
    meshtastic_AdminMessage setter = meshtastic_AdminMessage_init_zero;
    setter.which_payload_variant = meshtastic_AdminMessage_set_owner_tag;
    meshtastic_MeshPacket spoofed = makeAdminPacket(0x12345678, setter); // from != 0, the bypass

    meshtastic_AdminMessage out;
    TEST_ASSERT_EQUAL_MESSAGE((int)PhoneAPI::LocalAdminGate::DropUnauthorized,
                              (int)PhoneAPI::classifyLocalAdminPacket(spoofed, /*adminAuthorized=*/false, out),
                              "unauthorized admin with from != 0 must still be dropped");
    // Control: an authorized connection's identical packet passes through.
    TEST_ASSERT_EQUAL_MESSAGE((int)PhoneAPI::LocalAdminGate::AuthorizedPassThrough,
                              (int)PhoneAPI::classifyLocalAdminPacket(spoofed, /*adminAuthorized=*/true, out),
                              "authorized admin must not be dropped");

    // lockdown_auth is the authentication itself, so it is delivered inline regardless of from/auth.
    meshtastic_AdminMessage la = meshtastic_AdminMessage_init_zero;
    la.which_payload_variant = meshtastic_AdminMessage_lockdown_auth_tag;
    meshtastic_MeshPacket authPkt = makeAdminPacket(0x99, la);
    TEST_ASSERT_EQUAL((int)PhoneAPI::LocalAdminGate::LockdownAuth,
                      (int)PhoneAPI::classifyLocalAdminPacket(authPkt, /*adminAuthorized=*/false, out));

    // A non-admin packet is outside the gate entirely.
    meshtastic_MeshPacket text = meshtastic_MeshPacket_init_zero;
    text.which_payload_variant = meshtastic_MeshPacket_decoded_tag;
    text.decoded.portnum = meshtastic_PortNum_TEXT_MESSAGE_APP;
    TEST_ASSERT_EQUAL((int)PhoneAPI::LocalAdminGate::NotAdmin,
                      (int)PhoneAPI::classifyLocalAdminPacket(text, /*adminAuthorized=*/false, out));
}

// An ADMIN_APP packet whose payload is not a decodable AdminMessage must fall through to the
// normal reject path (NotAdmin), never be acted on as an admin command. The authorized control
// proves the decode-failure check runs before the auth branch, so it can't pass for the wrong reason.
static void test_lockdown_admin_gate_rejects_undecodable_admin(void)
{
    meshtastic_MeshPacket p = meshtastic_MeshPacket_init_zero;
    p.which_payload_variant = meshtastic_MeshPacket_decoded_tag;
    p.decoded.portnum = meshtastic_PortNum_ADMIN_APP;
    // Length-delimited field (tag 0x0A) claiming 16 bytes with none following: pb_decode fails.
    p.decoded.payload.bytes[0] = 0x0A;
    p.decoded.payload.bytes[1] = 0x10;
    p.decoded.payload.size = 2;

    meshtastic_AdminMessage out;
    TEST_ASSERT_EQUAL_MESSAGE((int)PhoneAPI::LocalAdminGate::NotAdmin,
                              (int)PhoneAPI::classifyLocalAdminPacket(p, /*adminAuthorized=*/false, out),
                              "undecodable ADMIN_APP payload must fall through to the reject path");
    TEST_ASSERT_EQUAL_MESSAGE((int)PhoneAPI::LocalAdminGate::NotAdmin,
                              (int)PhoneAPI::classifyLocalAdminPacket(p, /*adminAuthorized=*/true, out),
                              "undecodable ADMIN_APP payload must not pass through even when authorized");
}

static void test_want_config_includes_status_message_module_config(void)
{
    ScopedMeshService scopedService;
    NodeDB testNodeDB;
    NodeDB *const savedNodeDB = nodeDB;
    nodeDB = &testNodeDB;
    const auto savedModuleConfig = moduleConfig;
    moduleConfig.has_statusmessage = true;
    strncpy(moduleConfig.statusmessage.node_status, "Ready", sizeof(moduleConfig.statusmessage.node_status) - 1);
    moduleConfig.statusmessage.node_status[sizeof(moduleConfig.statusmessage.node_status) - 1] = '\0';

    meshtastic_ToRadio request = meshtastic_ToRadio_init_zero;
    request.which_payload_variant = meshtastic_ToRadio_want_config_id_tag;
    request.want_config_id = SPECIAL_NONCE_ONLY_CONFIG;
    uint8_t requestBytes[meshtastic_ToRadio_size];
    const size_t requestSize = pb_encode_to_bytes(requestBytes, sizeof(requestBytes), &meshtastic_ToRadio_msg, &request);

    PhoneAPITestShim api;
    api.handleToRadio(requestBytes, requestSize);

    bool foundStatusMessageConfig = false;
    for (unsigned i = 0; i < 64 && !foundStatusMessageConfig; ++i) {
        uint8_t responseBytes[meshtastic_FromRadio_size];
        const size_t responseSize = api.getFromRadio(responseBytes);
        meshtastic_FromRadio response = meshtastic_FromRadio_init_zero;
        TEST_ASSERT_TRUE(pb_decode_from_bytes(responseBytes, responseSize, &meshtastic_FromRadio_msg, &response));
        if (response.which_payload_variant == meshtastic_FromRadio_moduleConfig_tag &&
            response.moduleConfig.which_payload_variant == meshtastic_ModuleConfig_statusmessage_tag) {
            foundStatusMessageConfig = true;
            TEST_ASSERT_EQUAL_STRING("Ready", response.moduleConfig.payload_variant.statusmessage.node_status);
        }
    }

    api.close();
    moduleConfig = savedModuleConfig;
    nodeDB = savedNodeDB;
    TEST_ASSERT_TRUE(foundStatusMessageConfig);
}

/// Queue a packet as Router::dispatchReceived would have, before any time source existed.
static void queuePendingTimePlaceholderPacket(NodeNum from, uint32_t placeholderMillis)
{
    meshtastic_MeshPacket pending = meshtastic_MeshPacket_init_zero;
    pending.which_payload_variant = meshtastic_MeshPacket_decoded_tag;
    pending.decoded.portnum = meshtastic_PortNum_TEXT_MESSAGE_APP;
    pending.from = from;
    pending.to = NODENUM_BROADCAST;
    pending.rx_time = placeholderMillis;
    pending.has_rx_time = false;
    service->sendToPhone(packetPool.allocCopy(pending));
}

static void startHandshake(PhoneAPITestShim &api)
{
    meshtastic_ToRadio request = meshtastic_ToRadio_init_zero;
    request.which_payload_variant = meshtastic_ToRadio_want_config_id_tag;
    request.want_config_id = SPECIAL_NONCE_ONLY_CONFIG;
    uint8_t requestBytes[meshtastic_ToRadio_size];
    const size_t requestSize = pb_encode_to_bytes(requestBytes, sizeof(requestBytes), &meshtastic_ToRadio_msg, &request);
    api.handleToRadio(requestBytes, requestSize);
}

/// Drain the config stream looking for the first packet from `from`; false if never delivered.
static bool drainHandshakeForPacketFrom(PhoneAPITestShim &api, NodeNum from, meshtastic_MeshPacket &outPacket)
{
    for (unsigned i = 0; i < 256; ++i) {
        uint8_t responseBytes[meshtastic_FromRadio_size];
        const size_t responseSize = api.getFromRadio(responseBytes);
        if (responseSize == 0)
            return false;
        meshtastic_FromRadio response = meshtastic_FromRadio_init_zero;
        TEST_ASSERT_TRUE(pb_decode_from_bytes(responseBytes, responseSize, &meshtastic_FromRadio_msg, &response));
        if (response.which_payload_variant == meshtastic_FromRadio_packet_tag && response.packet.from == from) {
            outPacket = response.packet;
            return true;
        }
    }
    return false;
}

/// Swaps in a scratch NodeDB and the injected clock, restoring both plus the RTC on destruction.
/// Unity's TEST_ASSERT longjmps out on failure, so cleanup must not live at the end of the test.
class ScopedTimeFixture
{
  public:
    ScopedTimeFixture(uint32_t startMillis) : previous(nodeDB)
    {
        resetRTCStateForTests();
        nodeDB = &instance;
        Time::setTestMillis(startMillis);
    }
    ~ScopedTimeFixture()
    {
        nodeDB = previous;
        Time::useRealClock();
        resetRTCStateForTests();
    }

  private:
    NodeDB instance;
    NodeDB *previous;
};

// Time given at the start of the handshake, before the queued packet is drained: reconciliation
// (fired by the RTC quality crossing hook in RTC.cpp) rewrites the placeholder in place.
static void test_time_given_at_handshake_start_reconciles_queued_packet(void)
{
    ScopedMeshService scopedService;
    ScopedTimeFixture timeFixture(5000);

    const NodeNum sender = 0x12345678;
    queuePendingTimePlaceholderPacket(sender, 2000); // "received" 3s before the test's current millis()

    PhoneAPITestShim api;
    startHandshake(api);

    struct timeval networkTime;
    networkTime.tv_sec = time(NULL) + SEC_PER_DAY;
    networkTime.tv_usec = 0;
    TEST_ASSERT_EQUAL_INT(RTCSetResultSuccess, perhapsSetRTC(RTCQualityFromNet, &networkTime));

    meshtastic_MeshPacket delivered;
    TEST_ASSERT_TRUE_MESSAGE(drainHandshakeForPacketFrom(api, sender, delivered),
                             "queued packet was not delivered during the handshake");
    TEST_ASSERT_TRUE(delivered.has_rx_time);
    TEST_ASSERT_UINT32_WITHIN(2, (uint32_t)networkTime.tv_sec - 3, delivered.rx_time);

    api.close();
}

// Time given at the end - after the queued packet already left via the handshake: the delivered
// copy keeps its unresolved placeholder, since reconciliation can only rewrite what's still queued.
static void test_time_given_at_handshake_end_does_not_rewrite_already_sent_packet(void)
{
    ScopedMeshService scopedService;
    ScopedTimeFixture timeFixture(5000);

    const NodeNum sender = 0x12345678;
    queuePendingTimePlaceholderPacket(sender, 2000);

    PhoneAPITestShim api;
    startHandshake(api);

    // rx_time is proto3 optional, so has_rx_time false omits it from the wire entirely: the
    // decoded copy reads back 0 and the placeholder itself never left the device.
    meshtastic_MeshPacket delivered;
    TEST_ASSERT_TRUE_MESSAGE(drainHandshakeForPacketFrom(api, sender, delivered),
                             "queued packet was not delivered during the handshake");
    TEST_ASSERT_FALSE(delivered.has_rx_time);
    TEST_ASSERT_EQUAL_UINT32(0u, delivered.rx_time);

    // Time-giving transaction happens only now, at the end of the handshake.
    struct timeval networkTime;
    networkTime.tv_sec = time(NULL) + SEC_PER_DAY;
    networkTime.tv_usec = 0;
    TEST_ASSERT_EQUAL_INT(RTCSetResultSuccess, perhapsSetRTC(RTCQualityFromNet, &networkTime));

    // The already-delivered copy is a value, not a queue reference - untouched either way.
    TEST_ASSERT_FALSE(delivered.has_rx_time);
    TEST_ASSERT_EQUAL_UINT32(0u, delivered.rx_time);

    api.close();
}

/// Unity per-test setup; fixtures are local to each test.
void setUp(void) {}
/// Unity per-test teardown; fixtures clean themselves up.
void tearDown(void) {}

/// Initialize the native environment and run the stream regression suite.
void setup()
{
    initializeTestEnvironment();
    UNITY_BEGIN();
    RUN_TEST(test_frame_writer_continues_only_unwritten_tail);
    RUN_TEST(test_frame_writer_completes_retained_tail_before_new_session_frame);
    RUN_TEST(test_frame_writer_defers_main_behind_partial_log);
    RUN_TEST(test_frame_writer_rejects_best_effort_without_full_capacity);
    RUN_TEST(test_frame_writer_zero_progress_is_one_bounded_attempt);
    RUN_TEST(test_stream_api_full_write_frames_and_flushes);
    RUN_TEST(test_stream_api_short_write_reports_failure_without_flush);
    RUN_TEST(test_stream_api_finishes_pending_before_advancing_phone_api);
    RUN_TEST(test_stream_api_gates_logs_and_marks_them_best_effort);
    RUN_TEST(test_lockdown_admin_gate_ignores_wire_from);
    RUN_TEST(test_lockdown_admin_gate_rejects_undecodable_admin);
    RUN_TEST(test_want_config_includes_status_message_module_config);
    RUN_TEST(test_time_given_at_handshake_start_reconciles_queued_packet);
    RUN_TEST(test_time_given_at_handshake_end_does_not_rewrite_already_sent_packet);
    // usingProtobufs intentionally has no reset path, so this must run last.
    RUN_TEST(test_serial_console_suppresses_raw_output_in_protobuf_mode);
    exit(UNITY_END());
}

/// Unused Arduino loop required by the native Unity runner.
void loop() {}