From 3f7e85778aab163b0135d22fdcc94873f352bc72 Mon Sep 17 00:00:00 2001 From: Cameron Pfiffer Date: Wed, 15 Jul 2026 09:39:46 -0700 Subject: [PATCH] Isolate specialist consumer scheduling. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Run declaration-source queues independently under bounded concurrency so a busy observer cannot starve repair or other composed specialists while preserving order within each stream. 👾 Generated with [Letta Code](https://letta.com) Co-Authored-By: Letta Code --- src/agents/runtime.ts | 109 ++++++++++++++++++++++++++-------------- test/acceptance.test.ts | 65 ++++++++++++++++++++++++ 2 files changed, 135 insertions(+), 39 deletions(-) diff --git a/src/agents/runtime.ts b/src/agents/runtime.ts index db3c712..b9d8837 100644 --- a/src/agents/runtime.ts +++ b/src/agents/runtime.ts @@ -96,11 +96,13 @@ export class ThoughtAgentRuntime { await this.registerDeclarations(declarations); const unsubscribers: Array<() => void> = []; const subscriptions = new Set(); - let chain = Promise.resolve(); + const chains = new Map>(); + const withConsumerSlot = concurrentOperationLimiter(4); let stopped = false; const consumerErrors: unknown[] = []; - const enqueue = (operation: () => Promise): void => { - chain = chain.then(async () => { + const enqueue = (key: string, operation: () => Promise): void => { + const previous = chains.get(key) ?? Promise.resolve(); + const current = previous.then(() => withConsumerSlot(async () => { let lastError: unknown; for (let attempt = 1; attempt <= 3; attempt += 1) { try { @@ -112,43 +114,52 @@ export class ThoughtAgentRuntime { } } throw lastError; - }).catch((error: unknown) => { + })).catch((error: unknown) => { consumerErrors.push(error); process.stderr.write(`ThoughtStream consumer cycle failed: ${error instanceof Error ? error.message : String(error)}\n`); }); + chains.set(key, current); + void current.then(() => { + if (chains.get(key) === current) chains.delete(key); + }); }; const enabled = declarations.filter((candidate) => candidate.enabled); const install = async (declaration: ThoughtAgentDeclaration, source: string): Promise => { const key = `${declaration.id}@${declaration.version}:${source}`; if (stopped || subscriptions.has(key)) return; - const progress = await this.store.getConsumerProgress(progressId(declaration, source)); - const query = this.consumerQuery(declaration, source, progress?.lastSequence ?? 0); - const consumeAvailable = async (): Promise => { - for (;;) { - const currentProgress = await this.store.getConsumerProgress(progressId(declaration, source)); - const events = await this.store.queryConsumerEvents(this.consumerQuery( - declaration, - source, - currentProgress?.lastSequence ?? 0, - )); - let retryableFailure = false; - for (const event of events) { - const current = await this.store.getConsumerProgress(progressId(declaration, source)); - if ((current?.lastSequence ?? 0) >= event.sourceSequence) continue; - const result = await this.executeEvent(event, declaration); - if (result.retryable) { - retryableFailure = true; - break; + subscriptions.add(key); + try { + const progress = await this.store.getConsumerProgress(progressId(declaration, source)); + const query = this.consumerQuery(declaration, source, progress?.lastSequence ?? 0); + const consumeAvailable = async (): Promise => { + for (;;) { + const currentProgress = await this.store.getConsumerProgress(progressId(declaration, source)); + const events = await this.store.queryConsumerEvents(this.consumerQuery( + declaration, + source, + currentProgress?.lastSequence ?? 0, + )); + let retryableFailure = false; + for (const event of events) { + const current = await this.store.getConsumerProgress(progressId(declaration, source)); + if ((current?.lastSequence ?? 0) >= event.sourceSequence) continue; + const result = await this.executeEvent(event, declaration); + if (result.retryable) { + retryableFailure = true; + break; + } } + if (retryableFailure || events.length < 1_000) break; } - if (retryableFailure || events.length < 1_000) break; - } - }; - unsubscribers.push(this.store.subscribeConsumerEvents(query, () => { - if (!stopped) enqueue(consumeAvailable); - })); - subscriptions.add(key); - enqueue(consumeAvailable); + }; + unsubscribers.push(this.store.subscribeConsumerEvents(query, () => { + if (!stopped) enqueue(key, consumeAvailable); + })); + enqueue(key, consumeAvailable); + } catch (error) { + subscriptions.delete(key); + throw error; + } }; for (const declaration of enabled) { for (const source of await this.resolveSources(declaration)) await install(declaration, source); @@ -156,7 +167,7 @@ export class ThoughtAgentRuntime { if (enabled.some((declaration) => declaration.sourcePatterns.some((pattern) => pattern.includes("*")))) { unsubscribers.push(this.store.subscribeSources((sources) => { if (stopped) return; - enqueue(async () => { + enqueue("source-discovery", async () => { for (const declaration of enabled) { for (const source of sources) { if (declaration.sourcePatterns.some((pattern) => matchesPatternValue(pattern, source.id))) { @@ -169,19 +180,17 @@ export class ThoughtAgentRuntime { } return { drain: async () => { - for (;;) { - const pending = chain; - await pending; - if (pending === chain) { - if (consumerErrors.length > 0) throw new AggregateError(consumerErrors.splice(0), "ThoughtStream consumer cycles failed"); - return; - } + while (chains.size > 0) { + await Promise.all([...chains.values()]); } + if (consumerErrors.length > 0) throw new AggregateError(consumerErrors.splice(0), "ThoughtStream consumer cycles failed"); }, stop: async () => { stopped = true; for (const unsubscribe of unsubscribers) unsubscribe(); - await chain; + while (chains.size > 0) { + await Promise.all([...chains.values()]); + } if (consumerErrors.length > 0) throw new AggregateError(consumerErrors.splice(0), "ThoughtStream consumer cycles failed"); }, }; @@ -589,6 +598,28 @@ export class ThoughtAgentRuntime { } } +function concurrentOperationLimiter(maximum: number): (operation: () => Promise) => Promise { + let active = 0; + const waiting: Array<() => void> = []; + const acquire = async (): Promise => { + if (active < maximum) { + active += 1; + return; + } + await new Promise((resolve) => waiting.push(resolve)); + }; + return async (operation: () => Promise): Promise => { + await acquire(); + try { + return await operation(); + } finally { + const next = waiting.shift(); + if (next) next(); + else active -= 1; + } + }; +} + function executionKey(event: ThoughtEvent, declaration: ThoughtAgentDeclaration): string { return stableKey("execution", declaration.id, String(declaration.version), event.id); } diff --git a/test/acceptance.test.ts b/test/acceptance.test.ts index 6046f2a..f32ff57 100644 --- a/test/acceptance.test.ts +++ b/test/acceptance.test.ts @@ -175,6 +175,71 @@ describe("Jazz-native producer and consumer topology", () => { expect((await store.listSources()).find((source) => source.id === "agent:rss-fixture-consumer")?.lastSequence).toBe(6); }); + test("runs independent consumer queues concurrently without losing per-source ordering", async () => { + const project = await temporaryProject(); + roots.push(project); + const store = testStore(project); + stores.push(store); + const slow = { + ...subscriptionDeclaration(), + id: "slow-consumer", + name: "Slow consumer", + sourcePatterns: ["rss:slow"], + }; + const fast = { + ...subscriptionDeclaration(), + id: "fast-consumer", + name: "Fast consumer", + sourcePatterns: ["rss:fast"], + }; + let releaseSlow!: () => void; + let markFastStarted!: () => void; + const slowCanFinish = new Promise((resolve) => { releaseSlow = resolve; }); + const fastStarted = new Promise((resolve) => { markFastStarted = resolve; }); + const executionOrder: string[] = []; + const runner: AgentRunner = { + mode: "deterministic", + run: async ({ declaration, event }) => { + executionOrder.push(`${declaration.id}:${event.externalId}:start`); + if (declaration.id === slow.id) await slowCanFinish; + else { + markFastStarted(); + releaseSlow(); + } + executionOrder.push(`${declaration.id}:${event.externalId}:finish`); + return { + summary: `Observed ${event.externalId}`, + tags: ["rss"], + importance: "normal", + confidence: 1, + }; + }, + }; + await store.appendProducerBatch([ + { ...candidate("one"), source: "rss:slow", externalId: "slow-one", idempotencyKey: "slow-one" }, + { ...candidate("two"), source: "rss:slow", externalId: "slow-two", idempotencyKey: "slow-two" }, + ]); + await store.appendProducerBatch([ + { ...candidate("one"), source: "rss:fast", externalId: "fast-one", idempotencyKey: "fast-one" }, + ]); + + const consumers = await new ThoughtAgentRuntime(store, [runner]).startConsumers([slow, fast]); + await Promise.race([ + fastStarted, + new Promise((_resolve, reject) => setTimeout(() => reject(new Error("Fast consumer was starved by the slow queue")), 1_000)), + ]); + await waitFor(async () => (await store.listConsumerProgress()).filter((progress) => ( + [slow.id, fast.id].includes(progress.consumerId) + )).length === 2); + await consumers.stop(); + + expect(executionOrder.indexOf("fast-consumer:fast-one:start")) + .toBeLessThan(executionOrder.indexOf("slow-consumer:slow-one:finish")); + expect(executionOrder.indexOf("slow-consumer:slow-one:finish")) + .toBeLessThan(executionOrder.indexOf("slow-consumer:slow-two:start")); + expect((await store.listRuns()).filter((run) => run.status === "completed")).toHaveLength(3); + }); + test("keeps a live consumer ordered and recoverable after one subscription cycle throws", async () => { const project = await temporaryProject(); roots.push(project); -- 2.51.2