fix(bot-mode): do not retry failures from already queued sends

Keep failed-member exclusion across the room queue, rather than resetting
it per pending thread. A new user action after failure still permits a new
attempt. Share the drain activity epoch so a skipped queued thread cannot
hide the preceding member failure.

Proven red in real Electron: hold transport refusal, enqueue same-thread
and cross-thread sends, then release; old head submits three times, fixed
head once. Strengthen follow-up evidence with distinct provider replies,
exact public log order/count, and per-input inference counts.

apps/desktop/e2e/group-turn-integrity.spec.ts

@@ -3,8 +3,20 @@ import { MOCK_REPLY } from '../../../tests-js/scripts/mock-server'
 import { type MockBackendFixture, setupMockBackend, waitForAppReady } from './fixtures'
 import { expect, test } from './test'
 
+const FIRST_REPLY = 'FIRST_REPLY: initial work completed'
+const FOLLOWUP_REPLY = 'FOLLOWUP_REPLY: subsequent work completed'
 let fixture: MockBackendFixture | null = null
 
+async function publicLog(page: MockBackendFixture['page']) {
+  return page.evaluate(() => {
+    const rooms = JSON.parse(localStorage.getItem('hermes.plugin.hermes-bots.group-chats') || '{}')
+
+    return (rooms['Programmer, Reviewer']?.log || []).map((entry: any) => ({
+      from: entry.from.name, text: entry.text, thread: entry.thread
+    })) as { from: string; text: string; thread: string }[]
+  })
+}
+
 async function openBots(page: MockBackendFixture['page']): Promise<void> {
   const tab = page.getByRole('button', { name: 'Bots', exact: true }).or(page.getByRole('tab', { name: 'Bots', exact: true })).first()
   await tab.click()
@@ -51,7 +63,10 @@ async function createRoom(page: MockBackendFixture['page']) {
 }
 
 test.beforeEach(async () => {
-  fixture = await setupMockBackend({ mockServer: { holdFirstCompletionContaining: 'LANE_A_FIRST' } })
+  fixture = await setupMockBackend({ mockServer: {
+    holdFirstCompletionContaining: 'LANE_A_FIRST',
+    replyForPrompt: prompt => prompt.includes('LANE_A_FOLLOWUP') ? FOLLOWUP_REPLY : prompt.includes('LANE_A_FIRST') ? FIRST_REPLY : MOCK_REPLY
+  } })
   await waitForAppReady(fixture, 120_000)
 })
 
@@ -67,7 +82,7 @@ test('group follow-up waits for its active member and retains the reply', async
   await groupComposer.fill('@programmer LANE_A_FIRST')
   await groupComposer.press('Enter')
   await fixture!.mock.waitForHeldCompletion()
-  console.log('Held first inference; no second submit is allowed until release.')
+  console.log('PRODUCTION CLOCK: first inference held; no second submit before provider release.')
   await page.screenshot({ path: '/tmp/botmode-campaign/lane-a-held.png' })
   await page.getByRole('button', { name: 'Reply in thread', exact: true }).click()
   const replyComposer = page.getByRole('textbox', { name: 'Reply in thread', exact: true })
@@ -80,10 +95,21 @@ test('group follow-up waits for its active member and retains the reply', async
   expect(overlapping).toHaveLength(0)
   await expect.poll(() => fixture!.mock.receivedPrompts.some(p => p.includes('LANE_A_FOLLOWUP')), { timeout: 60000 }).toBe(true)
   const delivered = fixture!.mock.receivedPrompts.find(p => p.includes('LANE_A_FOLLOWUP'))!
-  expect(delivered).toContain(MOCK_REPLY)
+  expect(delivered).toContain(FIRST_REPLY)
   expect(delivered).not.toContain('LANE_A_FIRST')
-  expect(delivered.indexOf('LANE_A_FOLLOWUP')).toBeLessThan(delivered.indexOf(MOCK_REPLY))
-  await expect(page.getByText(MOCK_REPLY, { exact: true }).first()).toBeVisible()
+  expect(delivered.indexOf('LANE_A_FOLLOWUP')).toBeLessThan(delivered.indexOf(FIRST_REPLY))
+  await expect(page.getByText(FIRST_REPLY, { exact: true }).filter({ visible: true })).toHaveCount(1)
+  await expect(page.getByText(FOLLOWUP_REPLY, { exact: true }).filter({ visible: true })).toHaveCount(1)
+  await expect(page.getByRole('button', { name: 'Stop', exact: true })).toHaveCount(0)
+  const log = await publicLog(page)
+  expect(log.map(({ from, text }) => [from, text])).toEqual([
+    ['You', '@programmer LANE_A_FIRST'], ['You', '@programmer LANE_A_FOLLOWUP'],
+    ['programmer', FIRST_REPLY], ['programmer', FOLLOWUP_REPLY]
+  ])
+  expect(new Set(log.map(entry => entry.thread)).size).toBe(1)
+  expect(fixture!.mock.receivedPrompts.filter(p => p.includes('LANE_A_FIRST'))).toHaveLength(1)
+  expect(fixture!.mock.receivedPrompts.filter(p => p.includes('LANE_A_FOLLOWUP'))).toHaveLength(1)
+  console.log('PRODUCTION CLOCK: released; exact public log', JSON.stringify(log))
   await page.screenshot({ path: '/tmp/botmode-campaign/lane-a-followup-after.png' })
 })
 
@@ -120,7 +146,10 @@ test('quiet group still harvests a late answer after sixty observation ticks', a
   console.log('Harvest ticks before release:', await page.evaluate(() => (window as any).__harvestTicks))
   console.log('Activity before release:', await page.getByRole('button', { name: /^Activity/ }).textContent())
   fixture!.mock.releaseHeldStream()
-  await expect(page.getByText(MOCK_REPLY, { exact: true }).first()).toBeVisible({ timeout: 5000 })
+  await expect(page.getByText(FIRST_REPLY, { exact: true }).filter({ visible: true })).toHaveCount(1, { timeout: 5000 })
+  await page.waitForTimeout(1000)
+  expect((await publicLog(page)).filter(entry => entry.from === 'programmer').map(entry => entry.text)).toEqual([FIRST_REPLY])
+  console.log('ACCELERATED DEADLINE/OBSERVATION CLOCK ONLY: exact late public log', JSON.stringify(await publicLog(page)))
   await page.screenshot({ path: '/tmp/botmode-campaign/lane-a-late-after.png' })
 })
 
