From fb5023950e9d979b4a097b031d46f1cb1ca2c69a Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 01:57:44 -0700 Subject: [PATCH] perf(desktop): group chat rooms answer in the time of one bot, not the sum of all Bot Mode group rooms were slow by construction: the round engine ran every member's turn one after another, and each turn found out its bot had finished by re-reading session.resume on a fixed 2s timer. A 4-bot room paid 4 x (model latency + up to 2s) per round, serially. - group-rounds: members of a round now take their turns concurrently (Promise.all). Rounds stay serial so bots still build on each other's replies. Each member's delta is computed at its own turn start and its watermark advances only to the pre-turn log length, so sibling replies that land while it thinks are delivered next round exactly once; a member's own replies are excluded from its delta by author (they are already in its session). Message cap enforced per round; stop path interrupts every member mid-turn (room.turn -> room.turns map). - group-turns: the poll wakes on the member session's terminal frame (message.complete / error via host.onEvent), then re-checks at 250ms until session.running clears. The timer poll stays as a 5s backstop for hosts without the event tap. Feature-detected; node test harness unaffected. - group-chat-view: "X is thinking..." lists every member mid-turn. - docs: bot-mode.md describes concurrent rounds + push-woken replies. Live A/B (real tui_gateway over WS, 4 members, one round, same model): serial+2s poll 35.0s -> concurrent+push 8.6s; every turn woke on the event. Refs #92760 --- .../plugins/hermes-bots/group-chat-view.tsx | 8 +- .../src/plugins/hermes-bots/group-chat.ts | 5 +- .../plugins/hermes-bots/group-rounds.test.ts | 49 +- .../src/plugins/hermes-bots/group-rounds.ts | 656 +++++++++--------- .../plugins/hermes-bots/group-turns.test.ts | 39 ++ .../src/plugins/hermes-bots/group-turns.ts | 88 ++- website/docs/user-guide/bot-mode.md | 3 +- 7 files changed, 507 insertions(+), 341 deletions(-) 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 e9749dc3c3..65b0858573 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-chat-view.tsx +++ b/apps/desktop/src/plugins/hermes-bots/group-chat-view.tsx @@ -1256,8 +1256,12 @@ export function GroupChatWorkspace({ group, members, onBack, visible = true }: G
{roomClarifies.length ? b.group.waitingForAnswer - : room.turn - ? b.group.memberThinking(groupSpeakerLabel(room.turn)) + : Object.keys(room.turns || {}).length + ? b.group.memberThinking( + Object.values(room.turns || {}) + .map(name => groupSpeakerLabel(name)) + .join(', ') + ) : b.group.roomWorking}
) : null} diff --git a/apps/desktop/src/plugins/hermes-bots/group-chat.ts b/apps/desktop/src/plugins/hermes-bots/group-chat.ts index 9949c57824..84015a5c50 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-chat.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-chat.ts @@ -1375,12 +1375,13 @@ export interface GroupHoldStamp extends GroupHold { } /** The room record as the coordination engine handles it: `GroupChat` plus - * `turn`, the runtime-only name of the member currently mid-turn. Like + * `turns`, the runtime-only memberKey → name map of members currently + * mid-turn (several at once — a round's turns run concurrently). Like * `running`/`epoch` it never persists, so it has no place in the durable * shape. Holds carry the fuller live stamp. */ export interface GroupChatRoom extends GroupChat { holds?: Record - turn?: null | string + turns?: Record } /** Set or clear a group chat's room picture (small data URL, normalized by 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 fd9ec778bf..fbdc740b2e 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-rounds.test.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-rounds.test.ts @@ -236,6 +236,49 @@ describe('round lifecycle', () => { }) }) +describe('concurrent rounds', () => { + it('runs every responder of a round at once, so a round takes as long as its slowest member', async () => { + // Each turn holds until ALL three members have submitted: under the old + // serial loop the first member's turn could never finish (nobody else + // submits until it does) and this would deadlock at the drain bound. + let submitted = 0 + const release: Array<() => void> = [] + + const room = await loadRoom({ + turn: ({ profile, prompt }) => + new Promise(resolve => { + submitted += 1 + // Round 1 (the user's question is in the delta): speak. Later + // rounds (only sibling replies in the delta): pass. + release.push(() => resolve(prompt.includes('status?') ? `${profile} here` : '(pass)')) + + if (submitted % 3 === 0) { + for (const fn of release.splice(0)) { + fn() + } + } + }) + }) + + room.rounds.sendToGroupChat('Fast', MEMBERS, 'everyone, status?') + await settle(room, 'Fast') + + const replies = log(room, 'Fast').filter(entry => entry.from.kind === 'member') + + expect(replies.map(entry => entry.from.name).sort()).toEqual(['builder', 'ops', 'research']) + // Round 2 delivers every sibling's round-1 reply to each member exactly + // once — concurrent commits never eat or duplicate each other's deltas — + // and never echoes a member's own reply back to it. + expect(room.gateway.calls).toHaveLength(6) + + for (const call of room.gateway.calls.slice(3)) { + for (const name of MEMBERS.map(member => member.name)) { + expect(call.prompt.split(`${name} here`)).toHaveLength(name === call.profile ? 1 : 2) + } + } + }) +}) + describe('per-member delta', () => { it('feeds a second send only the NEW messages', async () => { const room = await loadRoom() @@ -726,7 +769,7 @@ describe('stopGroupThread (#91868/#94569)', () => { members: STOP_MEMBERS, running: true, sessions: { alpha: 'live-alpha-sid' }, - turn, + turns: turn ? { [turn]: turn } : {}, watermarks: {} } } as unknown as Record) @@ -742,7 +785,7 @@ describe('stopGroupThread (#91868/#94569)', () => { expect(state.epoch).toBe(4) expect(state.running).toBe(false) - expect(state.turn).toBeNull() + expect(state.turns).toEqual({}) for (const member of STOP_MEMBERS) { expect(state.holds?.[member.name]).toBeTruthy() @@ -758,7 +801,7 @@ describe('stopGroupThread (#91868/#94569)', () => { const interrupts = room.gateway.rpcFor('session.interrupt') - // Exactly one — the serial loop has one member in flight. + // Exactly one — only alpha is mid-turn in the seeded room. expect(interrupts).toHaveLength(1) expect(interrupts[0].params.session_id).toBe('live-alpha-sid') }) diff --git a/apps/desktop/src/plugins/hermes-bots/group-rounds.ts b/apps/desktop/src/plugins/hermes-bots/group-rounds.ts index ff65ff2745..3758c1625a 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-rounds.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-rounds.ts @@ -415,18 +415,19 @@ export function unaddressedGroupMentions(group: string, members: GroupMember[], * 2. Sets a #93129 hold for EVERY member — future turns stay skipped until * the user explicitly releases (resume / @all resume / direct mention), * the exact contract user-typed "@all stop" already has. - * 3. Sends session.interrupt to the member currently ON TURN (room.turn, - * runtime-only) via its own route, so the in-flight model call actually - * dies instead of grinding to completion in the background. Best-effort: - * an unreachable member still leaves the room stopped — the poll loop's - * staleness check (epoch moved AND member held) abandons the turn. + * 3. Sends session.interrupt to every member currently ON TURN + * (room.turns, runtime-only — a round's turns run concurrently) via its + * own route, so the in-flight model calls actually die instead of + * grinding to completion in the background. Best-effort: an unreachable + * member still leaves the room stopped — the poll loop's staleness check + * (epoch moved AND member held) abandons the turn. * * `members` is the live roster when the caller has one (the workspace); * falls back to the room's durable roster so a two-arg call still works. */ export async function stopGroupThread(group: string, thread: null | string, members: GroupMember[] | null = null) { const room = $groupChats.get()[group] || {} const roster = Array.isArray(members) && members.length ? members : room.members || [] - const turnName = room.turn || null + const onTurnKeys = Object.keys(room.turns || {}) const stamp: GroupHoldStamp = { at: Date.now(), @@ -437,7 +438,7 @@ export async function stopGroupThread(group: string, thread: null | string, memb updateGroupChat(group, (r: GroupChatRoom) => { r.epoch = (r.epoch || 0) + 1 r.running = false - r.turn = null + r.turns = {} // Same hold shape applyGroupHoldDirective mints for "@all stop" — the // held-skip path (watermark consume + 'held' activity note) and every @@ -470,28 +471,262 @@ export async function stopGroupThread(group: string, thread: null | string, memb thread: thread || null }) - // Interrupt the member actually mid-turn. room.turn is runtime-only and - // names exactly one member (the loop is serial); a settled room has none. - const onTurn = turnName ? roster.find((member: GroupMember) => member?.name === turnName) : null - const sessionId = onTurn ? (room.sessions || {})[groupMemberKey(onTurn)] : null + // Interrupt every member actually mid-turn. room.turns is runtime-only; + // a settled room has none. + await Promise.all( + roster + .filter((member: GroupMember) => onTurnKeys.includes(groupMemberKey(member))) + .map(async (onTurn: GroupMember) => { + const sessionId = (room.sessions || {})[groupMemberKey(onTurn)] - if (onTurn && sessionId) { - try { - await requestForBot(onTurn, 'session.interrupt', { - session_id: sessionId + if (!sessionId) { + return + } + + try { + await requestForBot(onTurn, 'session.interrupt', { + session_id: sessionId + }) + } catch { + /* best-effort — the epoch/hold legs above already stopped the room; + the abandoned poll loop exits on its staleness check */ + } }) - } catch { - /* best-effort — the epoch/hold legs above already stopped the room; - the abandoned poll loop exits on its staleness check */ - } - } + ) } -/** 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. - * Watermarks are per thread+member (`${thread}::${memberKey}`), so parallel - * topics never eat each other's deltas. */ +/** Why one member's turn ended without a committed reply. */ +type GroupTurnOutcome = 'cancelled' | 'passed' | 'skipped' | 'spoke' + +/** One member's full turn against the room: compute its unseen delta, skip + * (held / nothing new), run the turn, then commit the result under the + * #93127 staleness check. Pure with respect to its siblings — several + * members' turns run CONCURRENTLY within a round (they share nothing but + * the room log, which appends are serialized through the atom), and each + * member still sees only what was in the room when its turn started. */ +async function takeMemberTurn( + group: string, + members: GroupMember[], + member: GroupMember, + thread: string, + startEpoch: number, + attachImages: boolean +): Promise { + const room = $groupChats.get()[group] || { + log: [], + watermarks: {} + } + + const memberKey = groupMemberKey(member) + const markKey = `${thread}::${memberKey}` + const seen = room.watermarks[markKey] || 0 + + // Delta: NEW room entries, narrowed to this thread — the member's turn sees + // only the conversation it's part of — minus its OWN replies. Those already + // live in its session as assistant messages; echoing them back costs a + // turn that can only pass. (Concurrent rounds make this matter: a member's + // reply lands beside its siblings', so an index watermark can't cleanly + // step over "just mine".) + const delta = room.log + .slice(seen) + .filter((e: GroupMessage) => groupThreadOf(e) === thread && !isOwnGroupEntry(e, member)) + + if (!delta.length) { + return 'skipped' + } + + // #93129: a member the user told to stop is HELD — no turn until an + // explicit release (resume / @all resume / a direct non-stop mention). + // Consume the delta exactly once (watermark past the current log) so the + // same entries never re-trigger this skip, and surface WHY the bot is + // silent in the activity feed the first time. + const heldEntry = (room.holds || {})[memberKey] + + if (heldEntry) { + const advance = heldMemberWatermarkAdvance(seen, room.log.length) + updateGroupChat(group, (r: GroupChatRoom) => { + if (advance !== null) { + r.watermarks[markKey] = advance + } + + if (r.holds?.[memberKey] && !r.holds[memberKey].noted) { + r.holds = { + ...r.holds, + [memberKey]: { + ...r.holds[memberKey], + noted: true + } + } + } + + return r + }) + + if (!heldEntry.noted) { + recordGroupActivity(group, { + kind: 'held', + member: member.name, + thread + }) + } + + return 'skipped' + } + + const prompt = buildGroupChatTurnPrompt({ + groupName: group, + members, + viewer: member, + deltaLines: delta.slice(-GROUP_CHAT_HISTORY_LIMIT).map((e: GroupMessage) => formatGroupChatLine(e, member.name)) + }) + + // Images riding this delta (user attachments — member entries don't carry + // images today, but flatMap keeps this future-proof) get staged into the + // member's session so the model sees the pixels, not just the transcript's + // [attached image: …] marker. Continuation turns are text-only. + const deltaImages = attachImages + ? delta.flatMap((e: GroupMessage) => (Array.isArray(e.images) ? e.images : [])) + : undefined + + // Surface WHO is on turn (runtime-only, like running/epoch) so the room + // shows "Radar is thinking…" — several members can be mid-turn at once. + updateGroupChat(group, (r: GroupChatRoom) => { + r.turns = { + ...(r.turns || {}), + [memberKey]: member.name + } + + return r + }) + let reply: null | string = null + + try { + reply = await runGroupChatMemberTurn(group, member, prompt, thread, deltaImages) + + // Needs-attention hook (#93091 item 3): a turn that produced a real + // reply (or an explicit pass) is a good turn — clear the badge. A + // timed-out turn also returns null but never threw; leaving any prior + // badge in place there is the conservative choice. + if (reply !== null) { + clearBotAttention(memberKey) + } + } catch (error: any) { + const reason = String(error?.data?.reason || '').trim() + recordGroupActivity(group, { + kind: 'failed', + member: member.name, + thread, + ...(reason + ? { + reason + } + : {}) + }) + noteBotAttention(memberKey, reason || error?.message || error) + reply = null // a failed turn is a pass, never a room error + } finally { + updateGroupChat(group, (r: GroupChatRoom) => { + const next = { + ...(r.turns || {}) + } + + delete next[memberKey] + r.turns = next + + return r + }) + } + + // #93127: the turn may have finished AFTER a newer user send bumped the + // room epoch. That newer send's loop re-drives this member with the full + // delta, so committing this stale result (watermark advance + append) + // would double-deliver the same reply. Drop it here — BEFORE the watermark + // advance and BEFORE the append. Only a newer USER entry in THIS thread + // makes the re-drive premise true: a cross-thread send bumps the epoch + // too, but its loop filters this thread out and would never regenerate + // the finished reply. The during-turn tail is anchored by entry id, not + // index — the history trim drops entries from the FRONT, so an index + // slice could overshoot after a mid-turn trim and silently commit a stale + // turn. + const roomNow = $groupChats.get()[group] || { + log: [] + } + + 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. + const turnTail = anchorIdx >= 0 ? roomNow.log.slice(anchorIdx + 1) : roomNow.log + + const newerUserEntryInThread = turnTail.some( + (e: GroupMessage) => e.from?.kind === 'user' && groupThreadOf(e) === thread + ) + + if (!shouldCommitMemberTurn(startEpoch, epochNow, newerUserEntryInThread)) { + recordGroupActivity(group, { + kind: 'cancelled', + member: member.name, + thread + }) + + return 'cancelled' + } + + // The member has now seen everything up to the PRE-TURN log length — not + // the current one: a sibling's concurrent reply that landed while this + // member was thinking is genuinely unseen and must reach it next round. + // (Its own reply, appended below, is excluded from deltas by author.) + const seenThrough = room.log.length + updateGroupChat(group, (r: GroupChatRoom) => { + r.watermarks[markKey] = Math.min(seenThrough, r.log.length) + + return r + }) + + if (reply === null || isGroupPassText(reply)) { + return 'passed' + } + + appendGroupChatEntry( + group, + { + kind: 'member', + name: member.name, + ...(member.remoteSource + ? { + source: member.connectionLabel || member.connectionId + } + : {}) + }, + reply, + thread + ) + + return 'spoke' +} + +/** A room entry this member authored (same kind, name and — for + * cross-connection members — same source device). */ +function isOwnGroupEntry(entry: GroupMessage, member: GroupMember): boolean { + if (entry.from?.kind !== 'member' || entry.from.name !== member.name) { + return false + } + + const source = member.remoteSource ? member.connectionLabel || member.connectionId : undefined + + return (entry.from.source || undefined) === (source || undefined) +} + +/** Drive one bounded set of rounds for ONE THREAD. Within a round, every + * responder takes its turn CONCURRENTLY — a room of N bots answers in the + * time of the slowest one, not the sum of all N. Rounds stay serial: each + * round's prompts include the previous round's replies, so bots build on + * each other. A newer user send bumps the room epoch; this loop notices at + * the next round boundary (and every turn's commit check), bails, and the + * newest send's own loop takes over. 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) { const startEpoch = ($groupChats.get()[group] || {}).epoch || 0 const isCurrent = () => (($groupChats.get()[group] || {}).epoch || 0) === startEpoch @@ -502,23 +737,53 @@ export async function runGroupChatRounds(group: string, members: GroupMember[], // cap forced the exit — the activity feed must tell those apart. let exitKind: 'capped' | 'settled' = 'settled' + /** Run one set of members concurrently. Returns how many spoke, or null + * when a newer send superseded this drive mid-round. */ + const runConcurrentTurns = async (responders: GroupMember[], attachImages: boolean): Promise => { + // The message cap is enforced per ROUND: a round admits at most the + // remaining budget worth of speakers, and its concurrent turns can't + // overshoot it by more than the round size. + const budget = GROUP_CHAT_MAX_MESSAGES - posted + + if (budget <= 0) { + return 0 + } + + const outcomes = await Promise.all( + responders.slice(0, budget).map(member => takeMemberTurn(group, members, member, thread, startEpoch, attachImages)) + ) + + if (!isCurrent() || outcomes.includes('cancelled')) { + return null + } + + const spoke = outcomes.filter(outcome => outcome === 'spoke').length + posted += spoke + + return spoke + } + try { for (let round = 0; round < GROUP_CHAT_MAX_ROUNDS; round++) { // Deliver any replies that finished after their turn timed out — // every member, not just this round's responders, so long work is // late, never lost. - for (const member of members) { - if (!isCurrent()) { - recordGroupActivity(group, { - kind: 'cancelled', - member: null, - thread - }) + if (!isCurrent()) { + recordGroupActivity(group, { + kind: 'cancelled', + member: null, + thread + }) - return - } + return + } - await harvestStrandedGroupReply(group, member) + await Promise.all(members.map(member => harvestStrandedGroupReply(group, member))) + + if (posted >= GROUP_CHAT_MAX_MESSAGES) { + exitKind = 'capped' // message cap, not consensus (#94478) + + return } const roomLog = (($groupChats.get()[group] || {}).log || []).filter( @@ -543,195 +808,16 @@ export async function runGroupChatRounds(group: string, members: GroupMember[], (member: GroupMember) => !Object.prototype.hasOwnProperty.call(strandedNow, groupMemberKey(member)) ) - let spokeThisRound = 0 + let spokeThisRound = await runConcurrentTurns(responders, true) - for (const member of responders) { - if (!isCurrent() || posted >= GROUP_CHAT_MAX_MESSAGES) { - if (!isCurrent()) { - recordGroupActivity(group, { - kind: 'cancelled', - member: null, - thread - }) - } else { - exitKind = 'capped' // message cap, not consensus (#94478) - } - - return - } - - const room = $groupChats.get()[group] || { - log: [], - watermarks: {} - } - - const memberKey = groupMemberKey(member) - const markKey = `${thread}::${memberKey}` - const seen = room.watermarks[markKey] || 0 - // Delta: NEW room entries, narrowed to this thread — the member's - // turn sees only the conversation it's part of. - const delta = room.log.slice(seen).filter((e: GroupMessage) => groupThreadOf(e) === thread) - - if (!delta.length) { - continue - } - - // #93129: a member the user told to stop is HELD — no turn until an - // explicit release (resume / @all resume / a direct non-stop - // mention). Consume the delta exactly once (watermark past the - // current log) so the same entries never re-trigger this skip, and - // surface WHY the bot is silent in the activity feed the first time. - const heldEntry = (room.holds || {})[memberKey] - - if (heldEntry) { - const advance = heldMemberWatermarkAdvance(seen, room.log.length) - updateGroupChat(group, (r: GroupChatRoom) => { - if (advance !== null) { - r.watermarks[markKey] = advance - } - - if (r.holds?.[memberKey] && !r.holds[memberKey].noted) { - r.holds = { - ...r.holds, - [memberKey]: { - ...r.holds[memberKey], - noted: true - } - } - } - - return r - }) - - if (!heldEntry.noted) { - recordGroupActivity(group, { - kind: 'held', - member: member.name, - thread - }) - } - - continue - } - - const prompt = buildGroupChatTurnPrompt({ - groupName: group, - members, - viewer: member, - deltaLines: delta - .slice(-GROUP_CHAT_HISTORY_LIMIT) - .map((e: GroupMessage) => formatGroupChatLine(e, member.name)) + if (spokeThisRound === null) { + recordGroupActivity(group, { + kind: 'cancelled', + member: null, + thread }) - // Images riding this delta (user attachments — member entries don't - // carry images today, but flatMap keeps this future-proof) get staged - // into the member's session so the model sees the pixels, not just - // the transcript's [attached image: …] marker. - const deltaImages = delta.flatMap((e: GroupMessage) => (Array.isArray(e.images) ? e.images : [])) - - // Surface WHO is on turn (runtime-only, like running/epoch) so the - // room shows "Radar is thinking…" instead of a generic working line — - // long model turns otherwise read as the room being stuck. - updateGroupChat(group, (r: GroupChatRoom) => { - r.turn = member.name - - return r - }) - let reply: null | string = null - - try { - reply = await runGroupChatMemberTurn(group, member, prompt, thread, deltaImages) - - // Needs-attention hook (#93091 item 3): a turn that produced a real - // reply (or an explicit pass) is a good turn — clear the badge. - // A timed-out turn also returns null but never threw; leaving any - // prior badge in place there is the conservative choice. - if (reply !== null) { - clearBotAttention(groupMemberKey(member)) - } - } catch (error: any) { - const reason = String(error?.data?.reason || '').trim() - recordGroupActivity(group, { - kind: 'failed', - member: member.name, - thread, - ...(reason - ? { - reason - } - : {}) - }) - noteBotAttention(groupMemberKey(member), reason || error?.message || error) - reply = null // a failed turn is a pass, never a room error - } - - // #93127: the turn may have finished AFTER a newer user send bumped - // the room epoch. That newer send's loop re-drives this member with - // the full delta, so committing this stale result (watermark advance - // + append) would double-deliver the same reply. Drop it here — - // BEFORE the watermark advance and BEFORE the append. Only a newer - // USER entry in THIS thread makes the re-drive premise true: a - // cross-thread send bumps the epoch too, but its loop filters this - // thread out and would never regenerate the finished reply. The - // during-turn tail is anchored by entry id, not index — the history - // trim drops entries from the FRONT, so an index slice could - // overshoot after a mid-turn trim and silently commit a stale turn. - const roomNow = $groupChats.get()[group] || { - log: [] - } - - 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. - const turnTail = anchorIdx >= 0 ? roomNow.log.slice(anchorIdx + 1) : roomNow.log - - const newerUserEntryInThread = turnTail.some( - (e: GroupMessage) => e.from?.kind === 'user' && groupThreadOf(e) === thread - ) - - if (!shouldCommitMemberTurn(startEpoch, epochNow, newerUserEntryInThread)) { - recordGroupActivity(group, { - kind: 'cancelled', - member: member.name, - thread - }) - - return - } - - // The member has now seen everything up to the pre-reply log length. - updateGroupChat(group, (r: GroupChatRoom) => { - r.watermarks[markKey] = r.log.length - - return r - }) - - if (reply !== null && !isGroupPassText(reply)) { - appendGroupChatEntry( - group, - { - kind: 'member', - name: member.name, - ...(member.remoteSource - ? { - source: member.connectionLabel || member.connectionId - } - : {}) - }, - reply, - thread - ) - // Its own message counts as seen too. - updateGroupChat(group, (r: GroupChatRoom) => { - r.watermarks[markKey] = r.log.length - - return r - }) - posted += 1 - spokeThisRound += 1 - } + return } if (spokeThisRound === 0) { @@ -750,119 +836,27 @@ export async function runGroupChatRounds(group: string, members: GroupMember[], continuations += 1 if (pendingKeys.length && continuations <= GROUP_CHAT_MAX_CONTINUATIONS) { - const citedMembers = members.filter((member: GroupMember) => pendingKeys.includes(groupMemberKey(member))) + const strandedNow = ($groupChats.get()[group] || {}).stranded || {} - if (citedMembers.length && posted < GROUP_CHAT_MAX_MESSAGES) { - const strandedNow = ($groupChats.get()[group] || {}).stranded || {} + const citedMembers = members.filter( + (member: GroupMember) => + pendingKeys.includes(groupMemberKey(member)) && + !Object.prototype.hasOwnProperty.call(strandedNow, groupMemberKey(member)) + ) - const continuationResponders = citedMembers.filter( - (member: GroupMember) => !Object.prototype.hasOwnProperty.call(strandedNow, groupMemberKey(member)) - ) + // The continuation prompt centers on what each cited member + // missed: everything since its watermark, which includes the + // reply that cites it. Holds still apply (#93129). + const continued = citedMembers.length ? await runConcurrentTurns(citedMembers, false) : 0 - for (const member of continuationResponders) { - if (!isCurrent() || posted >= GROUP_CHAT_MAX_MESSAGES || continuations > GROUP_CHAT_MAX_CONTINUATIONS) { - break - } - - const room = $groupChats.get()[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) { - continue - } - - const heldEntry = (room.holds || {})[memberKey] - - if (heldEntry) { - continue // holds still apply to continuation turns (#93129) - } - - const prompt = buildGroupChatTurnPrompt({ - groupName: 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.name)) - }) - - updateGroupChat(group, (r: GroupChatRoom) => { - r.turn = member.name - - return r - }) - let continuationReply: null | string = null - - try { - continuationReply = await runGroupChatMemberTurn(group, member, prompt, thread) - - if (continuationReply !== null) { - clearBotAttention(memberKey) - } - } catch (error: any) { - recordGroupActivity(group, { - kind: 'failed', - member: member.name, - thread - }) - noteBotAttention(memberKey, error?.message || error) - continuationReply = null - } - - if (!isCurrent()) { - return - } - - updateGroupChat(group, (r: GroupChatRoom) => { - r.watermarks[markKey] = r.log.length - - return r - }) - - if (continuationReply !== null && !isGroupPassText(continuationReply)) { - appendGroupChatEntry( - group, - { - kind: 'member', - name: member.name, - ...(member.remoteSource - ? { - source: member.connectionLabel || member.connectionId - } - : {}) - }, - continuationReply, - thread - ) - updateGroupChat(group, (r: GroupChatRoom) => { - r.watermarks[markKey] = r.log.length - - return r - }) - posted += 1 - - // The continuation's own reply may cite someone else — fall - // through to the normal loop so the next round handles it via - // the same responder machinery. Reaching here means the loop - // continues rather than settling; the outer for-loop's next - // iteration re-evaluates everything. - spokeThisRound += 1 - } - } + if (continued === null) { + return } + + // The continuation's own replies may cite someone else — fall + // through to the normal loop so the next round handles it via the + // same responder machinery. + spokeThisRound = continued } if (spokeThisRound === 0) { @@ -895,7 +889,7 @@ export async function runGroupChatRounds(group: string, members: GroupMember[], }) updateGroupChat(group, (r: GroupChatRoom) => { r.running = false - r.turn = null + r.turns = {} return r }) diff --git a/apps/desktop/src/plugins/hermes-bots/group-turns.test.ts b/apps/desktop/src/plugins/hermes-bots/group-turns.test.ts index 90b7deaef4..b956c4883c 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-turns.test.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-turns.test.ts @@ -271,6 +271,45 @@ describe('per-turn socket lease', () => { }) }) +describe('push-woken poll', () => { + it("wakes on the member session's message.complete instead of sleeping out the backstop", async () => { + // Real timers here: the contract is about WHEN the poll re-reads. + vi.unstubAllGlobals() + const listeners = new Map void>>() + const room = await loadRoom({ pollsBusy: 1, turn: () => 'woken reply' }) + + host.onEvent = (type: string, listener: (event: unknown) => void) => { + const set = listeners.get(type) ?? new Set() + set.add(listener) + listeners.set(type, set) + + return () => set.delete(listener) + } + + const started = Date.now() + const turn = room.turns.runGroupChatMemberTurn('Room', LOCAL_MEMBER, 'hi', 't1', []) + + // Let the submit land and the first poll wait attach its listeners, then + // fire the terminal frame for the runtime id the submit used. + await new Promise(resolve => setTimeout(resolve, 50)) + // The harness mints a fresh runtime id on every resume; the frame carries + // whichever one the session currently answers to. + const runtime = room.gateway.sessions.get(room.gateway.calls[0]?.stored)?.runtime + expect(listeners.get('message.complete')?.size).toBe(1) + + for (const listener of listeners.get('message.complete') ?? []) { + listener({ type: 'message.complete', session_id: runtime }) + } + + expect(await turn).toBe('woken reply') + // Two quick re-reads (busy once, then done) — well under one 5s backstop tick. + expect(Date.now() - started).toBeLessThan(2000) + // Every listener was disposed once the turn finished. + expect(listeners.get('message.complete')?.size ?? 0).toBe(0) + expect(listeners.get('error')?.size ?? 0).toBe(0) + }) +}) + // #94376: a Codex intent-ack continuation nudge can land a substantive // answer, then get a synthetic "(pass)" reply to the nudge itself. describe('reply selection (#94376)', () => { diff --git a/apps/desktop/src/plugins/hermes-bots/group-turns.ts b/apps/desktop/src/plugins/hermes-bots/group-turns.ts index b9bd3e172c..7a79a37c14 100644 --- a/apps/desktop/src/plugins/hermes-bots/group-turns.ts +++ b/apps/desktop/src/plugins/hermes-bots/group-turns.ts @@ -231,7 +231,71 @@ export async function ensureGroupChatSession(group: string, member: GroupMember) } const GROUP_TURN_TIMEOUT_MS = 180000 -const GROUP_TURN_POLL_MS = 2000 +// Backstop cadence only. The turn normally wakes the instant the member's +// session emits its terminal frame (message.complete / error) via host.onEvent; +// this poll exists for hosts without the event tap, sessions whose events +// ride a socket this window doesn't hold, and frames lost to a reconnect. +const GROUP_TURN_POLL_MS = 5000 +const GROUP_TURN_SETTLE_RECHECK_MS = 250 +const GROUP_TURN_SETTLE_RECHECKS = 8 + +/** Resolve as soon as the member's session reports a terminal frame for the + * turn — or after `ms` as the backstop. Feature-detected: without + * `host.onEvent` (older shells, the node test harness) this is a plain sleep. + * `error` is terminal too (agent init failed → no message.complete follows). */ +function waitForTurnSignal(runtimeIds: string[], ms: number): Promise { + const ids = new Set(runtimeIds.filter(id => id.length > 0)) + + return new Promise(resolve => { + const unsubs: Array<() => void> = [] + let timer: null | ReturnType = null + let settled = false + + const done = (signalled: boolean) => { + if (settled) { + return + } + + settled = true + + if (timer !== null) { + clearTimeout(timer) + } + + for (const unsub of unsubs) { + try { + unsub() + } catch { + /* disposer already ran */ + } + } + + resolve(signalled) + } + + // (The test harness runs timers inline — `done` may already have fired + // by the time this assignment lands, hence the `settled` guard below.) + timer = setTimeout(() => done(false), ms) + + if (settled || typeof host.onEvent !== 'function' || !ids.size) { + return + } + + for (const type of ['message.complete', 'error']) { + try { + unsubs.push( + host.onEvent(type, (event: { session_id?: string }) => { + if (ids.has(String(event?.session_id || ''))) { + done(true) + } + }) + ) + } catch { + /* event tap unavailable — the timer still resolves */ + } + } + }) +} // --- group-turn session-lease helpers (#93602) ------------------------------ // A member turn is a session-scoped RPC SEQUENCE (resume → attach → submit → @@ -560,6 +624,9 @@ async function runGroupChatMemberTurnLeased( // Baseline: how many messages exist before our submit. let before = 0 + // Every runtime id this turn has seen for the member's session. Terminal + // frames are keyed by runtime id, and a resume can hand back a fresh one. + const runtimeIds = new Set([runtime]) try { const pre = (await requestForBot(member, 'session.resume', { @@ -568,6 +635,10 @@ async function runGroupChatMemberTurnLeased( })) as GroupSessionSnapshot before = Array.isArray(pre?.messages) ? pre.messages.length : pre?.message_count || 0 + + if (pre?.session_id) { + runtimeIds.add(pre.session_id) + } } catch { /* lazy session — zero messages */ } @@ -625,11 +696,20 @@ async function runGroupChatMemberTurnLeased( // minting and submitting. Tracks the runtime id the submit landed on so // the poll fallback below targets a live session. const liveRuntime = await submitGroupTurnPrompt(member, runtime, stored, turnText) + runtimeIds.add(liveRuntime) const started = Date.now() let deadline = started + GROUP_TURN_TIMEOUT_MS + // After the terminal frame fires, the gateway still has to flip + // session.running off in its turn `finally` — re-check quickly for a few + // beats instead of falling back to the slow backstop cadence. + let quickRechecks = 0 while (Date.now() < deadline) { - await new Promise(resolve => setTimeout(resolve, GROUP_TURN_POLL_MS)) + const signalled = quickRechecks + ? await waitForTurnSignal([], GROUP_TURN_SETTLE_RECHECK_MS) + : await waitForTurnSignal([...runtimeIds], GROUP_TURN_POLL_MS) + + quickRechecks = signalled ? GROUP_TURN_SETTLE_RECHECKS : Math.max(0, quickRechecks - 1) // #91868/#94569: an explicit stop (stopGroupThread) bumped the epoch AND // held this member — the member's session was interrupted, so nothing is @@ -654,6 +734,10 @@ async function runGroupChatMemberTurnLeased( continue } + if (state?.session_id) { + runtimeIds.add(state.session_id) + } + const messages = Array.isArray(state?.messages) ? state.messages : [] const busy = Boolean(state?.inflight || state?.running) // A clarify blocking inside the member's session is a question for the diff --git a/website/docs/user-guide/bot-mode.md b/website/docs/user-guide/bot-mode.md index 0e50502a3f..42ff758372 100644 --- a/website/docs/user-guide/bot-mode.md +++ b/website/docs/user-guide/bot-mode.md @@ -83,7 +83,8 @@ Groups are standalone rows in the same activity-ordered roster as Bot DMs. A Bot **Open chat** on any group row (2–6 Bots) opens a shared room where the whole group coordinates: -- Your message triggers up to **three serial rounds** of member turns. @-mentioned Bots respond (everyone responds when nobody is mentioned); each Bot replies briefly or passes, and the room settles when a full round stays silent. +- Your message triggers up to **three rounds** of member turns. Within a round every responding Bot thinks **at the same time**, so a room of five answers about as fast as one; rounds run in sequence so each Bot sees what the others just said before it speaks again. @-mentioned Bots respond (everyone responds when nobody is mentioned); each Bot replies briefly or passes, and the room settles when a full round stays silent. +- Replies land the moment a Bot finishes — the room listens for each member session's completion event rather than polling on a timer (a slow 5s poll remains as a backstop for older gateways). - Bots pull each other in with `@name`, and escalate real judgment calls to you with `@user` — the group row shows a **needs you** badge when that happens. - Hard caps (10 messages per send, 3 rounds) keep rooms from spinning. - Each member keeps its own persistent `Group: ` session, so room context survives like any other conversation.