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.