diff --git a/apps/desktop/e2e/group-turn-integrity.spec.ts b/apps/desktop/e2e/group-turn-integrity.spec.ts index 8f38fbc4d2..61129949b8 100644 --- a/apps/desktop/e2e/group-turn-integrity.spec.ts +++ b/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 { 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' }) }) diff --git a/apps/desktop/src/plugins/hermes-bots/group-rounds.test.ts b/apps/desktop/src/plugins/hermes-bots/group-rounds.test.ts index 168bbd6c88..3a237d363d 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-rounds.test.ts +++ b/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((_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 }) => { diff --git a/apps/desktop/src/plugins/hermes-bots/group-rounds.ts b/apps/desktop/src/plugins/hermes-bots/group-rounds.ts index 5013cc10cc..02ee476d17 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-rounds.ts +++ b/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()) { 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(), + failedMembers, binding, isCurrent } @@ -731,6 +731,7 @@ export function sendToGroupChat( } interface GroupChatDrive { + failedMembers: Set pending: Map binding: ReturnType } @@ -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()) { diff --git a/tests-js/scripts/mock-server.ts b/tests-js/scripts/mock-server.ts index f768a2e084..cf9b34e879 100644 --- a/tests-js/scripts/mock-server.ts +++ b/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 { + streamTextResponse(res, model, reply, holdThisStream || holdThisCompletion ? () => { if (holdThisCompletion) { heldCompletionCount++ } @@ -711,9 +716,9 @@ export function startMockServer(options: MockServerOptions = {}): Promise nonStreamingTextResponse(res, model, MOCK_REPLY)) + void heldStreamReleased.then(() => nonStreamingTextResponse(res, model, reply)) } else { - nonStreamingTextResponse(res, model, MOCK_REPLY) + nonStreamingTextResponse(res, model, reply) } } })