@@ -135,7 +164,9 @@ test('a rejected member turn stays visible when the room settles', async () => {
 
       if (frame.method === 'prompt.submit' && JSON.stringify(frame.params).includes('LANE_A_FAILURE')) {
         (window as any).__rejected = ((window as any).__rejected || 0) + 1
-        queueMicrotask(() => this.dispatchEvent(new MessageEvent('message', { data: JSON.stringify({ jsonrpc: '2.0', id: frame.id, error: { code: 4003, message: 'Controlled member admission refusal' } }) })))
+        const reject = () => this.dispatchEvent(new MessageEvent('message', { data: JSON.stringify({ jsonrpc: '2.0', id: frame.id, error: { code: 4003, message: 'Controlled member admission refusal' } }) }))
+
+        if ((window as any).__rejected === 1) { (window as any).__releaseRefusal = reject } else { queueMicrotask(reject) }
       } else {
         send.call(this, data)
       }
@@ -146,7 +177,18 @@ test('a rejected member turn stays visible when the room settles', async () => {
   await groupComposer.fill('@programmer LANE_A_FAILURE')
   await groupComposer.press('Enter')
   await expect.poll(() => page.evaluate(() => (window as any).__rejected), { timeout: 30000 }).toBe(1)
+  await page.getByRole('button', { name: 'Reply in thread', exact: true }).click()
+  const replyComposer = page.getByRole('textbox', { name: 'Reply in thread', exact: true })
+  await replyComposer.fill('@programmer prequeued LANE_A_FAILURE')
+  await replyComposer.press('Enter')
+  await groupComposer.fill('@programmer cross-thread LANE_A_FAILURE')
+  await groupComposer.press('Enter')
+  console.log('PRODUCTION CLOCK / CONTROLLED TRANSPORT: two sends queued before refusal released')
+  await page.evaluate(() => (window as any).__releaseRefusal())
   await expect(page.getByRole('button', { name: 'Stop', exact: true })).toHaveCount(0)
+  await page.waitForTimeout(2000)
+  expect(await page.evaluate(() => (window as any).__rejected)).toBe(1)
+  expect((await publicLog(page)).filter(entry => entry.from !== 'You')).toEqual([])
   await expect(page.getByRole('button', { name: /^Activity/ })).toContainText('Programmer hit an error')
   await page.screenshot({ path: '/tmp/botmode-campaign/lane-a-error-after.png' })
 })

apps/desktop/src/plugins/hermes-bots/group-rounds.test.ts

@@ -243,6 +243,27 @@ describe('round lifecycle', () => {
     expect(Object.keys(room.chat.$groupChats.get().Failure.watermarks).some(key => key.endsWith('::builder'))).toBe(false)
   })
 
+  it('does not retry an ambiguous submit from prequeued same-thread or cross-thread sends', async () => {
+    let reject!: (error: Error) => void
+    const held = new Promise<string>((_resolve, fail) => { reject = fail })
+    const room = await loadRoom({ turn: ({ n }) => n === 1 ? held : '(pass)' })
+    const members = [MEMBERS[0]]
+    const thread = room.rounds.sendToGroupChat('Failure', members, 'first')!
+    await drain(() => room.gateway.calls.length < 1)
+    room.rounds.sendToGroupChat('Failure', members, 'queued same-thread', thread)
+    room.rounds.sendToGroupChat('Failure', members, 'queued other-thread')
+    reject(new Error('Ambiguous admission failure'))
+    await settle(room, 'Failure')
+    await drain(() => false)
+    expect(room.gateway.calls).toHaveLength(1)
+    expect(room.chat.$groupChats.get().Failure.watermarks).toEqual({})
+
+    room.rounds.sendToGroupChat('Failure', members, '@research explicitly retry', thread)
+    await settle(room, 'Failure')
+    expect(room.gateway.calls).toHaveLength(2)
+    expect(room.gateway.calls[1].prompt).toMatch(/first[\s\S]*queued same-thread[\s\S]*explicitly retry/)
+  })
+
   it('treats a failed member turn as a pass, not a room error', async () => {
     const room = await loadRoom({
       turn: ({ profile }) => {

apps/desktop/src/plugins/hermes-bots/group-rounds.ts

@@ -418,7 +418,7 @@ export async function stopGroupThread(group: string, thread: null | string, memb
  *  epoch and discards queued continuations.
  *  Watermarks are per thread+member (`${thread}::${memberKey}`), so parallel
  *  topics never eat each other's deltas. */
-export async function runGroupChatRounds(group: string, members: GroupMember[], thread: string) {
+export async function runGroupChatRounds(group: string, members: GroupMember[], thread: string, failedMembers = new Set<string>()) {
   const binding = followGroupChat(group, name => {
     group = name
   })
@@ -433,7 +433,7 @@ export async function runGroupChatRounds(group: string, members: GroupMember[],
     members,
     thread,
     startEpoch,
-    failedMembers: new Set<string>(),
+    failedMembers,
     binding,
     isCurrent
   }
@@ -731,6 +731,7 @@ export function sendToGroupChat(
 }
 
 interface GroupChatDrive {
+  failedMembers: Set<string>
   pending: Map<string, GroupMember[]>
   binding: ReturnType<typeof followGroupChat>
 }
@@ -744,6 +745,8 @@ function queueGroupChatDrive(group: string, members: GroupMember[], thread: stri
   const active = groupChatDrives.get(key)
 
   if (active?.binding.isLive()) {
+    // Only a new user action AFTER failure authorizes another attempt.
+    active.failedMembers.clear()
     active.pending.set(thread, members)
 
     return
@@ -756,16 +759,19 @@ function queueGroupChatDrive(group: string, members: GroupMember[], thread: stri
     groupChatDrives.set(key, drive)
   })
 
-  const drive: GroupChatDrive = { pending: new Map([[thread, members]]), binding }
+  const drive: GroupChatDrive = { pending: new Map([[thread, members]]), failedMembers: new Set(), binding }
   groupChatDrives.set(key, drive)
+  // Queued threads share the activity epoch, so draining one cannot hide
+  // unresolved failures from the preceding thread. Stop still invalidates it.
+  updateGroupChat(group, room => ({ ...room, epoch: (room.epoch || 0) + 1 }))
 
   void (async () => {
     try {
       while (binding.isLive() && drive.pending.size) {
         const [nextThread, nextMembers] = drive.pending.entries().next().value!
         drive.pending.delete(nextThread)
-        updateGroupChat(group, room => ({ ...room, epoch: (room.epoch || 0) + 1, running: true }))
-        await runGroupChatRounds(group, nextMembers, nextThread)
+        updateGroupChat(group, room => ({ ...room, running: true }))
+        await runGroupChatRounds(group, nextMembers, nextThread, drive.failedMembers)
       }
     } catch {
       if (binding.isLive()) {

tests-js/scripts/mock-server.ts

@@ -31,6 +31,9 @@ import { pathToFileURL } from 'node:url'
 export const MOCK_REPLY = 'Hello from the mock inference server! The full boot chain is working.'
 
 export interface MockServerOptions {
+  /** Choose distinct replies from the latest input without replaying history. */
+  replyForPrompt?: (prompt: string) => string
+
   /** Pause the matching stream after its first token for session-switch E2E coverage. */
   holdFirstStreamForPrompt?: string
 /** Pause the first completion whose request JSON contains this text. */
@@ -692,13 +695,15 @@ export function startMockServer(options: MockServerOptions = {}): Promise<MockSe
             return
           }
 
+          const reply = options.replyForPrompt?.(userText) ?? MOCK_REPLY
+
           if (stream) {
             const holdThisStream = Boolean(
               options.holdFirstStreamForPrompt && typeof lastUserMessage?.content === 'string' &&
                 lastUserMessage.content.includes(options.holdFirstStreamForPrompt),
             )
 
-            streamTextResponse(res, model, MOCK_REPLY, holdThisStream || holdThisCompletion ? () => {
+            streamTextResponse(res, model, reply, holdThisStream || holdThisCompletion ? () => {
               if (holdThisCompletion) {
                 heldCompletionCount++
               }
@@ -711,9 +716,9 @@ export function startMockServer(options: MockServerOptions = {}): Promise<MockSe
             if (holdThisCompletion) {
               heldCompletionCount++
               resolveHeldStreamStarted?.()
-              void heldStreamReleased.then(() => nonStreamingTextResponse(res, model, MOCK_REPLY))
+              void heldStreamReleased.then(() => nonStreamingTextResponse(res, model, reply))
             } else {
-              nonStreamingTextResponse(res, model, MOCK_REPLY)
+              nonStreamingTextResponse(res, model, reply)
             }
           }
         })