From e80642df60033563e4c848ba06be978895414cff Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Mon, 14 Sep 2026 15:46:06 -0700 Subject: [PATCH] fix(bot-mode): keep group follow-ups ordered and late answers visible Serialize room drives through their actual member completion, freeze input watermarks by retained entry identity, and share the same completion path with handoff continuations. Stop discards queued work without releasing an active owner early; rename follows the existing room binding. Observe stranded replies for the hard-cap duration plus grace after the foreground wait, and retain unresolved failures in collapsed Activity. Never automatically retry an ambiguous failed submit within the same drive. Adapted from the queue and boundary approach in #92041 by @enwaiax and harvest-budget approach in #107193 by @Finn763; #106502 by @wadib identified failed-submit watermark consumption. The implementation retains current numeric watermark storage, room lifecycle bindings and serial round limits. Related: #92003, #105247, #100026 --- apps/desktop/e2e/group-turn-integrity.spec.ts | 176 ++++++++++++++++++ .../hermes-bots/group-activity.test.ts | 53 ++---- .../plugins/hermes-bots/group-chat-view.tsx | 21 ++- .../hermes-bots/group-round-members.ts | 136 +++----------- .../plugins/hermes-bots/group-rounds.test.ts | 32 ++++ .../src/plugins/hermes-bots/group-rounds.ts | 104 +++++++---- .../src/plugins/hermes-bots/group-turns.ts | 8 +- package-lock.json | 2 +- website/docs/user-guide/bot-mode.md | 10 + 9 files changed, 352 insertions(+), 190 deletions(-) create mode 100644 apps/desktop/e2e/group-turn-integrity.spec.ts diff --git a/apps/desktop/e2e/group-turn-integrity.spec.ts b/apps/desktop/e2e/group-turn-integrity.spec.ts new file mode 100644 index 0000000000..8f38fbc4d2 --- /dev/null +++ b/apps/desktop/e2e/group-turn-integrity.spec.ts @@ -0,0 +1,176 @@ +import { MOCK_REPLY } from '../../../tests-js/scripts/mock-server' + +import { type MockBackendFixture, setupMockBackend, waitForAppReady } from './fixtures' +import { expect, test } from './test' + +let fixture: MockBackendFixture | null = null + +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() + await expect(page.getByRole('button', { name: 'New bot or group chat' })).toBeVisible() +} + +async function createAgent(page: MockBackendFixture['page'], name: string, title: string): Promise { + await page.getByRole('button', { name: 'New bot or group chat' }).click() + await page.getByRole('menuitem', { name: 'New Bot' }).click() + + const dialog = page.getByRole('dialog', { name: 'New Bot' }) + await dialog.getByPlaceholder('inbox-triage').fill(name) + await dialog.getByPlaceholder('Inbox Triage').fill(title) + await dialog.getByRole('button', { name: 'Create Bot' }).click() + await expect(dialog).toBeHidden({ timeout: 30_000 }) + await expect(page.getByRole('button', { name: new RegExp(`^${title}\\b`) }).first()).toBeVisible({ timeout: 30_000 }) +} + +async function createRoom(page: MockBackendFixture['page']) { + await openBots(page) + await createAgent(page, 'programmer', 'Programmer') + await createAgent(page, 'reviewer', 'Reviewer') + + await page.getByRole('button', { name: 'New bot or group chat' }).click() + await page.getByRole('menuitem', { name: 'New Group Chat' }).click() + + const dialog = page.getByRole('dialog', { name: 'New Group Chat' }) + + for (const title of ['Programmer', 'Reviewer']) { + await dialog.getByText(title, { exact: true }).locator('xpath=ancestor::label').getByRole('checkbox').click() + } + + await dialog.getByRole('textbox', { name: 'Group name' }).fill('Programmer, Reviewer') + await dialog.getByRole('button', { name: 'Create Group (2)' }).click() + + const groupTab = page.getByRole('tab', { name: /Programmer, Reviewer Close/ }) + const groupComposer = page.getByRole('textbox', { name: 'Message Programmer, Reviewer' }).filter({ visible: true }) + await expect(groupTab).toBeVisible({ timeout: 20_000 }) + await expect(groupTab).toHaveAttribute('aria-selected', 'true') + await expect(groupComposer).toBeVisible() + + + return groupComposer +} + +test.beforeEach(async () => { + fixture = await setupMockBackend({ mockServer: { holdFirstCompletionContaining: 'LANE_A_FIRST' } }) + await waitForAppReady(fixture, 120_000) +}) + +test.afterEach(async () => { + await fixture?.cleanup() + fixture = null +}) + +test('group follow-up waits for its active member and retains the reply', async () => { + test.setTimeout(240_000) + const page = fixture!.page + const groupComposer = await createRoom(page) + 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.') + 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 }) + await replyComposer.fill('@programmer LANE_A_FOLLOWUP') + await replyComposer.press('Enter') + await page.waitForTimeout(3000) + const overlapping = fixture!.mock.receivedPrompts.filter(p => p.includes('LANE_A_FOLLOWUP')) + console.log('OVERLAPPING', overlapping) + fixture!.mock.releaseHeldStream() + 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).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() + await page.screenshot({ path: '/tmp/botmode-campaign/lane-a-followup-after.png' }) +}) + +test('quiet group still harvests a late answer after sixty observation ticks', async () => { + test.setTimeout(240_000) + const page = fixture!.page + const groupComposer = await createRoom(page) + + await groupComposer.fill('@programmer LANE_A_FIRST') + await groupComposer.press('Enter') + await fixture!.mock.waitForHeldCompletion() + // Jump only the deadline clock, not WebSocket heartbeats or reconnect timers. + await page.evaluate(() => { + const now = Date.now + Date.now = () => now() + 21 * 60_000 + + const timeout = window.setTimeout.bind(window) + + ;(window as any).__harvestTicks = 0 + window.setTimeout = ((handler: TimerHandler, delay?: number, ...args: any[]) => { + if (delay === 5000) { + return timeout(() => { + (window as any).__harvestTicks++ + + if (typeof handler === 'function') { handler(...args) } + }, 100) + } + + return timeout(handler, delay, ...args) + }) as typeof window.setTimeout + }) + await expect.poll(() => page.evaluate(() => (window as any).__harvestTicks), { timeout: 60000 }).toBeGreaterThanOrEqual(60) + await new Promise(resolve => setTimeout(resolve, 1500)) + 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 page.screenshot({ path: '/tmp/botmode-campaign/lane-a-late-after.png' }) +}) + +test('a rejected member turn stays visible when the room settles', async () => { + test.setTimeout(240_000) + const page = fixture!.page + await page.evaluate(() => { + const send = WebSocket.prototype.send + + WebSocket.prototype.send = function(data) { + const frame = JSON.parse(String(data)) + + 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' } }) }))) + } else { + send.call(this, data) + } + } + }) + const groupComposer = await createRoom(page) + + 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 expect(page.getByRole('button', { name: 'Stop', exact: true })).toHaveCount(0) + await expect(page.getByRole('button', { name: /^Activity/ })).toContainText('Programmer hit an error') + await page.screenshot({ path: '/tmp/botmode-campaign/lane-a-error-after.png' }) +}) + +test('Stop clears a queued follow-up and a direct mention resumes the held member', async () => { + test.setTimeout(240_000) + const page = fixture!.page + const groupComposer = await createRoom(page) + await groupComposer.fill('@programmer LANE_A_FIRST') + await groupComposer.press('Enter') + await fixture!.mock.waitForHeldCompletion() + 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 LANE_A_CANCELLED') + await replyComposer.press('Enter') + await page.getByRole('button', { name: 'Stop', exact: true }).click() + fixture!.mock.releaseHeldStream() + await expect(page.getByRole('button', { name: 'Stop', exact: true })).toHaveCount(0) + await page.waitForTimeout(2000) + expect(fixture!.mock.receivedPrompts.some(p => p.includes('LANE_A_CANCELLED'))).toBe(false) + console.log('Stop: queued inference count = 0; paused status:', await page.getByText(/Paused:/).allTextContents()) + await groupComposer.fill('@programmer LANE_A_RESUME') + await groupComposer.press('Enter') + await expect.poll(() => fixture!.mock.receivedPrompts.some(p => p.includes('LANE_A_RESUME')), { timeout: 60000 }).toBe(true) + await expect(page.getByText(MOCK_REPLY, { exact: true }).first()).toBeVisible() + await page.screenshot({ path: '/tmp/botmode-campaign/lane-a-stop-resume.png' }) +}) diff --git a/apps/desktop/src/plugins/hermes-bots/group-activity.test.ts b/apps/desktop/src/plugins/hermes-bots/group-activity.test.ts index dab490020b..1812365379 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-activity.test.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-activity.test.ts @@ -153,46 +153,25 @@ describe('turn arc', () => { }) describe('epoch scoping', () => { - it('a newer send interrupts the previous run and records cancelled in the CURRENT epoch', async () => { - const gates = new Map; resolve: (value: string) => void }>() - - const gate = (n: number) => { - const existing = gates.get(n) - - if (existing) { - return existing.promise - } - - let resolve!: (value: string) => void - - const promise = new Promise(settle => { - resolve = settle - }) - - gates.set(n, { promise, resolve }) - - return promise - } - - const room = await loadRoom({ turn: ({ n }) => gate(n) }) + it('queues follow-ups without cancelling the active turn or losing its reply delta', async () => { + let release!: (reply: string) => void + const first = new Promise(resolve => { release = resolve }) + const room = await loadRoom({ turn: ({ n }) => n === 1 ? first : '(pass)' }) const member: GroupMember[] = [{ name: 'research', title: '' }] - - room.rounds.sendToGroupChat('Busy', member, 'first ask') + const thread = room.rounds.sendToGroupChat('Busy', member, 'first ask')! await drain(() => room.gateway.calls.length < 1, 50) - room.rounds.sendToGroupChat('Busy', member, 'second ask, supersede') - await drain(() => room.gateway.calls.length < 2, 50) - - gates.get(2)?.resolve('from the new run') - await drain(() => Boolean(room.chat.$groupChats.get().Busy?.running)) - gates.get(1)?.resolve('late from the old run') + const epoch = room.chat.$groupChats.get().Busy.epoch + room.rounds.sendToGroupChat('Busy', member, 'follow-up', thread) await drain(() => false) - - const epoch = room.chat.$groupChats.get().Busy?.epoch || 0 - - expect(feed(room, 'Busy').some(event => event.kind === 'cancelled')).toBe(true) - // The view shows only the current run: every visible event is this epoch. - expect(room.activity.currentGroupActivity('Busy').every(event => (event.epoch || 0) === epoch)).toBe(true) - expect(room.activity.currentGroupActivity('Busy').some(event => event.kind === 'cancelled')).toBe(true) + expect(room.gateway.calls).toHaveLength(1) + expect(room.chat.$groupChats.get().Busy.epoch).toBe(epoch) + release('first reply') + await drain(() => room.gateway.calls.length < 2) + await drain(() => Boolean(room.chat.$groupChats.get().Busy?.running)) + expect(room.gateway.calls).toHaveLength(2) + expect(room.gateway.calls[1].prompt).toMatch(/follow-up[\s\S]*first reply/) + expect(room.gateway.calls[1].prompt).not.toContain('first ask') + expect(feed(room, 'Busy').some(event => event.kind === 'cancelled')).toBe(false) }) it('epoch filtering drops events from a superseded run', async () => { diff --git a/apps/desktop/src/plugins/hermes-bots/group-chat-view.tsx b/apps/desktop/src/plugins/hermes-bots/group-chat-view.tsx index ba24af0d62..dee5bf8ab1 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-chat-view.tsx +++ b/apps/desktop/src/plugins/hermes-bots/group-chat-view.tsx @@ -710,6 +710,23 @@ export function GroupChatWorkspace({ group, members, onBack, visible = true }: G // Events are epoch-tagged, so a superseded run's history drops out of view. const activityEvents: GroupActivityEntry[] = currentGroupActivity(group) const latestActivity = activityEvents.length ? activityEvents[activityEvents.length - 1] : null + // A later "settled" event must not hide a member that failed to answer. + // Successful completion for that member clears its unresolved warning. + const unresolvedFailures = new Map() + + for (const event of activityEvents) { + const key = event.member || '' + + if (event.kind === 'failed' || event.kind === 'timed-out') { + unresolvedFailures.set(key, event) + } else if (event.kind === 'replied' || event.kind === 'passed' || event.kind === 'delivered') { + unresolvedFailures.delete(key) + } + } + + const summaryActivity = !room.running && unresolvedFailures.size + ? [...unresolvedFailures.values()].at(-1)! + : latestActivity // #94570 shell rewired onto the real primitive (#91868/#94569): the button // must stop the ROUND, not just spray per-member interrupts — without the @@ -735,8 +752,8 @@ export function GroupChatWorkspace({ group, members, onBack, visible = true }: G > {b.group.activity} - {latestActivity ? ( - {`${groupActivityLabel(latestActivity)} · ${relativeTime(latestActivity.at)}`} + {summaryActivity ? ( + {`${groupActivityLabel(summaryActivity)} · ${relativeTime(summaryActivity.at)}`} ) : null} {room.running ? ( diff --git a/apps/desktop/src/plugins/hermes-bots/group-round-members.ts b/apps/desktop/src/plugins/hermes-bots/group-round-members.ts index 51037b227f..08207617e3 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-round-members.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-round-members.ts @@ -23,6 +23,7 @@ export interface GroupRoundMemberContext { startEpoch: number binding: { isLive(): boolean } isCurrent(): boolean + failedMembers?: Set } /** #93129: a held member's skip must consume its delta exactly once — @@ -129,6 +130,11 @@ export async function runGroupRoundMember( member: GroupMember ): Promise { const { thread, startEpoch, binding } = context + + if (context.failedMembers?.has(groupMemberKey(member))) { + return false + } + const prepared = prepareGroupRoundMember(context, member) if (!prepared) { @@ -136,10 +142,13 @@ export async function runGroupRoundMember( } const { room, markKey, prompt, deltaImages } = prepared + const anchorId = room.log.at(-1)?.id ?? null let reply: null | string = null + let accepted = false try { reply = await runVisibleMemberTurn(context, member, prompt, deltaImages) + accepted = true // Needs-attention hook (#93091 item 3): a turn that produced a real // reply (or an explicit pass) is a good turn — clear the badge. @@ -165,6 +174,7 @@ export async function runGroupRoundMember( : {}) }) noteBotAttention(groupMemberKey(member), reason || error?.message || error) + context.failedMembers?.add(groupMemberKey(member)) reply = null // a failed turn is a pass, never a room error } @@ -188,7 +198,6 @@ export async function runGroupRoundMember( } const epochNow = roomNow.epoch || 0 - const anchorId = room.log.length ? room.log[room.log.length - 1].id : null const anchorIdx = anchorId === null ? -1 : roomNow.log.findIndex((e: GroupMessage) => e.id === anchorId) // Anchor trimmed away ⇒ every pre-turn entry was dropped, so every // surviving entry is newer — scanning the whole log stays exact. @@ -198,7 +207,10 @@ export async function runGroupRoundMember( (e: GroupMessage) => e.from?.kind === 'user' && groupThreadOf(e) === thread ) - if (!shouldCommitMemberTurn(startEpoch, epochNow, newerUserEntryInThread)) { + if ( + (epochNow !== startEpoch && roomNow.holds?.[groupMemberKey(member)]) || + !shouldCommitMemberTurn(startEpoch, epochNow, newerUserEntryInThread) + ) { recordGroupActivity(context.group, { kind: 'cancelled', member: member.name, @@ -208,12 +220,16 @@ export async function runGroupRoundMember( return null } - // The member has now seen everything up to the pre-reply log length. - updateGroupChat(context.group, (r: GroupChatRoom) => { - r.watermarks[markKey] = r.log.length + // Resolve the frozen submit boundary against the retained log. If it was + // trimmed away, every surviving entry is still unseen. Throws do not + // acknowledge input, and a timed-out turn keeps its submitted boundary. + if (accepted) { + updateGroupChat(context.group, (r: GroupChatRoom) => { + r.watermarks[markKey] = anchorIdx + 1 - return r - }) + return r + }) + } if (reply !== null && !isGroupPassText(reply)) { appendGroupChatEntry( @@ -230,107 +246,11 @@ export async function runGroupRoundMember( reply, thread ) - // Its own message counts as seen too. + // A reply cannot acknowledge user entries that arrived during inference. updateGroupChat(context.group, (r: GroupChatRoom) => { - r.watermarks[markKey] = r.log.length - - return r - }) - - return true - } - - return false -} - -async function runGroupContinuationMember( - context: GroupRoundMemberContext, - member: GroupMember -): Promise { - const { members, thread, binding, isCurrent } = context - - const room = $groupChats.get()[context.group] || { - log: [], - watermarks: {} - } - - const memberKey = groupMemberKey(member) - const markKey = `${thread}::${memberKey}` - const seen = room.watermarks[markKey] || 0 - const delta = room.log.slice(seen).filter((e: GroupMessage) => groupThreadOf(e) === thread) - - // A cited member always has delta here (the citing reply IS in - // its tail); skip defensively anyway so an empty prompt never - // fires. - if (!delta.length) { - return false - } - - const heldEntry = (room.holds || {})[memberKey] - - if (heldEntry) { - return false // holds still apply to continuation turns (#93129) - } - - const prompt = buildGroupChatTurnPrompt({ - groupName: context.group, - members, - viewer: member, - // The continuation prompt centers on what the member missed: - // everything since its watermark, which includes the reply - // that cites it. - deltaLines: delta.slice(-GROUP_CHAT_HISTORY_LIMIT).map((e: GroupMessage) => formatGroupChatLine(e, member)) - }) - - let continuationReply: null | string = null - - try { - continuationReply = await runVisibleMemberTurn(context, member, prompt) - - if (continuationReply !== null) { - clearBotAttention(memberKey) - } - } catch (error: any) { - if (!binding.isLive()) { - return null - } - - recordGroupActivity(context.group, { - kind: 'failed', - member: member.name, - thread - }) - noteBotAttention(memberKey, error?.message || error) - continuationReply = null - } - - if (!isCurrent()) { - return null - } - - updateGroupChat(context.group, (r: GroupChatRoom) => { - r.watermarks[markKey] = r.log.length - - return r - }) - - if (continuationReply !== null && !isGroupPassText(continuationReply)) { - appendGroupChatEntry( - context.group, - { - kind: 'member', - name: member.name, - ...(member.remoteSource - ? { - source: member.connectionLabel || member.connectionId - } - : {}) - }, - continuationReply, - thread - ) - updateGroupChat(context.group, (r: GroupChatRoom) => { - r.watermarks[markKey] = r.log.length + if (r.watermarks[markKey] === r.log.length - 1) { + r.watermarks[markKey] = r.log.length + } return r }) @@ -365,7 +285,7 @@ export async function runGroupContinuationMembers( break } - const result = await runGroupContinuationMember(context, member) + const result = await runGroupRoundMember(context, member) if (result === null) { return null 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 a3d164937a..168bbd6c88 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-rounds.test.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-rounds.test.ts @@ -230,6 +230,19 @@ describe('round lifecycle', () => { expect(posted.length).toBeLessThanOrEqual(room.chat.GROUP_CHAT_MAX_MESSAGES) }) + it('does not retry ambiguous member admission in later rounds or continuations', async () => { + const room = await loadRoom({ turn: ({ profile }) => { + if (profile === 'builder') { throw new Error('Ambiguous admission failure') } + + return '@builder please investigate' + } }) + + room.rounds.sendToGroupChat('Failure', MEMBERS.slice(0, 2), '@research start') + await settle(room, 'Failure') + expect(room.gateway.calls.filter(call => call.profile === 'builder')).toHaveLength(1) + expect(Object.keys(room.chat.$groupChats.get().Failure.watermarks).some(key => key.endsWith('::builder'))).toBe(false) + }) + it('treats a failed member turn as a pass, not a room error', async () => { const room = await loadRoom({ turn: ({ profile }) => { @@ -292,6 +305,25 @@ describe('round lifecycle', () => { }) describe('per-member delta', () => { + it('retained-log trimming cannot acknowledge messages appended during inference', async () => { + let release!: (reply: string) => void + const held = new Promise(resolve => { release = resolve }) + const room = await loadRoom({ turn: ({ n }) => n === 1 ? held : '(pass)' }) + const members = [MEMBERS[0]] + const thread = room.rounds.sendToGroupChat('Trim', members, 'delivered')! + await drain(() => room.gateway.calls.length < 1) + + for (let i = 0; i < 100; i++) { + room.chat.appendGroupChatEntry('Trim', { kind: 'user', name: 'You' }, `unseen-${i}`, thread) + } + + release('(pass)') + await settle(room, 'Trim') + expect(room.chat.$groupChats.get().Trim.watermarks[`${thread}::research`]).toBe(0) + await room.rounds.runGroupChatRounds('Trim', members, thread) + expect(room.gateway.calls.at(-1)?.prompt).toContain('unseen-99') + }) + it('feeds a second send only the NEW messages', async () => { const room = await loadRoom() const member: GroupMember[] = [{ name: 'research', title: '' }] diff --git a/apps/desktop/src/plugins/hermes-bots/group-rounds.ts b/apps/desktop/src/plugins/hermes-bots/group-rounds.ts index ab2edf76a8..5013cc10cc 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-rounds.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-rounds.ts @@ -12,6 +12,7 @@ import { GROUP_CHAT_MAX_CONTINUATIONS, GROUP_CHAT_MAX_MESSAGES, GROUP_CHAT_MAX_ROUNDS, + groupChatRoomKey, groupThreadOf, mintGroupThreadId, updateGroupChat @@ -20,7 +21,7 @@ import type { GroupChatRoom, GroupHoldStamp } from './group-chat' import { durableGroupChatMembers, followGroupChat, groupMemberKey } from './group-membership' import { runGroupContinuationMembers, runGroupRoundMember } from './group-round-members' import { rejectGroupSlashCommand } from './group-slash' -import { harvestStrandedGroupReply } from './group-turns' +import { GROUP_TURN_HARD_CAP_MS, harvestStrandedGroupReply } from './group-turns' import { requestForBot } from './routing' import type { Attachment, GroupMember, GroupMessage } from './types' @@ -353,6 +354,7 @@ export async function stopGroupThread(group: string, thread: null | string, memb const room = $groupChats.get()[group] || {} const roster = Array.isArray(members) && members.length ? members : room.members || [] const onTurn = room.turn || null + groupChatDrives.get(groupChatRoomKey(group, room))?.pending.clear() const stamp: GroupHoldStamp = { at: Date.now(), @@ -412,8 +414,8 @@ export async function stopGroupThread(group: string, thread: null | string, memb } /** Drive one bounded round-robin turn for ONE THREAD. Serial — one member at - * a time. A newer user send bumps the room epoch; this loop notices at the - * next member boundary, bails, and the newest send's own loop takes over. + * a time. User follow-ups queue behind this drive; Stop invalidates its + * 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) { @@ -431,6 +433,7 @@ export async function runGroupChatRounds(group: string, members: GroupMember[], members, thread, startEpoch, + failedMembers: new Set(), binding, isCurrent } @@ -592,7 +595,8 @@ export async function runGroupChatRounds(group: string, members: GroupMember[], } /** Bounded background harvest for members whose replies outlived the turn - * loop. Polls every 5s for up to 5 minutes; stops early when nothing is + * loop. Watches for a further hard-cap duration plus a minute of grace + * after foreground polling ends; stops early when nothing is * stranded, a new loop takes the room over (it harvests on its own), or the * room record disappears (disband). */ async function harvestStrandedUntilSettled(group: string, members: GroupMember[], thread: string) { @@ -602,7 +606,7 @@ async function harvestStrandedUntilSettled(group: string, members: GroupMember[] try { const HARVEST_INTERVAL_MS = 5000 - const HARVEST_MAX_TRIES = 60 + const HARVEST_MAX_TRIES = Math.ceil((GROUP_TURN_HARD_CAP_MS + 60000) / HARVEST_INTERVAL_MS) for (let attempt = 0; attempt < HARVEST_MAX_TRIES; attempt++) { await new Promise(resolve => window.setTimeout(resolve, HARVEST_INTERVAL_MS)) @@ -649,8 +653,8 @@ async function harvestStrandedUntilSettled(group: string, members: GroupMember[] /** User send into a group room. `thread` continues that thread (its reply * box); omitted/null mints a NEW thread — the main composer's Slack shape. - * Appends, bumps the room epoch (supersedes any running loop at its next - * member boundary), and starts the turn drive for the target thread. + * Appends and queues the target thread behind the active room drive, + * without redirecting a member whose inference is still in flight. * Returns the thread id the message landed in. */ export function sendToGroupChat( group: string, @@ -696,10 +700,7 @@ export function sendToGroupChat( attached ) - const wasRunning = ($groupChats.get()[group] || {}).running === true updateGroupChat(group, (room: GroupChatRoom) => { - room.epoch = (room.epoch || 0) + 1 - room.running = true // #93129: user text is the ONLY input that changes member holds. An // explicit "stop @member" sets a sticky hold; "@member resume" (or // @all resume, or any direct non-stop mention of the held member) @@ -724,36 +725,59 @@ export function sendToGroupChat( thread: target }) - const binding = followGroupChat(group, name => { - group = name - }) - - const drive = () => { - if (!binding.isLive()) { - binding.dispose() - - return - } - - void runGroupChatRounds(group, members, target) - .catch(() => { - if (binding.isLive()) { - updateGroupChat(group, (r: GroupChatRoom) => { - r.running = false - - return r - }) - } - }) - .finally(binding.dispose) - } - - if (!wasRunning) { - drive() - } else { - // Preserve the existing newer-send handoff delay, without pinning its name. - setTimeout(drive, 250) - } + queueGroupChatDrive(group, members, target) return target } + +interface GroupChatDrive { + pending: Map + binding: ReturnType +} + +// Keep the owner until its awaited member releases, even after Stop. A +// rename follows the room identity; disband retires the binding permanently. +const groupChatDrives = new Map() + +function queueGroupChatDrive(group: string, members: GroupMember[], thread: string) { + let key = groupChatRoomKey(group, $groupChats.get()[group]) + const active = groupChatDrives.get(key) + + if (active?.binding.isLive()) { + active.pending.set(thread, members) + + return + } + + const binding = followGroupChat(group, name => { + groupChatDrives.delete(key) + group = name + key = groupChatRoomKey(group, $groupChats.get()[group]) + groupChatDrives.set(key, drive) + }) + + const drive: GroupChatDrive = { pending: new Map([[thread, members]]), binding } + groupChatDrives.set(key, drive) + + 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) + } + } catch { + if (binding.isLive()) { + recordGroupActivity(group, { kind: 'failed', member: null, thread }) + updateGroupChat(group, room => ({ ...room, running: false, turn: null })) + } + } finally { + binding.dispose() + + if (groupChatDrives.get(key) === drive) { + groupChatDrives.delete(key) + } + } + })() +} diff --git a/apps/desktop/src/plugins/hermes-bots/group-turns.ts b/apps/desktop/src/plugins/hermes-bots/group-turns.ts index ee5de72f1b..86acc313a8 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-turns.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-turns.ts @@ -432,7 +432,7 @@ async function submitGroupTurnPrompt( // timeout alone silently dropped long real turns: a 7-minute research run // timed out at 3 minutes, read as a pass, and its finished result never // reached the room (db's Aug 2026 report). -const GROUP_TURN_HARD_CAP_MS = 20 * 60000 +export const GROUP_TURN_HARD_CAP_MS = 20 * 60000 /** Mirror a member's pending prompt — clarify question OR command approval — * from its resume snapshot into the room store, keyed @@ -1069,7 +1069,11 @@ export async function harvestStrandedGroupReply(group: string, member: GroupMemb strandedThread ) updateGroupChat(group, (r: GroupChatRoom) => { - r.watermarks[`${strandedThread}::${memberKey}`] = r.log.length + const markKey = `${strandedThread}::${memberKey}` + + if (r.watermarks[markKey] === r.log.length - 1) { + r.watermarks[markKey] = r.log.length + } return r }) diff --git a/package-lock.json b/package-lock.json index 10a4728ba9..62e934e325 100644 --- a/package-lock.json +++ b/package-lock.json @@ -31,7 +31,7 @@ }, "apps/bootstrap-installer": { "name": "@hermes/bootstrap-installer", - "version": "0.0.1", + "version": "0.21.1", "dependencies": { "@nous-research/ui": "0.18.2", "@tailwindcss/typography": "0.5.20", diff --git a/website/docs/user-guide/bot-mode.md b/website/docs/user-guide/bot-mode.md index 3c90a8fa36..b99b5113ac 100644 --- a/website/docs/user-guide/bot-mode.md +++ b/website/docs/user-guide/bot-mode.md @@ -86,6 +86,16 @@ Routines are plain [Hermes cron jobs](./features/cron.md) namespaced `[bot: