fix(bot-mode): never resubmit into a still-running stranded group member

runGroupChatRounds' responder selection had no awareness of the stranded/
harvest state the previous round's harvest pass just confirmed. A member
whose turn timed out (marked stranded) but is STILL genuinely running
could be re-selected as a responder in a later round of the same
invocation — resolveGroupResponders has no busy filter, and a member's
watermark is bumped past the stranded timeout regardless of outcome, so a
fresh delta re-qualifies them.

Re-selecting them fires another prompt.submit into their live session.
tui_gateway's _handle_busy_submit treats that as a normal busy mid-turn
prompt: by default it either redirects the live turn in place or, for
older agents, hard-interrupts it and queues the new text as the next
turn. Either way the member's original in-flight work — exactly what the
stranded/harvest mechanism exists to protect — gets abandoned or killed,
undermining the "never lost, just late" guarantee.

Filter responders against the room's current stranded map (freshly
confirmed by this round's own harvest pass) before selecting who speaks.
A member with a live stranded marker is skipped; the next harvest pass
picks their reply up once it actually lands.

Added a regression test exercising runGroupChatRounds end-to-end with a
member confirmed still-busy: without the guard the round loop resubmits
into their session (asserted via prompt.submit call count); with the
guard it never does, and the marker survives untouched. Mutation-verified:
temporarily reverted the filter and confirmed the new test fails
(2 !== 0) fast, without a real wall-clock wait.

Full hermes-bots plugin suite: 246/246 pass (47 files). node --check
clean on both changed files.

apps/desktop/src/plugins/hermes-bots/plugin.js

@@ -4290,7 +4290,21 @@ async function runGroupChatRounds(group, members, thread) {
       }
 
       const roomLog = (($groupChats.get()[group] || {}).log || []).filter(e => groupThreadOf(e) === thread)
+      // Exclude members the harvest pass just above confirmed are STILL
+      // running (their stranded marker survived harvest because
+      // state.inflight/running was true). Re-selecting one here would
+      // prompt.submit into their live session — the gateway's default busy
+      // policy redirects or hard-interrupts that turn (tui_gateway's
+      // _handle_busy_submit), killing exactly the long-running work this
+      // stranded/harvest mechanism exists to protect. Skip them; the next
+      // harvest pass picks the reply up once it actually lands. A marker's
+      // mere presence means "still stranded" (harvestStrandedGroupReply
+      // deletes it once the member is confirmed done/dead) — presence, not
+      // value shape, since markers are a bare number pre-thread or
+      // {before, thread} post-thread.
+      const strandedNow = ($groupChats.get()[group] || {}).stranded || {}
       const responders = rotateGroupSpeakers(resolveGroupResponders(roomLog, members), round)
