From b9cc9eb71cb6aeb9518ccdfcc9627f12741dea85 Mon Sep 17 00:00:00 2001 From: dmitrii-f-t27 Date: Fri, 2 Oct 2026 22:12:12 -0300 Subject: [PATCH] fix(queen): bees keep their context and survive a release Summaries: of 245 compactions on the runners in fifty minutes 194 came back empty, 158 of them at the fixed 60 s abort. TRIOS_COMPACTION_SUMMARIZER_MODELS lets a fast model on the bee's own key write the summary first (the bee's own model still gets its turn), and TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS names a longer wait (10 s..10 min, default 60 s). Drain: the server left ~2 s after SIGTERM and took every bee with it. With TRIOS_BEE_RUNNER_DRAIN_SECONDS a runner takes no new orders and waits for its bees; at the deadline it salvages and stores the branch of each one still running. The entrypoint now waits for a draining server instead of leaving at the first TERM. --- .../apps/server/src/agent/ai-sdk-agent.ts | 24 +++- .../apps/server/src/agent/compaction.ts | 79 +++++++---- .../apps/server/src/agent/compaction/utils.ts | 47 ++++++- .../server/src/api/services/queen-runner.ts | 123 +++++++++++++++-- trios/agent-server/apps/server/src/main.ts | 5 +- .../server/tests/agent/compaction-e2e.test.ts | 82 +++++++++++ .../agent/compaction-summarizer-env.test.ts | 87 ++++++++++++ .../server/tests/api/queen-runner.test.ts | 128 ++++++++++++++++++ trios/agent-server/docker-entrypoint.sh | 7 + 9 files changed, 544 insertions(+), 38 deletions(-) create mode 100644 trios/agent-server/apps/server/tests/agent/compaction-summarizer-env.test.ts diff --git a/trios/agent-server/apps/server/src/agent/ai-sdk-agent.ts b/trios/agent-server/apps/server/src/agent/ai-sdk-agent.ts index 357a25e4aa..32edbb2f10 100644 --- a/trios/agent-server/apps/server/src/agent/ai-sdk-agent.ts +++ b/trios/agent-server/apps/server/src/agent/ai-sdk-agent.ts @@ -1,4 +1,3 @@ -import { isBrowserlessByConfiguration } from './browserless' import { devToolsMiddleware } from '@ai-sdk/devtools' import type { LanguageModelV3, @@ -31,8 +30,10 @@ import { buildFilesystemToolSet } from '../tools/filesystem/build-toolset' import type { ToolContext } from '../tools/framework' import { buildMemoryToolSet } from '../tools/memory/build-toolset' import type { ToolRegistry } from '../tools/tool-registry' +import { isBrowserlessByConfiguration } from './browserless' import { CHAT_MODE_ALLOWED_TOOLS } from './chat-mode' import { createCompactionPrepareStep, type StepWithUsage } from './compaction' +import { summarizerModelFor } from './compaction/utils' import { buildMcpServerSpecs, createMcpClients, @@ -280,8 +281,29 @@ export class AiSdkAgent { }) // Configure compaction for context window management + // A summary written by a fast model on the bee's own key, when the + // operator named one for this provider (see summarizerModelFor). + const summarizerName = summarizerModelFor( + config.resolvedConfig.provider, + config.resolvedConfig.model, + ) + const summarizer = summarizerName + ? createLanguageModel({ + ...config.resolvedConfig, + model: summarizerName, + }) + : undefined + if (summarizerName) { + logger.info('Summaries by a separate model', { + conversationId: config.resolvedConfig.conversationId, + provider: config.resolvedConfig.provider, + beeModel: config.resolvedConfig.model, + summarizer: summarizerName, + }) + } const compactionPrepareStep = createCompactionPrepareStep({ contextWindow, + summarizer, }) const normalizationOptions = getMessageNormalizationOptions( config.resolvedConfig, diff --git a/trios/agent-server/apps/server/src/agent/compaction.ts b/trios/agent-server/apps/server/src/agent/compaction.ts index 953d1a209a..488fb79cbb 100644 --- a/trios/agent-server/apps/server/src/agent/compaction.ts +++ b/trios/agent-server/apps/server/src/agent/compaction.ts @@ -45,6 +45,24 @@ export { export interface CompactionConfig { contextWindow: number + /** A faster model on the bee's own key that writes summaries first. */ + summarizer?: LanguageModel +} + +/** + * Ask each model in turn and keep the first summary. The summarizer is tried + * first; when it is busy or refuses, the bee's own model still gets its turn, + * so a fast summarizer can only add summaries, never take one away. + */ +async function firstSummary( + models: LanguageModel[], + summarize: (model: LanguageModel) => Promise, +): Promise { + for (const model of models) { + const summary = await summarize(model) + if (summary) return summary + } + return null } async function consumeStreamText( @@ -130,7 +148,7 @@ async function summarizeTurnPrefix( } async function compactMessages( - model: LanguageModel, + models: LanguageModel[], messages: ModelMessage[], config: ComputedConfig, state: CompactionState, @@ -239,18 +257,22 @@ async function compactMessages( if (isSplitTurn && summarizedTurnPrefix.length > 0) { if (toSummarize.length > 0) { const [historySummary, turnPrefixSummary] = await Promise.all([ - summarizeMessages( - model, - toSummarize, - state.existingSummary, - config.summarizationTimeoutMs, - config.summarizerMaxOutputTokens, + firstSummary(models, (model) => + summarizeMessages( + model, + toSummarize, + state.existingSummary, + config.summarizationTimeoutMs, + config.summarizerMaxOutputTokens, + ), ), - summarizeTurnPrefix( - model, - summarizedTurnPrefix, - config.summarizationTimeoutMs, - turnPrefixOutputBudget, + firstSummary(models, (model) => + summarizeTurnPrefix( + model, + summarizedTurnPrefix, + config.summarizationTimeoutMs, + turnPrefixOutputBudget, + ), ), ]) @@ -260,20 +282,24 @@ async function compactMessages( summary = historySummary ?? turnPrefixSummary } } else { - summary = await summarizeTurnPrefix( - model, - summarizedTurnPrefix, - config.summarizationTimeoutMs, - turnPrefixOutputBudget, + summary = await firstSummary(models, (model) => + summarizeTurnPrefix( + model, + summarizedTurnPrefix, + config.summarizationTimeoutMs, + turnPrefixOutputBudget, + ), ) } } else { - summary = await summarizeMessages( - model, - toSummarize, - state.existingSummary, - config.summarizationTimeoutMs, - config.summarizerMaxOutputTokens, + summary = await firstSummary(models, (model) => + summarizeMessages( + model, + toSummarize, + state.existingSummary, + config.summarizationTimeoutMs, + config.summarizerMaxOutputTokens, + ), ) } @@ -333,6 +359,8 @@ export function createCompactionPrepareStep( minSummarizableTokens: config.minSummarizableTokens, maxSummarizationInput: config.maxSummarizationInput, summarizerMaxOutputTokens: config.summarizerMaxOutputTokens, + summarizationTimeoutMs: config.summarizationTimeoutMs, + summarizer: userConfig?.summarizer !== undefined, }) return async ({ @@ -402,7 +430,10 @@ export function createCompactionPrepareStep( }, ) - const compacted = await compactMessages(model, reduced, config, state) + const models = userConfig?.summarizer + ? [userConfig.summarizer, model] + : [model] + const compacted = await compactMessages(models, reduced, config, state) return { messages: compacted, experimental_context: state } } } diff --git a/trios/agent-server/apps/server/src/agent/compaction/utils.ts b/trios/agent-server/apps/server/src/agent/compaction/utils.ts index 9a4f977d08..c0af343561 100644 --- a/trios/agent-server/apps/server/src/agent/compaction/utils.ts +++ b/trios/agent-server/apps/server/src/agent/compaction/utils.ts @@ -61,6 +61,51 @@ function safeJsonStringify(value: unknown): string { } } +/** + * HOW LONG A SUMMARY MAY TAKE. Measured 2026-10-03 on the swarm's runners: of + * 245 compactions in fifty minutes 194 came back empty, 158 of them 59-64 s + * after they began - the fixed 60 s abort, not an answer. z-ai/glm-5.3 under + * load and the free glm-4.5-flash cannot summarize 7-30k tokens in a minute, + * and an aborted stream ends quietly, so the bee fell back to the sliding + * window and lost the start of its own work. TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS + * names a longer wait; anything unreadable or outside 10 s..10 min keeps 60 s. + */ +export function summarizationTimeoutMs( + env: Record = process.env, +): number { + const raw = env.TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS?.trim() + const value = raw && /^\d+$/.test(raw) ? Number(raw) : Number.NaN + return value >= 10_000 && value <= 600_000 + ? value + : AGENT_LIMITS.COMPACTION_SUMMARIZATION_TIMEOUT_MS +} + +/** + * WHICH MODEL WRITES THE SUMMARY. A summary is a short, plain task that a fast + * model does in seconds, while the bee's own model may be a slow reasoner. + * TRIOS_COMPACTION_SUMMARIZER_MODELS maps a provider to a model on the bee's + * own key and endpoint, e.g. + * `openai-compatible=nvidia/nemotron-3-super-120b-a12b`. Unset, unmapped, or + * the bee's own model: null, and the bee summarizes with its own model. + */ +export function summarizerModelFor( + provider: string, + beeModel: string, + env: Record = process.env, +): string | null { + for (const pair of (env.TRIOS_COMPACTION_SUMMARIZER_MODELS ?? '').split( + ',', + )) { + const at = pair.indexOf('=') + if (at < 1) continue + const model = pair.slice(at + 1).trim() + if (pair.slice(0, at).trim() !== provider) continue + if (!/^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$/.test(model)) return null + return model === beeModel ? null : model + } + return null +} + export function computeConfig(contextWindow: number): ComputedConfig { const reserveTokens = contextWindow <= AGENT_LIMITS.COMPACTION_SMALL_CONTEXT_WINDOW @@ -117,7 +162,7 @@ export function computeConfig(contextWindow: number): ComputedConfig { minSummarizableTokens, maxSummarizationInput, summarizerMaxOutputTokens, - summarizationTimeoutMs: AGENT_LIMITS.COMPACTION_SUMMARIZATION_TIMEOUT_MS, + summarizationTimeoutMs: summarizationTimeoutMs(), fixedOverhead, safetyMultiplier: AGENT_LIMITS.COMPACTION_SAFETY_MULTIPLIER, imageTokenEstimate: AGENT_LIMITS.COMPACTION_IMAGE_TOKEN_ESTIMATE, diff --git a/trios/agent-server/apps/server/src/api/services/queen-runner.ts b/trios/agent-server/apps/server/src/api/services/queen-runner.ts index e013edf404..7ec79310f4 100644 --- a/trios/agent-server/apps/server/src/api/services/queen-runner.ts +++ b/trios/agent-server/apps/server/src/api/services/queen-runner.ts @@ -39,7 +39,7 @@ import type { Pool } from 'pg' import { createQueenPool } from '../../lib/db/queen-pool' import { logger } from '../../lib/logger' import { bundleOfBranch } from '../routes/queen-export' -import { runClaimedBee } from './queen-dispatch' +import { runClaimedBee, salvageDispatch } from './queen-dispatch' import { queenLeaseDatabaseUrl } from './queen-lease' export interface BeeOrder { @@ -221,16 +221,112 @@ export async function runOneOrder( issue: order.issue, branch: order.branch, }) - const outcome = await runClaimedBee(pool, order) - if (!outcome.started) return order - await waitForEnding(pool, order, runner) - await storeBundle(pool, order.issue, runner).catch((error) => { - logger.warn('Runner could not store the bundle of a finished bee', { - issue: order.issue, - error: error instanceof Error ? error.message : String(error), + inFlight.set(order.issue, order) + try { + const outcome = await runClaimedBee(pool, order) + if (!outcome.started) return order + await waitForEnding(pool, order, runner) + await storeBundle(pool, order.issue, runner).catch((error) => { + logger.warn('Runner could not store the bundle of a finished bee', { + issue: order.issue, + error: error instanceof Error ? error.message : String(error), + }) }) - }) - return order + return order + } finally { + inFlight.delete(order.issue) + } +} + +/** The orders this process is carrying right now, by issue. */ +const inFlight = new Map() + +/** The runner this process started, so a shutdown can drain it. */ +let active: { pool: Pool; runner: string; stop: () => void } | null = null + +/** + * HOW LONG A STOPPING RUNNER WAITS FOR ITS BEES. Measured 2026-10-03: the + * server leaves about 2 s after SIGTERM, so every release killed every bee in + * flight - thirty at a time, each up to an hour of work - and their issues + * waited for the reaper. `TRIOS_BEE_RUNNER_DRAIN_SECONDS` names the wait + * (0..3600; unset or unreadable is 0, the old behaviour). The platform must + * allow it too: on Railway, RAILWAY_DEPLOYMENT_DRAINING_SECONDS at least as long. + */ +export function drainSeconds( + env: Record = process.env, +): number { + const raw = env.TRIOS_BEE_RUNNER_DRAIN_SECONDS?.trim() ?? '' + if (!/^\d+$/.test(raw)) return 0 + return Math.min(Number(raw), 3600) +} + +export interface DrainDeps { + seconds?: number + now?: () => number + sleep?: (ms: number) => Promise + salvage?: typeof salvageDispatch + store?: typeof storeBundle + orders?: Map +} + +/** + * Take no more orders and let the bees in flight end on their own, up to the + * drain deadline. A bee still running at the deadline is not thrown away: its + * uncommitted work is salvaged into its branch and the branch is stored as a + * bundle, so the Queen reviews what it did and the next attempt starts from it. + */ +export async function drainBeeRunner( + pool: Pool, + runner: string, + deps: DrainDeps = {}, +): Promise<{ waited: number; saved: number[] }> { + const orders = deps.orders ?? inFlight + const seconds = deps.seconds ?? drainSeconds() + const now = deps.now ?? Date.now + const sleep = + deps.sleep ?? ((ms: number) => new Promise((r) => setTimeout(r, ms))) + const started = now() + const deadline = started + seconds * 1000 + if (orders.size > 0) { + logger.info('Runner draining: no new orders, waiting for its bees', { + runner, + bees: orders.size, + seconds, + }) + } + while (orders.size > 0 && now() < deadline) { + await sleep(Math.min(2000, Math.max(0, deadline - now()))) + } + const saved: number[] = [] + for (const order of [...orders.values()]) { + try { + await (deps.salvage ?? salvageDispatch)(pool, order.issue, 'reaped', { + conversationId: order.conversationId, + }) + await (deps.store ?? storeBundle)(pool, order.issue, runner) + saved.push(order.issue) + } catch (error) { + logger.warn('Runner could not save the branch of a bee it stopped', { + issue: order.issue, + error: error instanceof Error ? error.message : String(error), + }) + } + } + if (saved.length > 0) { + logger.info('Runner saved the branches of bees it could not wait for', { + runner, + issues: saved, + }) + } + return { waited: Math.round((now() - started) / 1000), saved } +} + +/** For the server's shutdown: drain the runner this process started, if any. */ +export async function drainActiveRunner(): Promise { + if (!active) return + active.stop() + if (drainSeconds() === 0) return + await drainBeeRunner(active.pool, active.runner) } /** @@ -283,9 +379,14 @@ export function startBeeRunner(): void { }, every * 1000) take() - process.once('SIGTERM', () => { + const stop = (): void => { stopped = true clearInterval(timer) + } + active = { pool, runner, stop } + + process.once('SIGTERM', () => { + stop() // The bees in flight are NOT abandoned here. Their rows stay claimed and // unfinished, which is exactly what a Queen that died mid-bee always left, // and the boot reaper clears it the same way. diff --git a/trios/agent-server/apps/server/src/main.ts b/trios/agent-server/apps/server/src/main.ts index c1ac716b3f..a98aceba82 100644 --- a/trios/agent-server/apps/server/src/main.ts +++ b/trios/agent-server/apps/server/src/main.ts @@ -17,7 +17,7 @@ import { configureVmRuntime, peekOpenClawService, } from './api/services/openclaw/openclaw-service' -import { startBeeRunner } from './api/services/queen-runner' +import { drainActiveRunner, startBeeRunner } from './api/services/queen-runner' import { startQueenTick } from './api/services/queen-tick' import { CdpBackend } from './browser/backends/cdp' import { Browser } from './browser/browser' @@ -249,6 +249,9 @@ export class Application { async stop(reason?: string): Promise { logger.info('Shutting down server...', { reason }) + // Bees first, while the database and the HTTP server are still up: a + // runner given a drain window lets its bees end before anything closes. + if (reason === 'SIGTERM') await drainActiveRunner().catch(() => {}) stopSkillSync() // Only what was actually started. `getOpenClawService` constructs on // demand, so calling it here built the runtime it was meant to stop. diff --git a/trios/agent-server/apps/server/tests/agent/compaction-e2e.test.ts b/trios/agent-server/apps/server/tests/agent/compaction-e2e.test.ts index d126d5acf3..d55d84c6bd 100644 --- a/trios/agent-server/apps/server/tests/agent/compaction-e2e.test.ts +++ b/trios/agent-server/apps/server/tests/agent/compaction-e2e.test.ts @@ -734,6 +734,88 @@ describe('compaction E2E — summarization & fallbacks', () => { }) }) +describe('compaction E2E — a separate summarizer', () => { + it('a named summarizer writes the summary and the bee model is not asked', async () => { + const contextWindow = 10_000 + const config = computeConfig(contextWindow) + const triggerAt = Math.floor(contextWindow * config.triggerRatio) + let beeCalls = 0 + const bee = createMock(async () => { + beeCalls += 1 + return summaryResponse(200) + }) + const summarizer = createMock(async () => summaryResponse(200)) + const prepareStep = createCompactionPrepareStep({ + contextWindow, + summarizer, + }) + + const result = await prepareStep({ + messages: buildTextHeavyMessages(8, 2000), + steps: [{ usage: { inputTokens: triggerAt + 1000 } }] as StepsStub, + model: bee, + experimental_context: null, + }) + + expect( + (result.experimental_context as CompactionState).compactionCount, + ).toBe(1) + expect(summarizer.doStreamCalls.length).toBeGreaterThan(0) + expect(beeCalls).toBe(0) + }) + + it('an empty summarizer answer leaves the summary to the bee model', async () => { + const contextWindow = 10_000 + const config = computeConfig(contextWindow) + const triggerAt = Math.floor(contextWindow * config.triggerRatio) + const bee = createMock(async () => summaryResponse(200)) + const summarizer = createMock(async () => textResponse('', 200)) + const prepareStep = createCompactionPrepareStep({ + contextWindow, + summarizer, + }) + + const result = await prepareStep({ + messages: buildTextHeavyMessages(8, 2000), + steps: [{ usage: { inputTokens: triggerAt + 1000 } }] as StepsStub, + model: bee, + experimental_context: null, + }) + + const state = result.experimental_context as CompactionState + expect(state.compactionCount).toBe(1) + expect(state.existingSummary).toContain('## Goal') + expect(bee.doStreamCalls.length).toBeGreaterThan(0) + }) + + it('a failing summarizer and a failing bee model still fall back to the window', async () => { + const contextWindow = 10_000 + const config = computeConfig(contextWindow) + const triggerAt = Math.floor(contextWindow * config.triggerRatio) + const failing = () => + createMock(async () => { + throw new Error('busy') + }) + const prepareStep = createCompactionPrepareStep({ + contextWindow, + summarizer: failing(), + }) + const messages = buildTextHeavyMessages(8, 2000) + + const result = await prepareStep({ + messages, + steps: [{ usage: { inputTokens: triggerAt + 1000 } }] as StepsStub, + model: failing(), + experimental_context: null, + }) + + expect( + (result.experimental_context as CompactionState).compactionCount, + ).toBe(0) + expect(result.messages.length).toBeLessThanOrEqual(messages.length) + }) +}) + // --------------------------------------------------------------------------- // E2E: Iterative compaction // --------------------------------------------------------------------------- diff --git a/trios/agent-server/apps/server/tests/agent/compaction-summarizer-env.test.ts b/trios/agent-server/apps/server/tests/agent/compaction-summarizer-env.test.ts new file mode 100644 index 0000000000..7ce44cb1a0 --- /dev/null +++ b/trios/agent-server/apps/server/tests/agent/compaction-summarizer-env.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it } from 'bun:test' +import { AGENT_LIMITS } from '@browseros/shared/constants/limits' +import { + summarizationTimeoutMs, + summarizerModelFor, +} from '../../src/agent/compaction/utils' + +const NEMOTRON = 'nvidia/nemotron-3-super-120b-a12b' + +describe('summarizationTimeoutMs', () => { + it('keeps the 60 s default when unset or unreadable', () => { + expect(summarizationTimeoutMs({})).toBe( + AGENT_LIMITS.COMPACTION_SUMMARIZATION_TIMEOUT_MS, + ) + expect( + summarizationTimeoutMs({ + TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS: '3m', + }), + ).toBe(60_000) + expect( + summarizationTimeoutMs({ + TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS: '-1', + }), + ).toBe(60_000) + }) + + it('takes a named wait inside 10 s..10 min and refuses one outside it', () => { + expect( + summarizationTimeoutMs({ + TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS: '180000', + }), + ).toBe(180_000) + expect( + summarizationTimeoutMs({ + TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS: ' 10000 ', + }), + ).toBe(10_000) + expect( + summarizationTimeoutMs({ + TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS: '600000', + }), + ).toBe(600_000) + expect( + summarizationTimeoutMs({ + TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS: '9999', + }), + ).toBe(60_000) + expect( + summarizationTimeoutMs({ + TRIOS_COMPACTION_SUMMARIZATION_TIMEOUT_MS: '600001', + }), + ).toBe(60_000) + }) +}) + +describe('summarizerModelFor', () => { + const env = { + TRIOS_COMPACTION_SUMMARIZER_MODELS: ` openai-compatible = ${NEMOTRON} , zai=glm-4.5-flash`, + } + + it('maps the bee provider to its named summarizer', () => { + expect(summarizerModelFor('openai-compatible', 'z-ai/glm-5.3', env)).toBe( + NEMOTRON, + ) + }) + + it('names nothing for an unmapped provider, an unset variable or the bee model itself', () => { + expect(summarizerModelFor('anthropic', 'claude', env)).toBeNull() + expect( + summarizerModelFor('openai-compatible', 'z-ai/glm-5.3', {}), + ).toBeNull() + expect(summarizerModelFor('zai', 'glm-4.5-flash', env)).toBeNull() + }) + + it('refuses a malformed model name rather than sending it to a provider', () => { + expect( + summarizerModelFor('openai-compatible', 'x', { + TRIOS_COMPACTION_SUMMARIZER_MODELS: 'openai-compatible=bad model', + }), + ).toBeNull() + expect( + summarizerModelFor('openai-compatible', 'x', { + TRIOS_COMPACTION_SUMMARIZER_MODELS: '=m,openai-compatible', + }), + ).toBeNull() + }) +}) diff --git a/trios/agent-server/apps/server/tests/api/queen-runner.test.ts b/trios/agent-server/apps/server/tests/api/queen-runner.test.ts index 99dbe904ec..aedf198499 100644 --- a/trios/agent-server/apps/server/tests/api/queen-runner.test.ts +++ b/trios/agent-server/apps/server/tests/api/queen-runner.test.ts @@ -20,6 +20,9 @@ import { workerProviderForKeyIndex, } from '../../src/api/services/queen-dispatch' import { + type BeeOrder, + drainBeeRunner, + drainSeconds, runnerName, runnerSlots, waitForEnding, @@ -343,3 +346,128 @@ describe('the Queen brings a runner branch into her checkout before judging it', expect(sh(queen, 'rev-parse', 'queen-7')).toBe(head) }) }) + +describe('a stopping runner drains its bees', () => { + const order = (issue: number): BeeOrder => ({ + issue, + branch: `queen-${issue}`, + brief: '', + ownedPaths: [], + conversationId: `c-${issue}`, + keyIndex: 0, + }) + const pool = {} as Pool + + it('reads the drain window from the environment, bounded', () => { + expect(drainSeconds({})).toBe(0) + expect(drainSeconds({ TRIOS_BEE_RUNNER_DRAIN_SECONDS: 'soon' })).toBe(0) + expect(drainSeconds({ TRIOS_BEE_RUNNER_DRAIN_SECONDS: '-5' })).toBe(0) + expect(drainSeconds({ TRIOS_BEE_RUNNER_DRAIN_SECONDS: ' 900 ' })).toBe(900) + expect(drainSeconds({ TRIOS_BEE_RUNNER_DRAIN_SECONDS: '99999' })).toBe(3600) + }) + + it('waits for the bees in flight and saves nothing when they all end in time', async () => { + const orders = new Map([ + [1, order(1)], + [2, order(2)], + ]) + let clock = 0 + const saved: number[] = [] + const result = await drainBeeRunner(pool, 'r', { + orders, + seconds: 600, + now: () => clock, + sleep: async (ms) => { + clock += ms + // One bee ends per wait. + orders.delete([...orders.keys()][0]) + }, + salvage: async () => { + throw new Error('nothing should be salvaged') + }, + store: async (_pool, issue) => { + saved.push(issue) + }, + }) + expect(result.saved).toEqual([]) + expect(saved).toEqual([]) + expect(orders.size).toBe(0) + }) + + it('at the deadline salvages and stores every bee still running, and only those', async () => { + const orders = new Map([ + [7, order(7)], + [8, order(8)], + ]) + let clock = 0 + const salvaged: Array<[number, string | null | undefined]> = [] + const stored: number[] = [] + const result = await drainBeeRunner(pool, 'r', { + orders, + seconds: 10, + now: () => clock, + sleep: async (ms) => { + clock += ms + orders.delete(8) + }, + salvage: (async ( + _pool: Pool, + issue: number, + _reason: string, + deps: { conversationId?: string | null }, + ) => { + salvaged.push([issue, deps.conversationId]) + return { committed: true, files: [], left: [], sha: null, detail: '' } + }) as never, + store: async (_pool, issue) => { + stored.push(issue) + }, + }) + expect(result.saved).toEqual([7]) + expect(salvaged).toEqual([[7, 'c-7']]) + expect(stored).toEqual([7]) + expect(clock).toBeLessThanOrEqual(10_000) + }) + + it('with no window it saves the running bees at once instead of waiting', async () => { + const orders = new Map([[3, order(3)]]) + let slept = 0 + const result = await drainBeeRunner(pool, 'r', { + orders, + seconds: 0, + now: () => 0, + sleep: async () => { + slept += 1 + }, + salvage: (async () => ({ + committed: false, + files: [], + left: [], + sha: null, + detail: '', + })) as never, + store: async () => {}, + }) + expect(slept).toBe(0) + expect(result.saved).toEqual([3]) + }) + + it('a bee whose branch cannot be saved does not stop the others from being saved', async () => { + const orders = new Map([ + [4, order(4)], + [5, order(5)], + ]) + const result = await drainBeeRunner(pool, 'r', { + orders, + seconds: 0, + now: () => 0, + sleep: async () => {}, + salvage: (async (_p: Pool, issue: number) => { + if (issue === 4) throw new Error('git is gone') + return { committed: false, files: [], left: [], sha: null, detail: '' } + }) as never, + store: async () => {}, + }) + expect(result.saved).toEqual([5]) + }) +}) diff --git a/trios/agent-server/docker-entrypoint.sh b/trios/agent-server/docker-entrypoint.sh index f3ef2d40fc..abd82d1815 100755 --- a/trios/agent-server/docker-entrypoint.sh +++ b/trios/agent-server/docker-entrypoint.sh @@ -156,8 +156,15 @@ run_supervised() { done ) & set +e + # A trapped TERM ends `wait` at once (POSIX), while the server is still + # draining its bees; leaving here then would end the container and every bee + # with it. So wait again for as long as the server is alive. wait "$server" code=$? + while kill -0 "$server" 2>/dev/null; do + wait "$server" + code=$? + done # A server that was ended for not answering exits by signal, which `wait` # reports as 128+N. Anything but a clean 0 must read as a failure to the # platform, or ON_FAILURE would not restart it.