DMShellTest: end a session when the output is complete, not at a timer

-DDMSHELL_TEST_UNTIL_DONE=1 types a printf after the command whose output
is a marker. When the marker arrives in order, all of the command's output
has arrived. Typing then stops and a second marker is typed. Its output
follows everything typed before it, so seeing it means the shell has
caught up, and the session closes. DMSHELL_TEST_SESSION_MS becomes a cap,
30 minutes by default in this mode. The stats add complete_ms and
caught_up_ms. With the flag unset nothing changes.

Co-Authored-By: Jonathan Bennett <jbennett@incomsystems.biz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01REkPVFh6kvG4AZJ5A8AtM9

src/modules/DMShellRecovery.h

@@ -6,6 +6,7 @@
 
 #include "mesh/Throttle.h"
 #include <stdint.h>
+#include <string.h>
 
 /// What the receive side wants done with one inbound sequenced frame.
 struct DMShellRxDecision {
@@ -453,3 +454,32 @@ inline bool retransmitDue(uint32_t nowMs, uint32_t lastSentMs, uint32_t interval
 {
     return (uint32_t)(nowMs - lastSentMs) >= intervalMs;
 }
+
+/// Finds a fixed marker in a byte stream fed one byte at a time, so a match may span frames.
+class DMShellOutputMarker
+{
+  public:
+    explicit DMShellOutputMarker(const char *marker) : text(marker) {}
+
+    void reset() { matched = 0; }
+
+    /// True on the byte that completes the marker; matching then starts afresh, so matches never overlap.
+    bool feed(uint8_t b)
+    {
+        // The longest prefix of the marker that ends at b: the bytes before b are the prefix matched so far
+        uint32_t k = matched + 1;
+        while (k > 0 && ((uint8_t)text[k - 1] != b || memcmp(text, text + matched + 1 - k, k - 1) != 0)) {
+            k--;
+        }
+        matched = k;
+        if (text[matched] == '\0') {
+            matched = 0;
+            return true;
+        }
+        return false;
+    }
+
+  private:
+    const char *text;
+    uint32_t matched = 0;
+};

src/modules/DMShellTestModule.cpp

@@ -30,10 +30,19 @@
 #ifndef DMSHELL_TEST_CR_EVERY
 #define DMSHELL_TEST_CR_EVERY 10
 #endif
+// 1 = a session types until the command's output has all arrived, then closes once the shell has caught up with the
+// typing; DMSHELL_TEST_SESSION_MS is then only a cap
+#ifndef DMSHELL_TEST_UNTIL_DONE
+#define DMSHELL_TEST_UNTIL_DONE 0
+#endif
 // How long each session types for, from OPEN_OK; then it drains its input and closes
 #ifndef DMSHELL_TEST_SESSION_MS
+#if DMSHELL_TEST_UNTIL_DONE
+#define DMSHELL_TEST_SESSION_MS 1800000
+#else
 #define DMSHELL_TEST_SESSION_MS 300000
 #endif
+#endif
 #ifndef DMSHELL_TEST_SESSIONS
 #define DMSHELL_TEST_SESSIONS 1
 #endif
@@ -78,9 +87,14 @@ static constexpr uint32_t OPEN_TIMEOUT_MS = 180000;
 static constexpr uint32_t HEARTBEAT_IDLE_MS = 5000;
 static constexpr uint32_t HEARTBEAT_REPEAT_MS = 15000;
 static constexpr uint32_t DRAIN_TIMEOUT_MS = 10000;
+static constexpr uint32_t CATCH_UP_TIMEOUT_MS = 60000;
 static constexpr uint32_t CLOSE_WAIT_MS = 7000;
 static constexpr uint32_t ACTIVE_POLL_MS = 20;
 
+// Typed after the command and when typing stops; printf's arguments keep the terminal's echo from matching the markers
+static const char DONE_COMMAND[] = "printf '<%s:%s>\\n' dmshell done\n";
+static const char IDLE_COMMAND[] = "printf '<%s:%s>\\n' dmshell idle\n";
+
 DMShellTestModule *dmShellTestModule;
 
 static const char *opName(meshtastic_RemoteShell_OpCode op)
@@ -122,6 +136,8 @@ DMShellTestModule::DMShellTestModule()
              (unsigned)DMSHELL_TEST_PEER, (unsigned)DMSHELL_TEST_SESSIONS, (unsigned)DMSHELL_TEST_SESSION_MS,
              (unsigned)DMSHELL_TEST_START_DELAY_MS, DMSHELL_TEST_BATCH_PIPE ? "pipe" : "terminal",
              (unsigned)DMSHELL_TEST_INPUT_WINDOW, (unsigned)DMSHELL_TEST_COMMAND_AGAIN_MS);
+    if (DMSHELL_TEST_UNTIL_DONE)
+        LOG_INFO("DMShellTest: a session ends when its output is complete, capped at %u ms", (unsigned)DMSHELL_TEST_SESSION_MS);
 }
 
 int32_t DMShellTestModule::runOnce()
