diff --git a/apps/desktop/src/plugins/hermes-bots/plugin.js b/apps/desktop/src/plugins/hermes-bots/plugin.js index 2f1a4917cf..e620e41f9c 100644 --- a/apps/desktop/src/plugins/hermes-bots/plugin.js +++ b/apps/desktop/src/plugins/hermes-bots/plugin.js @@ -4290,7 +4290,21 @@ async function runGroupChatRounds(group, members, thread) { } const roomLog = (($groupChats.get()[group] || {}).log || []).filter(e => groupThreadOf(e) === thread) + // Exclude members the harvest pass just above confirmed are STILL + // running (their stranded marker survived harvest because + // state.inflight/running was true). Re-selecting one here would + // prompt.submit into their live session — the gateway's default busy + // policy redirects or hard-interrupts that turn (tui_gateway's + // _handle_busy_submit), killing exactly the long-running work this + // stranded/harvest mechanism exists to protect. Skip them; the next + // harvest pass picks the reply up once it actually lands. A marker's + // mere presence means "still stranded" (harvestStrandedGroupReply + // deletes it once the member is confirmed done/dead) — presence, not + // value shape, since markers are a bare number pre-thread or + // {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))) let spokeThisRound = 0 for (const member of responders) { 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 f7e2411983..1bacfca9ba 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) { +function load(turnScript, { busyUntilResumeCall } = {}) { const values = new Map() const atom = initial => { const slot = { get: () => values.get(slot), set: value => values.set(slot, value) } @@ -20,6 +20,12 @@ function load(turnScript) { const runtimeToStored = new Map() const titleToStored = new Map() let sessionSequence = 0 + // busyUntilResumeCall[profile] = N: this profile's session.resume reports + // inflight/running for its first N calls, then flips to done — simulating + // a turn that is genuinely still running when harvested, without an + // unbounded real-time wait if a caller polls it (used to exercise the + // stranded/busy-responder guard). + const resumeCallCounts = new Map() const resolveSession = (profile, target) => { const stored = runtimeToStored.get(target) || (sessions.has(target) ? target : titleToStored.get(`${profile}::${target}`)) @@ -52,13 +58,18 @@ function load(turnScript) { if (!session) { throw new Error(`session not found: ${params.session_id}`) } + const profile = params.profile + const seen = (resumeCallCounts.get(profile) || 0) + 1 + resumeCallCounts.set(profile, seen) + const limit = busyUntilResumeCall && busyUntilResumeCall[profile] + const busy = Boolean(limit && seen <= limit) return { session_id: session.runtime, session_key: session.stored, message_count: session.messages.length, messages: [...session.messages], - inflight: false, - running: false + inflight: busy, + running: busy } } if (method === 'prompt.submit') { @@ -565,6 +576,62 @@ 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('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 + // two rounds. If the responder guard is missing, research gets re- + // selected and picks up two EXTRA resume calls of its own (session + // resolution + turn baseline) before its very first post-resubmit poll + // — call #4 — which then reports done, so the mutated run still finishes + // fast (no real wall-clock wait) while still proving the resubmission + // happened. + const gc = load(profile => (profile === 'builder' ? 'builder here, all good' : '(pass)'), { + busyUntilResumeCall: { research: 2 } + }) + + // research's session is pre-seeded and resolvable, exactly like the + // sibling stranded-harvest test above — its stranded marker is the + // pre-thread bare-number shape (still supported: harvestStrandedGroupReply + // normalizes both shapes, and presence in `stranded` is what the round-loop + // guard checks, not the marker's value shape). + gc.sessions.set('sid-research', { + stored: 'sid-research', + runtime: 'rt-research', + profile: 'research', + title: 'Group: Grind', + messages: [] + }) + + // research is already stranded from an earlier (unmodeled) timeout, and + // its session is STILL genuinely running per session.resume. Both members + // are mentioned, so without the busy-responder guard both would be + // re-selected this round — and resubmitting into research's live session + // would trigger the gateway's default busy policy (redirect/hard- + // interrupt), destroying the very turn the stranded marker exists to wait + // out. + gc.updateGroupChat('Grind', r => { + r.stranded = { research: 0 } + r.sessions = { research: 'sid-research' } + r.log = [{ from: { kind: 'user', name: 'You' }, text: '@research @builder status?', at: 1 }] + r.watermarks = { 'legacy::research': 0, 'legacy::builder': 0 } + return r + }) + + await gc.runGroupChatRounds('Grind', [{ name: 'research', title: '' }, { name: 'builder', title: '' }], 'legacy') + + assert.equal( + gc.calls.filter(c => c.profile === 'research').length, + 0, + 'research (still busy) must never receive a new prompt.submit' + ) + assert.equal( + gc.$groupChats.get().Grind.stranded.research, + 0, + 'marker survives untouched — harvest confirmed research is still running' + ) + assert.equal(gc.calls.filter(c => c.profile === 'builder').length, 1, 'builder (not stranded) still gets its turn') +}) + test('stranded harvest: a late (pass) or no-new-message consumes the marker without posting', async () => { const gc = load(() => '(pass)')