+        .filter(member => !Object.prototype.hasOwnProperty.call(strandedNow, groupMemberKey(member)))
       let spokeThisRound = 0
 
       for (const member of responders) {

apps/desktop/src/plugins/hermes-bots/tests/group-chat.test.mjs

@@ -8,7 +8,7 @@ const pluginSource = readFileSync(new URL('../plugin.js', import.meta.url), 'utf
 /** Load the plugin in a vm with a scripted cli.exec so member turns are
  *  deterministic. `turnScript(profile, prompt)` returns the member's reply
  *  text (or throws to simulate a failed turn). */
-function load(turnScript) {
+function load(turnScript, { busyUntilResumeCall } = {}) {
   const values = new Map()
   const atom = initial => {
     const slot = { get: () => values.get(slot), set: value => values.set(slot, value) }
@@ -20,6 +20,12 @@ function load(turnScript) {
   const runtimeToStored = new Map()
   const titleToStored = new Map()
   let sessionSequence = 0
+  // busyUntilResumeCall[profile] = N: this profile's session.resume reports
+  // inflight/running for its first N calls, then flips to done — simulating
+  // a turn that is genuinely still running when harvested, without an
+  // unbounded real-time wait if a caller polls it (used to exercise the
+  // stranded/busy-responder guard).
+  const resumeCallCounts = new Map()
 
   const resolveSession = (profile, target) => {
     const stored = runtimeToStored.get(target) || (sessions.has(target) ? target : titleToStored.get(`${profile}::${target}`))
@@ -52,13 +58,18 @@ function load(turnScript) {
           if (!session) {
             throw new Error(`session not found: ${params.session_id}`)
           }
+          const profile = params.profile
+          const seen = (resumeCallCounts.get(profile) || 0) + 1
+          resumeCallCounts.set(profile, seen)
+          const limit = busyUntilResumeCall && busyUntilResumeCall[profile]
+          const busy = Boolean(limit && seen <= limit)
           return {
             session_id: session.runtime,
             session_key: session.stored,
             message_count: session.messages.length,
             messages: [...session.messages],
-            inflight: false,
-            running: false
+            inflight: busy,
+            running: busy
           }
         }
         if (method === 'prompt.submit') {
@@ -565,6 +576,62 @@ test('stranded harvest: a timed-out turn whose reply landed late posts into the
   assert.equal(gc.$groupChats.get().Late.stranded.research, undefined, 'marker consumed')
 })
 
+test('stranded + still busy: the round loop never re-submits into a member whose harvest just confirmed they are still running', async () => {
+  // research is confirmed busy on exactly its first two session.resume
+  // calls — the number of harvest-only touches the FIXED code makes across
+  // two rounds. If the responder guard is missing, research gets re-
+  // selected and picks up two EXTRA resume calls of its own (session
+  // resolution + turn baseline) before its very first post-resubmit poll
+  // — call #4 — which then reports done, so the mutated run still finishes
+  // fast (no real wall-clock wait) while still proving the resubmission
+  // happened.
+  const gc = load(profile => (profile === 'builder' ? 'builder here, all good' : '(pass)'), {
+    busyUntilResumeCall: { research: 2 }
+  })
+
+  // research's session is pre-seeded and resolvable, exactly like the
+  // sibling stranded-harvest test above — its stranded marker is the
+  // pre-thread bare-number shape (still supported: harvestStrandedGroupReply
+  // normalizes both shapes, and presence in `stranded` is what the round-loop
+  // guard checks, not the marker's value shape).
+  gc.sessions.set('sid-research', {
+    stored: 'sid-research',
+    runtime: 'rt-research',
+    profile: 'research',
+    title: 'Group: Grind',
+    messages: []
+  })
+
+  // research is already stranded from an earlier (unmodeled) timeout, and
+  // its session is STILL genuinely running per session.resume. Both members
+  // are mentioned, so without the busy-responder guard both would be
+  // re-selected this round — and resubmitting into research's live session
+  // would trigger the gateway's default busy policy (redirect/hard-
+  // interrupt), destroying the very turn the stranded marker exists to wait
+  // out.
+  gc.updateGroupChat('Grind', r => {
+    r.stranded = { research: 0 }
+    r.sessions = { research: 'sid-research' }
+    r.log = [{ from: { kind: 'user', name: 'You' }, text: '@research @builder status?', at: 1 }]
+    r.watermarks = { 'legacy::research': 0, 'legacy::builder': 0 }
+    return r
+  })
+
+  await gc.runGroupChatRounds('Grind', [{ name: 'research', title: '' }, { name: 'builder', title: '' }], 'legacy')
+
+  assert.equal(
+    gc.calls.filter(c => c.profile === 'research').length,
+    0,
+    'research (still busy) must never receive a new prompt.submit'
+  )
+  assert.equal(
+    gc.$groupChats.get().Grind.stranded.research,
+    0,
+    'marker survives untouched — harvest confirmed research is still running'
+  )
+  assert.equal(gc.calls.filter(c => c.profile === 'builder').length, 1, 'builder (not stranded) still gets its turn')
+})
+
 test('stranded harvest: a late (pass) or no-new-message consumes the marker without posting', async () => {
   const gc = load(() => '(pass)')