@@ -189,6 +205,15 @@ void DMShellTestModule::startSession(uint32_t now)
     pendingInputLen = 0;
     commandTyped = false;
     commandRepeated = false;
+    commandsTyped = 0;
+    doneMarker.reset();
+    idleMarker.reset();
+    doneMarksSeen = 0;
+    completeAtMs = 0;
+    caughtUpAtMs = 0;
+    idleTyped = false;
+    typingEnded = false;
+    typingEndedAtMs = 0;
     keystrokesTyped = 0;
     batchLen = 0;
     batchFirstByteMs = batchLastByteMs = 0;
@@ -221,24 +246,42 @@ void DMShellTestModule::endSession(uint32_t now, const char *reason)
 
 void DMShellTestModule::serviceActive(uint32_t now)
 {
-    const bool typing = !Throttle::deadlinePassedAt(now, openedAtMs + DMSHELL_TEST_SESSION_MS);
-    if (typing)
-        typeWorkload(now);
-    else
-        closeBatch(); // time is up: whatever is still being batched goes out now, as EOF ends the client's read
+    if (!typingEnded) {
+        const bool capped = Throttle::deadlinePassedAt(now, openedAtMs + DMSHELL_TEST_SESSION_MS);
+        if (capped || completeAtMs) {
+            typingEnded = true;
+            typingEndedAtMs = now;
+            if (DMSHELL_TEST_UNTIL_DONE && !completeAtMs)
+                LOG_WARN("DMShellTest t=%u cap of %u ms reached before the output completed", (unsigned)now,
+                         (unsigned)DMSHELL_TEST_SESSION_MS);
+            closeBatch(); // whatever is still being batched goes out now, as EOF ends the client's read
+            if (completeAtMs) {
+                // Its output follows everything typed before it, so seeing it means the shell has caught up
+                typeText(IDLE_COMMAND, now);
+                closeBatch();
+                idleTyped = true;
+            }
+        } else {
+            typeWorkload(now);
+        }
+    }
     flushInput(now);
     serviceRetransmit(now);
     if (phase != Phase::Active)
         return; // the peer stopped answering and the session ended
     serviceHeartbeat(now);
-    if (typing)
+    if (!typingEnded)
         return;
 
     const bool drained = pendingInputLen == 0 && batchLen == 0 && txWindow.outstanding() == 0;
-    if (drained || Throttle::deadlinePassedAt(now, openedAtMs + DMSHELL_TEST_SESSION_MS + DRAIN_TIMEOUT_MS)) {
+    const bool caughtUp = !idleTyped || caughtUpAtMs;
+    if ((drained && caughtUp) ||
+        Throttle::deadlinePassedAt(now, typingEndedAtMs + (idleTyped ? CATCH_UP_TIMEOUT_MS : DRAIN_TIMEOUT_MS))) {
         if (!drained)
             LOG_WARN("DMShellTest t=%u closing with input unacknowledged: %u queued, %u outstanding", (unsigned)now,
                      (unsigned)pendingInputLen, (unsigned)txWindow.outstanding());
+        if (!caughtUp)
+            LOG_WARN("DMShellTest t=%u closing before the shell caught up", (unsigned)now);
         sendFrame(meshtastic_RemoteShell_OpCode_CLOSE, now);
         closeSent = true;
         phase = Phase::Closing;
@@ -284,17 +327,44 @@ void DMShellTestModule::closeBatch()
     batchFirstByteMs = 0;
 }
 
-void DMShellTestModule::typeCommand(uint32_t now)
+void DMShellTestModule::typeText(const char *text, uint32_t now)
 {
-    static const char command[] = DMSHELL_TEST_COMMAND;
-    for (size_t i = 0; i + 1 < sizeof(command); i++) {
-        addByte((uint8_t)command[i], now);
+    for (const char *c = text; *c; c++) {
+        addByte((uint8_t)*c, now);
         if (batchDue(now))
             closeBatch();
     }
+}
+
+void DMShellTestModule::typeCommand(uint32_t now)
+{
+    static const char command[] = DMSHELL_TEST_COMMAND;
+    typeText(command, now);
+    if (DMSHELL_TEST_UNTIL_DONE)
+        typeText(DONE_COMMAND, now);
+    commandsTyped++;
     LOG_INFO("DMShellTest t=%u typed the command (%u bytes)", (unsigned)now, (unsigned)(sizeof(command) - 1));
 }
 
+// Only in-order output reaches here, so a marker means everything the shell printed before it has arrived
+void DMShellTestModule::noteOutput(const uint8_t *bytes, size_t len, uint32_t now)
+{
+    const uint32_t commandsWanted = DMSHELL_TEST_COMMAND_AGAIN_MS ? 2 : 1;
+    for (size_t i = 0; i < len; i++) {
+        if (doneMarker.feed(bytes[i])) {
+            doneMarksSeen++;
+            LOG_INFO("DMShellTest t=%u command output %u of %u complete", (unsigned)now, (unsigned)doneMarksSeen,
+                     (unsigned)commandsWanted);
+            if (!completeAtMs && doneMarksSeen >= commandsWanted)
+                completeAtMs = now;
+        }
+        if (idleMarker.feed(bytes[i]) && idleTyped && !caughtUpAtMs) {
+            caughtUpAtMs = now;
+            LOG_INFO("DMShellTest t=%u shell caught up", (unsigned)now);
+        }
+    }
+}
+
 // The bench workload: after the settle time the command once, then one keystroke per interval
 void DMShellTestModule::typeWorkload(uint32_t now)
 {
@@ -510,6 +580,8 @@ void DMShellTestModule::handleInOrder(const meshtastic_RemoteShell &frame, uint3
         break;
     case meshtastic_RemoteShell_OpCode_OUTPUT:
         stats.rxOutputBytes += frame.payload.size;
+        if (DMSHELL_TEST_UNTIL_DONE)
+            noteOutput(frame.payload.bytes, frame.payload.size, now);
         break;
     case meshtastic_RemoteShell_OpCode_ERROR:
         LOG_WARN("DMShellTest t=%u remote error seq=%u: %.*s", (unsigned)now, (unsigned)frame.seq, (int)frame.payload.size,
@@ -705,6 +777,12 @@ void DMShellTestModule::logStats(uint32_t now)
              (unsigned)stats.replayUnavailable, (unsigned)stats.replayEvicted, (unsigned)stats.openRetries);
     LOG_INFO("DMShellTest stats session=0x%08x recovery input_window_closed=%u input_dropped=%u", id,
              (unsigned)stats.inputWindowClosed, (unsigned)stats.pendingInputDropped);
+    if (DMSHELL_TEST_UNTIL_DONE) {
+        // From OPEN_OK; 0 = never
+        LOG_INFO("DMShellTest stats session=0x%08x until_done complete_ms=%u caught_up_ms=%u commands=%u", id,
+                 (unsigned)(completeAtMs ? completeAtMs - openedAtMs : 0),
+                 (unsigned)(caughtUpAtMs ? caughtUpAtMs - openedAtMs : 0), (unsigned)commandsTyped);
+    }
 }
 
 #endif

src/modules/DMShellTestModule.h

@@ -80,6 +80,16 @@ class DMShellTestModule : public SinglePortModule, private concurrency::OSThread
     uint32_t pendingInputLen = 0;
     bool commandTyped = false;
     bool commandRepeated = false;
+    uint32_t commandsTyped = 0;
+    // DMSHELL_TEST_UNTIL_DONE: what the shell prints once the command's output is complete, and once it has caught up
+    DMShellOutputMarker doneMarker{"<dmshell:done>"};
+    DMShellOutputMarker idleMarker{"<dmshell:idle>"};
+    uint32_t doneMarksSeen = 0;
+    uint32_t completeAtMs = 0;
+    uint32_t caughtUpAtMs = 0;
+    bool idleTyped = false;
+    bool typingEnded = false;
+    uint32_t typingEndedAtMs = 0;
     uint32_t keystrokesTyped = 0;
     uint32_t nextKeystrokeMs = 0;
     // The batch the client's stdin read would return, before it becomes one INPUT
@@ -96,6 +106,8 @@ class DMShellTestModule : public SinglePortModule, private concurrency::OSThread
     void serviceActive(uint32_t now);
     void typeWorkload(uint32_t now);
     void typeCommand(uint32_t now);
+    void typeText(const char *text, uint32_t now);
+    void noteOutput(const uint8_t *bytes, size_t len, uint32_t now);
     void addByte(uint8_t b, uint32_t now);
     bool batchDue(uint32_t now) const;
     void closeBatch();

test/test_dmshell_recovery/test_main.cpp

@@ -760,6 +760,52 @@ void test_dmshell_retransmit_due_survives_millis_wrap()
     TEST_ASSERT_TRUE(retransmitDue(sent + 1400, sent, 1400));
 }
 
+static uint32_t feedAll(DMShellOutputMarker &marker, const char *bytes)
+{
+    uint32_t matches = 0;
+    for (const char *c = bytes; *c; c++) {
+        if (marker.feed((uint8_t)*c)) {
+            matches++;
+        }
+    }
+    return matches;
+}
+
+void test_dmshell_output_marker_matches_across_frames()
+{
+    // OUTPUT frames split wherever the PTY read did, so the marker can straddle two of them
+    DMShellOutputMarker marker("<dmshell:done>");
+    TEST_ASSERT_EQUAL_UINT32(0, feedAll(marker, "tail of the file\r\n<dmsh"));
+    TEST_ASSERT_EQUAL_UINT32(1, feedAll(marker, "ell:done>\r\n$ "));
+}
+
+void test_dmshell_output_marker_ignores_the_echoed_command()
+{
+    // The terminal echoes what was typed; printf's arguments are chosen so the echo cannot match
+    DMShellOutputMarker marker("<dmshell:done>");
+    TEST_ASSERT_EQUAL_UINT32(0, feedAll(marker, "$ printf '<%s:%s>\\n' dmshell done\r\n"));
+    TEST_ASSERT_EQUAL_UINT32(1, feedAll(marker, "<dmshell:done>\r\n"));
+}
+
+void test_dmshell_output_marker_recovers_from_a_false_start()
+{
+    // A partial match broken by a byte that itself starts the marker again
+    DMShellOutputMarker marker("<dmshell:done>");
+    TEST_ASSERT_EQUAL_UINT32(1, feedAll(marker, "<dmsh<dmshell:done>"));
+    // A marker whose prefix repeats inside it: restarting from its first byte alone would miss this one
+    DMShellOutputMarker repeating("aab");
+    TEST_ASSERT_EQUAL_UINT32(1, feedAll(repeating, "aaab"));
+}
+
+void test_dmshell_output_marker_counts_each_occurrence()
+{
+    DMShellOutputMarker marker("<dmshell:done>");
+    TEST_ASSERT_EQUAL_UINT32(2, feedAll(marker, "<dmshell:done>\r\n$ <dmshell:done>"));
+    marker.feed('<');
+    marker.reset();
+    TEST_ASSERT_EQUAL_UINT32(0, feedAll(marker, "dmshell:done>"));
+}
+
 void setup()
 {
     initializeTestEnvironment();
@@ -811,6 +857,10 @@ void setup()
     RUN_TEST(test_dmshell_ack_latency_interval_is_twice_the_estimate_within_bounds);
     RUN_TEST(test_dmshell_retransmit_waits_for_the_frame_not_the_poll);
     RUN_TEST(test_dmshell_retransmit_due_survives_millis_wrap);
+    RUN_TEST(test_dmshell_output_marker_matches_across_frames);
+    RUN_TEST(test_dmshell_output_marker_ignores_the_echoed_command);
+    RUN_TEST(test_dmshell_output_marker_recovers_from_a_false_start);
+    RUN_TEST(test_dmshell_output_marker_counts_each_occurrence);
     exit(UNITY_END());
 }