diff --git a/apps/server/src/orchestration-v2/ShellStream.test.ts b/apps/server/src/orchestration-v2/ShellStream.test.ts index fbc610235dd01..ad39c266b434c 100644 --- a/apps/server/src/orchestration-v2/ShellStream.test.ts +++ b/apps/server/src/orchestration-v2/ShellStream.test.ts @@ -282,6 +282,35 @@ describe("omitOlderSettledShells", () => { expect(result.threads.map((thread) => thread.id).slice(25)).toEqual(["tie"]); }); + it("finds the same cut whatever order the settled threads arrive in", () => { + const ascending = Array.from({ length: 200 }, (_, index) => + row(`${index % 2 === 0 ? "a" : "b"}-${index}`, index), + ); + let seed = 7; + const shuffled = [...ascending]; + for (let index = shuffled.length - 1; index > 0; index--) { + seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0; + const other = seed % (index + 1); + [shuffled[index], shuffled[other]] = [shuffled[other]!, shuffled[index]!]; + } + const newestIds = new Set(ascending.slice(0, 25).map((thread) => thread.id)); + for (const threads of [ascending, [...ascending].reverse(), shuffled]) { + const result = omit(threads); + expect(result.threads.map((thread) => thread.id)).toEqual( + threads.filter((thread) => newestIds.has(thread.id)).map((thread) => thread.id), + ); + expect(result.olderSettled.before).toBe( + DateTime.formatIso(DateTime.makeUnsafe(100 * DAY - 24 * DAY)), + ); + expect(result.olderSettled.counts).toEqual( + expect.arrayContaining([ + { projectId: "project-a", count: 87 }, + { projectId: "project-b", count: 88 }, + ]), + ); + } + }); + it("omits nothing with fewer than 25 settled threads, however old", () => { const old = Array.from({ length: 5 }, (_, index) => row(`old-${index}`, 300 + index)); const result = omit(old); diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 5ee1005b16830..2a7d4151e83f2 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -202,6 +202,7 @@ function shellTrimFacts(rows: ReadonlyArray): ShellTrimFacts { const subagentParentIds = new Set(); const subagentThreadIds = new Set(); const remoteTaskParentIds = new Set(); + // A bounded min-heap keeps the 25th newest timestamp at its root. const settledTimes: Array = []; for (const row of rows) { if (row.hasRemoteTasks === true) remoteTaskParentIds.add(row.id); @@ -213,12 +214,33 @@ function shellTrimFacts(rows: ReadonlyArray): ShellTrimFacts { row.settledOverride === "settled" && row.settledAt !== null ) { - settledTimes.push(DateTime.toEpochMillis(row.settledAt)); + const settledAtMs = DateTime.toEpochMillis(row.settledAt); + let position = settledTimes.length; + if (position < NEWEST_SETTLED_KEPT) { + settledTimes.push(settledAtMs); + while (position > 0) { + const parent = (position - 1) >>> 1; + if (settledTimes[parent]! <= settledAtMs) break; + settledTimes[position] = settledTimes[parent]!; + position = parent; + } + } else { + if (settledAtMs <= settledTimes[0]!) continue; + position = 0; + while (position * 2 + 1 < settledTimes.length) { + let child = position * 2 + 1; + if (child + 1 < settledTimes.length && settledTimes[child + 1]! < settledTimes[child]!) + child++; + if (settledAtMs <= settledTimes[child]!) break; + settledTimes[position] = settledTimes[child]!; + position = child; + } + } + settledTimes[position] = settledAtMs; } } - settledTimes.sort((left, right) => right - left); // With fewer than 25, the oldest one is the cut, so nothing is older. - const olderSettledBeforeMs = settledTimes[NEWEST_SETTLED_KEPT - 1] ?? settledTimes.at(-1); + const olderSettledBeforeMs = settledTimes[0]; const counts = new Map(); if (olderSettledBeforeMs !== undefined) { for (const row of rows) {