From 24eeafa78f584d8a24f40b4e1681f617d17003ef Mon Sep 17 00:00:00 2001 From: funky-xamarin <30426178+Wenfengcheng@users.noreply.github.com> Date: Fri, 21 Aug 2026 16:17:17 +0800 Subject: [PATCH] fix(bot-mode): isolate group thread sessions --- .../desktop/src/plugins/hermes-bots/plugin.js | 378 +++++++++++++----- .../hermes-bots/tests/group-chat.test.mjs | 259 +++++++++++- .../tests/group-turn-lease.test.mjs | 42 +- 3 files changed, 572 insertions(+), 107 deletions(-) diff --git a/apps/desktop/src/plugins/hermes-bots/plugin.js b/apps/desktop/src/plugins/hermes-bots/plugin.js index 0470defffd7a0..ea3c73963e0cf 100644 --- a/apps/desktop/src/plugins/hermes-bots/plugin.js +++ b/apps/desktop/src/plugins/hermes-bots/plugin.js @@ -441,6 +441,7 @@ function focusedRosterOwner(owner) { /** Group-chat rooms: { [group]: { log: [{from:{kind,name},text,at}], watermarks:{[member]:idx}, epoch, running } }. * Log + watermarks persist via plugin storage; epoch/running are runtime-only. */ const $groupChats = atom({}) +const groupSessionFlights = new Map() /** Group whose room view is open in the Bots pane (secondary navigation * inside the pane; a normal row click returns to the roster). */ const $groupChatWorkspace = atom(null) @@ -1047,6 +1048,7 @@ function durableGroupChatRooms(all = $groupChats.get()) { log: room.log, watermarks: room.watermarks || {}, sessions: room.sessions || {}, + sessionOwners: room.sessionOwners || {}, stranded: room.stranded || {}, members: Array.isArray(room.members) ? room.members : [], // Immutable room identity: without this, a room merged in via the @@ -1055,6 +1057,7 @@ function durableGroupChatRooms(all = $groupChats.get()) { // name-keyed identity — same field updateGroupChat's inline map // already carries. roomId: typeof room.roomId === 'string' && room.roomId ? room.roomId : null, + sessionScopeId: typeof room.sessionScopeId === 'string' && room.sessionScopeId ? room.sessionScopeId : null, image: room.image || null, syncRevision: Math.max(0, Number(room.syncRevision || 0)) } @@ -2306,21 +2309,26 @@ function hideOwnedBotSessions() { } const persisted = room?.sessionOwners?.[key] - const derived = (room?.members || []).find(member => groupMemberKey(member) === key) + const persistedKey = persisted ? groupMemberKey(persisted) : null + const persistedOwnsKey = persistedKey && ( + key === persistedKey || (persisted?.threadScoped === true && key.endsWith(`::${persistedKey}`)) + ) + const derived = persistedOwnsKey ? null : (room?.members || []).find(member => groupMemberKey(member) === key) // Bare keys are legacy local rooms. A source-qualified key without its // immutable owner is unsafe: never let it fall through ambient routing. - const owner = persisted || derived || (!key.includes('::') ? { name: key } : null) + const owner = persistedOwnsKey ? persisted : derived || (!key.includes('::') ? { name: key } : null) - if (key.includes('::')) { + if (owner && (owner.sourceScoped || owner.remoteSource)) { const route = owner?.route - const sourceMarked = owner?.sourceScoped || owner?.remoteSource const routeKey = route?.connectionId && route?.profile ? `${route.connectionId}::${route.profile}` : '' - if (!sourceMarked || !route?.targetProfile || routeKey !== key) { + if (!route?.targetProfile || routeKey !== groupMemberKey(owner)) { return null } + } else if (!persistedOwnsKey && key.includes('::')) { + return null } return owner ? { owner, id, dedupe: `${key}\u0000${id}` } : null @@ -5476,6 +5484,10 @@ function groupMemberKey(member) { return member?.sourceScoped || member?.remoteSource ? botRosterKey(member) : member?.name } +function groupThreadMemberKey(thread, member) { + return `${thread}::${groupMemberKey(member)}` +} + /** Serializable immutable owner captured beside every group plumbing session. */ function groupSessionOwner(member) { const route = botConnectionRoute(member) @@ -5494,6 +5506,56 @@ function groupSessionOwner(member) { } } +function groupSessionScopeForRoom(room, group) { + const roomId = typeof room?.roomId === 'string' ? room.roomId : '' + + return roomId && roomId.length <= 40 + ? roomId + : (typeof room?.sessionScopeId === 'string' && room.sessionScopeId) || group +} + +function ensureGroupSessionScope(group) { + const room = $groupChats.get()[group] || {} + const roomId = typeof room.roomId === 'string' ? room.roomId : '' + + if (roomId && roomId.length <= 40) { + return roomId + } + + if (typeof room.sessionScopeId === 'string' && room.sessionScopeId) { + return room.sessionScopeId + } + + const sessionScopeId = mintGroupRoomId() + updateGroupChat(group, current => { + current.sessionScopeId ||= sessionScopeId + return current + }) + + return $groupChats.get()[group]?.sessionScopeId || sessionScopeId +} + +function updateGroupSessionPointer(group, scope, key, stored, member) { + if (!stored) { + return + } + + const rooms = $groupChats.get() + const currentGroup = rooms[group] && groupSessionScopeForRoom(rooms[group], group) === scope + ? group + : Object.keys(rooms).find(name => groupSessionScopeForRoom(rooms[name], name) === scope) + + if (!currentGroup) { + return + } + + updateGroupChat(currentGroup, room => { + room.sessions = { ...(room.sessions || {}), [key]: stored } + room.sessionOwners = { ...(room.sessionOwners || {}), [key]: { ...groupSessionOwner(member), threadScoped: true } } + return room + }) +} + // ── alias identity for connection rows (#89131) ───────────────────────────── // A Desktop per-profile alias (profile `moxie` with a Cloud/URL/SSH override) // routes to a remote backend's root profile: its route reads @@ -6973,6 +7035,7 @@ function updateGroupChat(group, mutate, { sync = true } = {}) { members: Array.isArray(room.members) ? room.members : [], // Immutable room identity: the member-session title for new rooms. roomId: typeof room.roomId === 'string' && room.roomId ? room.roomId : null, + sessionScopeId: typeof room.sessionScopeId === 'string' && room.sessionScopeId ? room.sessionScopeId : null, // Room picture (small data URL, same normalization as bot avatars). image: room.image || null, syncRevision: Math.max(0, Number(room.syncRevision || 0)) @@ -7061,6 +7124,7 @@ async function disbandGroupChat(group, members) { sessionOwners: room.sessionOwners || {}, members: Array.isArray(room.members) ? room.members : [], roomId: typeof room.roomId === 'string' && room.roomId ? room.roomId : null, + sessionScopeId: typeof room.sessionScopeId === 'string' && room.sessionScopeId ? room.sessionScopeId : null, image: room.image || null, syncRevision: Math.max(0, Number(room.syncRevision || 0)) } @@ -7279,19 +7343,53 @@ function uniqueGroupChatName(base, taken) { throw new Error('No free name for the group.') } -/** Ensure the member's per-group session exists and return a LIVE runtime +/** Ensure the member's per-group-thread session exists and return a LIVE runtime * session id for it. Gateway-native: session.create mints the session * (lazy until its first message), session.resume by stored id — or by * title, which also covers rehydrated rooms whose sid was lost — reopens * it after restarts. Cross-connection members route to their OWN source * via requestForBot; the window's gateway never switches. */ -async function ensureGroupChatSession(group, member) { +async function ensureGroupChatSession(group, member, thread = null) { + const legacyOwner = thread === null || thread === undefined + const normalizedThread = legacyOwner ? null : String(thread || 'legacy') + const room = $groupChats.get()[group] + + if (!room || room.tombstone) { + throw new Error(`Group ${group} no longer exists`) + } + + const scope = legacyOwner ? room.roomId || group : ensureGroupSessionScope(group) + const key = legacyOwner ? groupMemberKey(member) : groupThreadMemberKey(normalizedThread, member) + const owner = `${scope}::${key}` + const inflight = groupSessionFlights.get(owner) + + if (inflight) { + return inflight + } + + const run = ensureGroupChatSessionOnce(group, member, normalizedThread, legacyOwner, scope, key) + .finally(() => { + if (groupSessionFlights.get(owner) === run) { + groupSessionFlights.delete(owner) + } + }) + + groupSessionFlights.set(owner, run) + + return run +} + +async function ensureGroupChatSessionOnce(group, member, thread, legacyOwner, scope, key) { const room = $groupChats.get()[group] || {} - // New rooms title member sessions by their immutable roomId so a - // same-name recreate never resumes the old room's sessions by title; - // legacy rooms without a roomId fall back to the display name. - const title = `Group: ${room.roomId || group}` - const key = groupMemberKey(member) + // Calls without a thread retain the pre-thread title/pointer contract for + // tests and old integrations. Every live room turn passes a stable thread. + const title = legacyOwner + ? `Group: ${room.roomId || group}` + : `Group: ${scope} / Thread: ${thread}` + + if (!legacyOwner && title.length > 100) { + throw new Error('Group thread session title exceeds the 100-character limit') + } const known = room.sessions && room.sessions[key] // Try resuming what we know (stored sid first, then title lookup). @@ -7322,14 +7420,22 @@ async function ensureGroupChatSession(group, member) { const stored = res.session_key || known if (stored) { - updateGroupChat(group, current => { - current.sessions = { ...(current.sessions || {}), [key]: stored } - current.sessionOwners = { ...(current.sessionOwners || {}), [key]: groupSessionOwner(member) } - return current - }) + if (legacyOwner) { + updateGroupChat(group, current => { + current.sessions = { ...(current.sessions || {}), [key]: stored } + current.sessionOwners = { ...(current.sessionOwners || {}), [key]: groupSessionOwner(member) } + return current + }) + } else { + updateGroupSessionPointer(group, scope, key, stored, member) + } } - return { runtime: res.session_id, stored } + return { + runtime: res.session_id, + stored, + bootstrap: !legacyOwner && Number(res.message_count) === 0 + } } } catch (error) { if (error?.code !== 4007) { @@ -7349,14 +7455,18 @@ async function ensureGroupChatSession(group, member) { const stored = created?.stored_session_id || null if (stored) { - updateGroupChat(group, r => { - r.sessions = { ...(r.sessions || {}), [key]: stored } - r.sessionOwners = { ...(r.sessionOwners || {}), [key]: groupSessionOwner(member) } - return r - }) + if (legacyOwner) { + updateGroupChat(group, r => { + r.sessions = { ...(r.sessions || {}), [key]: stored } + r.sessionOwners = { ...(r.sessionOwners || {}), [key]: groupSessionOwner(member) } + return r + }) + } else { + updateGroupSessionPointer(group, scope, key, stored, member) + } } - return { runtime: created?.session_id || null, stored } + return { runtime: created?.session_id || null, stored, bootstrap: !legacyOwner } } const GROUP_TURN_TIMEOUT_MS = 180000 @@ -7465,8 +7575,10 @@ const GROUP_TURN_HARD_CAP_MS = 20 * 60000 * payload always sync to "no prompt". Clarify wins when both are somehow * present (approvals resolve inside tool batches; clarify is the outer * blocker). */ -function syncGroupClarify(group, member, state) { - const key = `${group}::${groupMemberKey(member)}` +function syncGroupClarify(group, member, state, thread = 'legacy') { + const room = $groupChats.get()[group] || {} + const roomScope = groupSessionScopeForRoom(room, group) + const key = `${roomScope}::${groupThreadMemberKey(thread, member)}` const clarify = state && typeof state.pending_clarify === 'object' ? state.pending_clarify : null const approval = state && typeof state.pending_approval === 'object' ? state.pending_approval : null const pending = clarify || approval @@ -7492,7 +7604,10 @@ function syncGroupClarify(group, member, state) { const base = { requestId, + ownerKey: key, group, + roomScope, + thread, member: member.name, memberKey: groupMemberKey(member), // approval.respond keys on the session, not just the request — carry the @@ -7586,7 +7701,7 @@ async function answerGroupClarify(entry, member, answers) { } const all = $groupClarify.get() - const key = `${entry.group}::${entry.memberKey}` + const key = entry.ownerKey || `${entry.roomScope || entry.group}::${entry.thread || 'legacy'}::${entry.memberKey}` if (all[key]?.requestId === entry.requestId) { const next = { ...all } @@ -7602,7 +7717,19 @@ async function answerGroupClarify(entry, member, answers) { * so slow models aren't cut off mid-run. A turn that still times out * records a stranded marker so the finished reply can be harvested into * the room at the member's next turn instead of being lost. */ -async function runGroupChatMemberTurn(group, member, prompt, thread, images) { +async function runGroupChatMemberTurn( + group, + member, + prompt, + thread, + images, + ensuredSession = null, + routeLeaseHeld = false +) { + if (routeLeaseHeld) { + return runGroupChatMemberTurnLeased(group, member, prompt, thread, images, ensuredSession) + } + // #93602: hold the member's route socket for the whole turn. Without the // lease, every RPC below rides its own request-scoped socket lease; the // socket that minted `runtime` can close between RPCs, the gateway reaps @@ -7610,14 +7737,14 @@ async function runGroupChatMemberTurn(group, member, prompt, thread, images) { const releaseTurnLease = await retainGroupTurnRoute(member) try { - return await runGroupChatMemberTurnLeased(group, member, prompt, thread, images) + return await runGroupChatMemberTurnLeased(group, member, prompt, thread, images, ensuredSession) } finally { releaseTurnLease() } } -async function runGroupChatMemberTurnLeased(group, member, prompt, thread, images) { - const { runtime, stored } = await ensureGroupChatSession(group, member) +async function runGroupChatMemberTurnLeased(group, member, prompt, thread, images, ensuredSession = null) { + const { runtime, stored } = ensuredSession || await ensureGroupChatSession(group, member, thread) if (!runtime) { return null @@ -7714,7 +7841,7 @@ async function runGroupChatMemberTurnLeased(group, member, prompt, thread, image // A clarify blocking inside the member's session is a question for the // HUMAN (#90694) — mirror it into the room store so a card renders, and // hold the turn open: the member isn't stalling, it's waiting on us. - const awaitingUser = syncGroupClarify(group, member, state) + const awaitingUser = syncGroupClarify(group, member, state, thread) const done = !busy && !awaitingUser if (messages.length > before && done) { @@ -7757,9 +7884,10 @@ async function runGroupChatMemberTurnLeased(group, member, prompt, thread, image // (runtime-only) so the finished reply can be posted late into the RIGHT // thread instead of vanishing. recordGroupActivity(group, { kind: 'timed-out', member: member.name, thread }) - syncGroupClarify(group, member, null) + syncGroupClarify(group, member, null, thread) updateGroupChat(group, r => { - r.stranded = { ...(r.stranded || {}), [groupMemberKey(member)]: { before, thread } } + const sessionKey = groupThreadMemberKey(thread, member) + r.stranded = { ...(r.stranded || {}), [sessionKey]: { before, thread, sessionKey } } return r }) @@ -7769,10 +7897,26 @@ async function runGroupChatMemberTurnLeased(group, member, prompt, thread, image /** Post a timed-out member's finished reply into the room, if it landed * after we stopped waiting. Called at the member's next turn boundary and * on user sends, so long-running work is delivered late rather than lost. */ -async function harvestStrandedGroupReply(group, member) { +async function harvestStrandedGroupReply(group, member, thread = null) { + const memberKey = groupMemberKey(member) + const room = $groupChats.get()[group] || {} + const markers = Object.entries(room.stranded || {}).filter(([key]) => { + if (key === memberKey) { + return true + } + + return key.endsWith(`::${memberKey}`) + && (thread === null || key === groupThreadMemberKey(thread, member)) + }) + + for (const [markerKey, marker] of markers) { + await harvestOneStrandedGroupReply(group, member, markerKey, marker) + } +} + +async function harvestOneStrandedGroupReply(group, member, markerKey, marker) { const memberKey = groupMemberKey(member) const room = $groupChats.get()[group] || {} - const marker = room.stranded?.[memberKey] // Markers were a bare number before threads; normalize both shapes. const strandedBefore = typeof marker === 'number' ? marker : marker?.before const strandedThread = (typeof marker === 'object' && marker?.thread) || 'legacy' @@ -7784,9 +7928,14 @@ async function harvestStrandedGroupReply(group, member) { let state = null try { - const stored = room.sessions?.[memberKey] + const sessionKey = typeof marker === 'object' && marker?.sessionKey ? marker.sessionKey : markerKey + const legacyMarker = markerKey === memberKey + const stored = room.sessions?.[sessionKey] || (legacyMarker ? room.sessions?.[memberKey] : null) + const title = legacyMarker + ? `Group: ${room.roomId || group}` + : `Group: ${groupSessionScopeForRoom(room, group)} / Thread: ${strandedThread}` state = await requestForBot(member, 'session.resume', { - session_id: stored || `Group: ${room.roomId || group}`, + session_id: stored || title, profile: member.name }) } catch { @@ -7799,14 +7948,16 @@ async function harvestStrandedGroupReply(group, member) { // A stranded member blocked on a clarify is not "grinding" — surface the // question card (#90694) and keep the marker until it resolves. - if (syncGroupClarify(group, member, state)) { + if (syncGroupClarify(group, member, state, strandedThread)) { return } // Done (or dead): the marker is consumed either way. updateGroupChat(group, r => { const next = { ...(r.stranded || {}) } - delete next[memberKey] + if (next[markerKey] === marker) { + delete next[markerKey] + } r.stranded = next return r }) @@ -7847,6 +7998,13 @@ async function harvestStrandedGroupReply(group, member) { } } +function hasStrandedGroupOwner(stranded, thread, member) { + const memberKey = groupMemberKey(member) + + return Object.prototype.hasOwnProperty.call(stranded || {}, memberKey) + || Object.prototype.hasOwnProperty.call(stranded || {}, groupThreadMemberKey(thread, member)) +} + // --- room-turn decision helpers (#93127) — pure, vm-sliced by tests --- /** #93127: whether a finished member turn may still commit (append its reply @@ -8000,7 +8158,7 @@ async function runGroupChatRounds(group, members, thread) { return } - await harvestStrandedGroupReply(group, member) + await harvestStrandedGroupReply(group, member, thread) } const roomLog = (($groupChats.get()[group] || {}).log || []).filter(e => groupThreadOf(e) === thread) @@ -8018,7 +8176,7 @@ async function runGroupChatRounds(group, members, thread) { // {before, thread} post-thread. const strandedNow = ($groupChats.get()[group] || {}).stranded || {} const responders = rotateGroupSpeakers(resolveGroupResponders(roomLog, members), round) - .filter(member => !Object.prototype.hasOwnProperty.call(strandedNow, groupMemberKey(member))) + .filter(member => !hasStrandedGroupOwner(strandedNow, thread, member)) let spokeThisRound = 0 for (const member of responders) { @@ -8033,13 +8191,8 @@ async function runGroupChatRounds(group, members, thread) { 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 => groupThreadOf(e) === thread) - - if (!delta.length) { - continue - } + const incrementalDelta = room.log.slice(seen).filter(e => groupThreadOf(e) === thread) + const bootstrapDelta = room.log.filter(e => groupThreadOf(e) === thread) // #93129: a member the user told to stop is HELD — no turn until an // explicit release (resume / @all resume / a direct non-stop @@ -8049,6 +8202,10 @@ async function runGroupChatRounds(group, members, thread) { const heldEntry = (room.holds || {})[memberKey] if (heldEntry) { + if (!incrementalDelta.length) { + continue + } + const advance = heldMemberWatermarkAdvance(seen, room.log.length) updateGroupChat(group, r => { @@ -8070,49 +8227,89 @@ async function runGroupChatRounds(group, members, thread) { continue } - const prompt = buildGroupChatTurnPrompt({ - groupName: group, - members, - viewer: member, - deltaLines: delta.slice(-GROUP_CHAT_HISTORY_LIMIT).map(e => formatGroupChatLine(e, member.name)) - }) + const releaseRoundTurnLease = await retainGroupTurnRoute(member) + let reply = null - // 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 => (Array.isArray(e.images) ? e.images : [])) + try { + let ensuredSession = null - // 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 => { - r.turn = member.name - return r - }) + try { + ensuredSession = await ensureGroupChatSession(group, member, thread) + } catch (error) { + const reason = String(error?.data?.reason || '').trim() + recordGroupActivity(group, { + kind: 'failed', + member: member.name, + thread, + ...(reason ? { reason } : {}) + }) + noteBotAttention(groupMemberKey(member), reason || error?.message || error) + } - let reply = null + if (ensuredSession) { + // A newly created (or still-empty) per-thread session has never seen + // this topic, even when a legacy shared-session watermark says the + // member already consumed it. Bootstrap from retained history for + // this thread only; established sessions keep the incremental delta. + const delta = ensuredSession.bootstrap ? bootstrapDelta : incrementalDelta - 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)) + if (!delta.length) { + continue + } + + const prompt = buildGroupChatTurnPrompt({ + groupName: group, + members, + viewer: member, + deltaLines: delta.slice(-GROUP_CHAT_HISTORY_LIMIT).map(e => 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. + const deltaImages = delta.flatMap(e => (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 => { + r.turn = member.name + return r + }) + + try { + reply = await runGroupChatMemberTurn( + group, + member, + prompt, + thread, + deltaImages, + ensuredSession, + 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. + // 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) { + 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 + } } - } catch (error) { - 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 + } finally { + releaseRoundTurnLease() } // #93127: the turn may have finished AFTER a newer user send bumped @@ -8211,14 +8408,14 @@ async function harvestStrandedUntilSettled(group, members, thread) { const stranded = room.stranded || {} - if (!Object.keys(stranded).length) { + if (!members.some(member => hasStrandedGroupOwner(stranded, thread, member))) { return } for (const member of members) { - if (Object.prototype.hasOwnProperty.call(stranded, groupMemberKey(member))) { + if (hasStrandedGroupOwner(stranded, thread, member)) { try { - await harvestStrandedGroupReply(group, member) + await harvestStrandedGroupReply(group, member, thread) } catch { // Best-effort: the next tick retries; the bound stops runaways. } @@ -15397,6 +15594,7 @@ export default { holds: room.holds && typeof room.holds === 'object' ? room.holds : {}, members: Array.isArray(room.members) ? room.members : [], roomId: typeof room.roomId === 'string' && room.roomId ? room.roomId : null, + sessionScopeId: typeof room.sessionScopeId === 'string' && room.sessionScopeId ? room.sessionScopeId : null, image: typeof room.image === 'string' && room.image ? room.image : null, syncRevision: Math.max(0, Number(room.syncRevision || 0)), epoch: 0, diff --git a/apps/desktop/src/plugins/hermes-bots/tests/group-chat.test.mjs b/apps/desktop/src/plugins/hermes-bots/tests/group-chat.test.mjs index 1bcbd2c85dba4..56987adb655d2 100644 --- a/apps/desktop/src/plugins/hermes-bots/tests/group-chat.test.mjs +++ b/apps/desktop/src/plugins/hermes-bots/tests/group-chat.test.mjs @@ -8,7 +8,7 @@ const pluginSource = readFileSync(new URL('../plugin.js', import.meta.url), 'utf /** Load the plugin in a vm with a scripted cli.exec so member turns are * deterministic. `turnScript(profile, prompt)` returns the member's reply * text (or throws to simulate a failed turn). */ -function load(turnScript, { busyUntilResumeCall, clarifyUntilResumeCall, approvalUntilResumeCall, conflictOnce = false, deferredTimers = false } = {}) { +function load(turnScript, { busyUntilResumeCall, clarifyUntilResumeCall, approvalUntilResumeCall, conflictOnce = false, deferredTimers = false, sessionCreateFailure = null } = {}) { const values = new Map() const atom = initial => { const slot = { get: () => values.get(slot), set: value => { @@ -118,6 +118,11 @@ function load(turnScript, { busyUntilResumeCall, clarifyUntilResumeCall, approva return { applied: { ui_meta: true, ui_meta_revisions: { ...uiMetaRevisions } } } } if (method === 'session.create') { + if (sessionCreateFailure?.profile === params.profile) { + const error = new Error(sessionCreateFailure.message || 'session create failed') + error.data = { reason: sessionCreateFailure.reason } + throw error + } sessionSequence += 1 const stored = `sid-${params.profile}-${sessionSequence}` const runtime = `rt-${params.profile}-${sessionSequence}` @@ -204,7 +209,7 @@ function load(turnScript, { busyUntilResumeCall, clarifyUntilResumeCall, approva .replace(/^import .* from 'react\/jsx-runtime'\r?\n/m, '') .replace('export default {', 'globalThis.plugin = {') .concat( - '\nglobalThis.__gc = { sendToGroupChat, runGroupChatRounds, harvestStrandedGroupReply, resolveGroupResponders, parseGroupChatMentions, rotateGroupSpeakers, isGroupPassText, formatGroupChatLine, groupSpeakerLabel, buildGroupChatTurnPrompt, trimGroupChatLog, groupChatSyncSnapshot, groupChatGatewayJsonSize, mergeGroupChatSyncSnapshots, mergeRemoteGroupChatSnapshotIntoRooms, scheduleGroupChatServerSync, disbandGroupChat, renameGroupChat, updateGroupChat, durableGroupChatRooms, persistGroupChatRooms, ensureGroupChatSession, uniqueGroupChatName, liveGroupChatNames, groupChatNames, openGroupChat, closeGroupChatMainTab, shouldRenderGroupChatInPane, syncGroupClarify, clearGroupClarify, answerGroupClarify, $groupClarify, $groupChats, $groupNeedsYou, $groupChatWorkspace, $groupMainTabsRev, $botMeta, $lastRoster, GROUP_CHAT_MAX_ROUNDS, GROUP_CHAT_MAX_MESSAGES };\n' + '\nglobalThis.__gc = { sendToGroupChat, runGroupChatRounds, harvestStrandedGroupReply, resolveGroupResponders, parseGroupChatMentions, rotateGroupSpeakers, isGroupPassText, formatGroupChatLine, groupSpeakerLabel, buildGroupChatTurnPrompt, trimGroupChatLog, groupChatSyncSnapshot, groupChatGatewayJsonSize, mergeGroupChatSyncSnapshots, mergeRemoteGroupChatSnapshotIntoRooms, scheduleGroupChatServerSync, disbandGroupChat, renameGroupChat, updateGroupChat, durableGroupChatRooms, persistGroupChatRooms, ensureGroupChatSession, uniqueGroupChatName, liveGroupChatNames, groupChatNames, openGroupChat, closeGroupChatMainTab, shouldRenderGroupChatInPane, syncGroupClarify, clearGroupClarify, answerGroupClarify, $groupClarify, $groupChats, $groupActivity, $botAttention, $groupNeedsYou, $groupChatWorkspace, $groupMainTabsRev, $botMeta, $lastRoster, GROUP_CHAT_MAX_ROUNDS, GROUP_CHAT_MAX_MESSAGES };\n' ) vm.runInNewContext(source, context, { filename: 'plugin.js' }) const storageWrites = new Map() @@ -321,6 +326,31 @@ test('failed member turn is a pass, not a room error', async () => { assert.equal(log.length, 1) // just the user message; no error entries }) +test('failed thread-session ensure is one typed pass and consumes the delta', async () => { + const gc = load( + () => '(pass)', + { + sessionCreateFailure: { + profile: 'builder', + reason: 'provider_auth_or_access' + } + } + ) + + const thread = gc.sendToGroupChat('Ensure failure', MEMBERS, 'anyone around?') + for (let i = 0; i < 200 && (gc.$groupChats.get()['Ensure failure'] || {}).running; i++) { + await new Promise(resolve => setImmediate(resolve)) + } + + const room = gc.$groupChats.get()['Ensure failure'] + assert.equal(room.watermarks[`${thread}::builder`], room.log.length) + const failed = gc.$groupActivity.get()['Ensure failure'].events.find( + event => event.kind === 'failed' && event.member === 'builder' + ) + assert.equal(failed.reason, 'provider_auth_or_access') + assert.equal(gc.$botAttention.get().builder.reason, 'provider_auth_or_access') +}) + test('delta injection: a second user send only feeds members the NEW messages', async () => { const prompts = [] const gc = load((profile, prompt) => { @@ -348,8 +378,8 @@ test('concurrent groups sharing one member keep sessions, deltas, and context is const sharedMember = [{ name: 'research', title: '' }] // Start both rooms without waiting for either drive to finish. - gc.sendToGroupChat('Alpha', sharedMember, 'ALPHA_ONLY_1') - gc.sendToGroupChat('Beta', sharedMember, 'BETA_ONLY_1') + const alphaThread = gc.sendToGroupChat('Alpha', sharedMember, 'ALPHA_ONLY_1') + const betaThread = gc.sendToGroupChat('Beta', sharedMember, 'BETA_ONLY_1') for (let i = 0; i < 400; i++) { const rooms = gc.$groupChats.get() if (!rooms.Alpha?.running && !rooms.Beta?.running) { @@ -358,8 +388,8 @@ test('concurrent groups sharing one member keep sessions, deltas, and context is await new Promise(resolve => setImmediate(resolve)) } - const alphaFirst = gc.calls.find(call => call.title === 'Group: Alpha') - const betaFirst = gc.calls.find(call => call.title === 'Group: Beta') + const alphaFirst = gc.calls.find(call => call.prompt.includes('ALPHA_ONLY_1')) + const betaFirst = gc.calls.find(call => call.prompt.includes('BETA_ONLY_1')) assert.ok(alphaFirst && betaFirst, 'the shared member took one turn in each room') assert.notEqual(alphaFirst.stored, betaFirst.stored, 'each room owns a distinct stored session') assert.notEqual(alphaFirst.runtime, betaFirst.runtime, 'each room owns a distinct runtime session') @@ -369,14 +399,14 @@ test('concurrent groups sharing one member keep sessions, deltas, and context is assert.equal(betaFirst.prompt.includes('ALPHA_ONLY_1'), false) const roomsAfterFirst = gc.$groupChats.get() - assert.equal(roomsAfterFirst.Alpha.sessions.research, alphaFirst.stored) - assert.equal(roomsAfterFirst.Beta.sessions.research, betaFirst.stored) + assert.equal(roomsAfterFirst.Alpha.sessions[`${alphaThread}::research`], alphaFirst.stored) + assert.equal(roomsAfterFirst.Beta.sessions[`${betaThread}::research`], betaFirst.stored) // Interleave a second pair. Each room resumes its own session and receives // only its unseen room delta, never the sibling room's messages. const firstCallCount = gc.calls.length - gc.sendToGroupChat('Alpha', sharedMember, 'ALPHA_ONLY_2') - gc.sendToGroupChat('Beta', sharedMember, 'BETA_ONLY_2') + gc.sendToGroupChat('Alpha', sharedMember, 'ALPHA_ONLY_2', alphaThread) + gc.sendToGroupChat('Beta', sharedMember, 'BETA_ONLY_2', betaThread) for (let i = 0; i < 400; i++) { const rooms = gc.$groupChats.get() if (!rooms.Alpha?.running && !rooms.Beta?.running) { @@ -386,8 +416,8 @@ test('concurrent groups sharing one member keep sessions, deltas, and context is } const secondCalls = gc.calls.slice(firstCallCount) - const alphaSecond = secondCalls.find(call => call.title === 'Group: Alpha') - const betaSecond = secondCalls.find(call => call.title === 'Group: Beta') + const alphaSecond = secondCalls.find(call => call.prompt.includes('ALPHA_ONLY_2')) + const betaSecond = secondCalls.find(call => call.prompt.includes('BETA_ONLY_2')) assert.ok(alphaSecond && betaSecond, 'both rooms resumed for the second pair') assert.equal(alphaSecond.stored, alphaFirst.stored) assert.equal(betaSecond.stored, betaFirst.stored) @@ -404,6 +434,138 @@ test('concurrent groups sharing one member keep sessions, deltas, and context is assert.equal(betaSession.messages.some(message => String(message.content).includes('ALPHA_ONLY')), false) }) +test('one member in two room threads owns distinct sessions and backend transcripts', async () => { + const gc = load(() => '(pass)') + const member = { name: 'research', title: '' } + + gc.updateGroupChat('Core', room => { + room.roomId = 'r-core' + room.log = [ + { id: 'a-1', at: 1, from: { kind: 'user', name: 'You' }, text: 'THREAD_A_ONLY', thread: 'thread-a' }, + { id: 'b-1', at: 2, from: { kind: 'user', name: 'You' }, text: 'THREAD_B_ONLY', thread: 'thread-b' } + ] + return room + }) + + await gc.runGroupChatRounds('Core', [member], 'thread-a') + await gc.runGroupChatRounds('Core', [member], 'thread-b') + + const threadA = gc.calls.find(call => call.prompt.includes('THREAD_A_ONLY')) + const threadB = gc.calls.find(call => call.prompt.includes('THREAD_B_ONLY')) + assert.ok(threadA && threadB) + assert.notEqual(threadA.stored, threadB.stored) + assert.equal(gc.sessions.get(threadA.stored).messages.some(message => String(message.content).includes('THREAD_B_ONLY')), false) + assert.equal(gc.sessions.get(threadB.stored).messages.some(message => String(message.content).includes('THREAD_A_ONLY')), false) +}) + +test('the same room thread and member reuse one hidden session', async () => { + const gc = load(() => '(pass)') + const member = { name: 'research', title: '' } + + gc.updateGroupChat('Core', room => { + room.roomId = 'r-core' + return room + }) + + const first = await gc.ensureGroupChatSession('Core', member, 'thread-a') + const second = await gc.ensureGroupChatSession('Core', member, 'thread-a') + + assert.equal(second.stored, first.stored) + assert.equal(gc.requests.filter(request => request.method === 'session.create').length, 1) + assert.equal(gc.$groupChats.get().Core.sessions['thread-a::research'], first.stored) +}) + +test('a first thread session bootstraps retained history for only that thread and preserves the legacy pointer', async () => { + const gc = load(() => '(pass)') + const member = { name: 'research', title: '' } + + gc.updateGroupChat('Core', room => { + room.roomId = 'r-core' + room.log = [ + { id: 'a-1', at: 1, from: { kind: 'user', name: 'You' }, text: 'RETAINED_A', thread: 'thread-a' }, + { id: 'b-1', at: 2, from: { kind: 'user', name: 'You' }, text: 'RETAINED_B', thread: 'thread-b' } + ] + room.watermarks = { 'thread-a::research': room.log.length } + room.sessions = { research: 'sid-legacy-shared' } + return room + }) + + await gc.runGroupChatRounds('Core', [member], 'thread-a') + + const call = gc.calls.find(candidate => candidate.prompt.includes('RETAINED_A')) + assert.ok(call, 'the empty per-thread session receives its retained thread history') + assert.equal(call.prompt.includes('RETAINED_B'), false) + assert.equal(gc.$groupChats.get().Core.sessions.research, 'sid-legacy-shared') + assert.equal(gc.$groupChats.get().Core.sessions['thread-a::research'], call.stored) +}) + +test('concurrent ensure for one room thread and member creates one hidden session', async () => { + const gc = load(() => '(pass)') + const member = { name: 'research', title: '' } + + gc.updateGroupChat('Core', room => { + room.roomId = 'r-core' + return room + }) + + const [first, second] = await Promise.all([ + gc.ensureGroupChatSession('Core', member, 'thread-a'), + gc.ensureGroupChatSession('Core', member, 'thread-a') + ]) + + assert.equal(second.stored, first.stored) + assert.equal(gc.requests.filter(request => request.method === 'session.create').length, 1) + assert.equal(gc.sessions.size, 1) +}) + +test('thread session scope stays length-safe and stable across a room rename', async () => { + const gc = load(() => '(pass)') + const member = { name: 'research', title: '' } + + gc.updateGroupChat('Long Room', room => { + room.roomId = 'r'.repeat(120) + return room + }) + + const first = await gc.ensureGroupChatSession('Long Room', member, 'thread-a') + const beforeRename = gc.$groupChats.get()['Long Room'] + const scope = beforeRename.sessionScopeId + assert.ok(scope) + assert.ok(gc.sessions.get(first.stored).title.length <= 100) + assert.equal(gc.storageWrites.get('group-chats')['Long Room'].sessionScopeId, scope) + + await gc.renameGroupChat('Long Room', 'Renamed Room', [member]) + gc.updateGroupChat('Renamed Room', room => { + const sessions = { ...room.sessions } + delete sessions['thread-a::research'] + room.sessions = sessions + return room + }) + + const resumed = await gc.ensureGroupChatSession('Renamed Room', member, 'thread-a') + assert.equal(gc.$groupChats.get()['Renamed Room'].sessionScopeId, scope) + assert.equal(resumed.stored, first.stored) +}) + +test('a late thread ensure cannot resurrect a disbanded room', async () => { + const gc = load(() => '(pass)') + const member = { name: 'research', title: '' } + + gc.updateGroupChat('Gone', room => { + room.roomId = 'r'.repeat(120) + room.members = [member] + return room + }) + await gc.disbandGroupChat('Gone', [member]) + + await assert.rejects( + () => gc.ensureGroupChatSession('Gone', member, 'thread-a'), + /no longer exists/ + ) + assert.equal(gc.$groupChats.get().Gone, undefined) + assert.equal(gc.storageWrites.get('group-chats')?.Gone, undefined) +}) + test('recreating a same-name group after disband mints fresh member sessions', async () => { const gc = load(() => '(pass)') const member = { name: 'research', title: '' } @@ -577,9 +739,9 @@ test('turn transport is gateway-native (session RPCs) and hostile text rides ver assert.equal(call.profile, 'research') // Hostile text is a JSON string in an RPC param — never a shell string. assert.equal(call.prompt.includes('hello "there" `whoami` $(id)'), true) - // The per-group session is created with the room title. + // The per-thread session is created with stable room + thread identity. assert.match(pluginSource, /title,\n/) - assert.match(pluginSource, /const title = `Group: \$\{room\.roomId \|\| group\}`/) + assert.match(pluginSource, /`Group: \$\{scope\} \/ Thread: \$\{thread\}`/) }) test('log trimming keeps watermarks consistent', () => { @@ -1527,6 +1689,46 @@ test('stranded harvest: a timed-out turn whose reply landed late posts into the assert.equal(gc.$groupChats.get().Late.stranded.research, undefined, 'marker consumed') }) +test('thread-scoped stranded harvest resumes and clears only that thread session', async () => { + const gc = load(() => '(pass)') + const member = { name: 'research', title: '' } + + gc.updateGroupChat('Late', room => { + room.roomId = 'r-late' + return room + }) + const threadA = await gc.ensureGroupChatSession('Late', member, 'thread-a') + const threadB = await gc.ensureGroupChatSession('Late', member, 'thread-b') + gc.sessions.get(threadA.stored).messages.push({ role: 'assistant', content: 'LATE_A_ONLY' }) + gc.sessions.get(threadB.stored).messages.push({ role: 'assistant', content: 'LATE_B_ONLY' }) + gc.updateGroupChat('Late', room => { + room.stranded = { + 'thread-a::research': { before: 0, thread: 'thread-a', sessionKey: 'thread-a::research' }, + 'thread-b::research': { before: 0, thread: 'thread-b', sessionKey: 'thread-b::research' } + } + return room + }) + + await gc.harvestStrandedGroupReply('Late', member, 'thread-a') + + const room = gc.$groupChats.get().Late + assert.equal(room.log.some(entry => entry.thread === 'thread-a' && entry.text === 'LATE_A_ONLY'), true) + assert.equal(room.log.some(entry => entry.text === 'LATE_B_ONLY'), false) + assert.equal(room.stranded['thread-a::research'], undefined) + assert.ok(room.stranded['thread-b::research']) +}) + +test('background stranded harvest stops when only sibling-thread markers remain', () => { + const start = pluginSource.indexOf('async function harvestStrandedUntilSettled') + const end = pluginSource.indexOf('/** User send into a group room.', start) + const source = pluginSource.slice(start, end) + + assert.match( + source, + /if \(!members\.some\(member => hasStrandedGroupOwner\(stranded, thread, member\)\)\) \{\s*return\s*\}/ + ) +}) + test('stranded + still busy: the round loop never re-submits into a member whose harvest just confirmed they are still running', async () => { // research is confirmed busy on exactly its first two session.resume // calls — the number of harvest-only touches the FIXED code makes across @@ -1834,6 +2036,33 @@ test('syncGroupClarify mirrors, badges needs-you, and is idempotent per request' assert.equal(Object.keys(gc.$groupClarify.get()).length, 0) }) +test('clarify ownership is isolated by room thread and member', () => { + const gc = load(() => '(pass)') + const member = { name: 'research', title: '' } + gc.updateGroupChat('Core', room => { + room.roomId = 'r-core' + return room + }) + + gc.syncGroupClarify('Core', member, { + session_id: 'rt-a', + pending_clarify: { ...CLARIFY_PAYLOAD, request_id: 'req-a' } + }, 'thread-a') + gc.syncGroupClarify('Core', member, { + session_id: 'rt-b', + pending_clarify: { ...CLARIFY_PAYLOAD, request_id: 'req-b' } + }, 'thread-b') + + const entries = Object.values(gc.$groupClarify.get()) + assert.equal(entries.length, 2) + assert.deepEqual([...entries.map(entry => entry.thread)].sort(), ['thread-a', 'thread-b']) + + gc.syncGroupClarify('Core', member, {}, 'thread-a') + const remaining = Object.values(gc.$groupClarify.get()) + assert.equal(remaining.length, 1) + assert.equal(remaining[0].thread, 'thread-b') +}) + test('older backends without pending_clarify never mirror a question', () => { const gc = load(() => '(pass)') @@ -1895,7 +2124,7 @@ test('disband clears the room mirrored questions', async () => { test('source contract: room renders clarify cards and the poll gates on them', () => { assert.match(pluginSource, /function GroupClarifyCard\(/) - assert.match(pluginSource, /syncGroupClarify\(group, member, state\)/) + assert.match(pluginSource, /syncGroupClarify\(group, member, state, thread\)/) assert.match(pluginSource, /const done = !busy && !awaitingUser/) assert.match(pluginSource, /busy \|\| awaitingUser/) assert.match(pluginSource, /roomClarifies\.map\(entry =>/) diff --git a/apps/desktop/src/plugins/hermes-bots/tests/group-turn-lease.test.mjs b/apps/desktop/src/plugins/hermes-bots/tests/group-turn-lease.test.mjs index 109682ef05e2b..e58757e5032b2 100644 --- a/apps/desktop/src/plugins/hermes-bots/tests/group-turn-lease.test.mjs +++ b/apps/desktop/src/plugins/hermes-bots/tests/group-turn-lease.test.mjs @@ -170,13 +170,28 @@ function load({ failFirstSubmitWith = null, failEverySubmitWith = null, reply = .replace(/^import .* from 'react\/jsx-runtime'\r?\n/m, '') .replace('export default {', 'globalThis.plugin = {') .concat( - '\nglobalThis.__lease = { runGroupChatMemberTurn, submitGroupTurnPrompt, isSessionGoneError, $groupChats };\n' + '\nglobalThis.__lease = { runGroupChatMemberTurn, runGroupChatRounds, submitGroupTurnPrompt, isSessionGoneError, $groupChats };\n' ) vm.runInNewContext(source, context, { filename: 'plugin.js' }) context.plugin.register({ storage: { get: () => null, set: () => undefined }, register: () => undefined }) + // runGroupChatMemberTurn is an internal room-member seam. Production only + // calls it after the room exists; establish that invariant here too. The + // thread-isolation contract keys its durable session by thread + member. + context.__lease.$groupChats.set({ + Room: { + log: [], + watermarks: {}, + sessions: {}, + sessionOwners: {}, + stranded: {}, + members: [LOCAL_MEMBER], + roomId: 'room-lease-test', + sessionScopeId: 'scope-lease-test' + } + }) return { ...context.__lease, context, @@ -215,7 +230,7 @@ test('a 4001 on the first prompt.submit recovers via session.resume on the STORE assert.equal(gc.stats().submits, 2) // The recovery re-resumed the durable stored id, not the dead runtime id. const room = gc.$groupChats.get().Room - assert.ok(room.sessions.helper, 'stored session id survives in the room record') + assert.ok(room.sessions['t1::helper'], 'thread-scoped stored session id survives in the room record') }) test('a persistent non-4001 submit failure is NOT retried and still surfaces', async () => { @@ -253,6 +268,29 @@ test('the per-turn lease is released after the turn — refcount returns to zero assert.equal(gc.stats().disposals, 1) }) +test('the production round path retains the route before ensure/create and through the turn', async () => { + const gc = load({ reply: '(pass)' }) + const room = gc.$groupChats.get().Room + gc.$groupChats.set({ + Room: { + ...room, + log: [{ id: 'user:1', from: { kind: 'user', name: 'You' }, text: 'hello', at: 1, thread: 't1' }] + } + }) + + await gc.runGroupChatRounds('Room', [ROUTED_MEMBER], 't1') + + assert.equal(gc.timeline[0], 'retain') + const firstRelease = gc.timeline.indexOf('release') + assert.ok(firstRelease > gc.timeline.indexOf('session.create'), 'lease covers session creation') + assert.ok(firstRelease > gc.timeline.indexOf('prompt.submit'), 'lease covers prompt submit') + for (const rpc of gc.rpcLog) { + assert.ok(rpc.refcountAfter >= 1, `${rpc.method} left refcount ${rpc.refcountAfter} before turn release`) + } + assert.equal(gc.timeline.at(-1), 'release') + assert.equal(gc.stats().refcount, 0) +}) + test('the per-turn lease is released even when the turn fails', async () => { const fatal = new Error('backend exploded') const gc = load({ failEverySubmitWith: fatal })