From 75c73ce6f0e19dbbb507223d429d3c14b08edc5d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Tue, 14 Jul 2026 21:19:14 -0400 Subject: [PATCH] The founder probes through provider cooldowns; a served request lifts the park MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A single transient failure from a background run used to silence the whole company — founder included — for the full ten-minute cooldown. The interactive lane now claims one probe per park window: the founder's message tries the provider immediately, a success lifts the cooldown for everyone (a served request is proof of health), and a failed probe re-arms the park with the probe spent, so a real outage is never hammered. Autonomous work still waits cooldowns out unconditionally. --- src/main/app/task-queue-worker.ts | 6 +- src/main/services/task-execution-service.ts | 36 +++++- tests/integration/interactive-lane.test.ts | 128 +++++++++++++++++++- 3 files changed, 166 insertions(+), 4 deletions(-) diff --git a/src/main/app/task-queue-worker.ts b/src/main/app/task-queue-worker.ts index 7053d4f5..03181169 100644 --- a/src/main/app/task-queue-worker.ts +++ b/src/main/app/task-queue-worker.ts @@ -96,10 +96,14 @@ export function startTaskQueueWorker(input: TaskQueueWorkerInput): () => void { const dispatch = (slot: "interactive" | "autonomous"): void => { if (inFlight[slot]) return; + // The interactive lane may probe a cooled provider once per park window + // (a human is asking; a blip should not cost them ten silent minutes); + // autonomous work always waits the cooldown out. const claimed = slot === "interactive" - ? claimNextQueuedRun(input.connection, taskExecution.cooledProviderIds(), "interactive") + ? claimNextQueuedRun(input.connection, taskExecution.interactiveCooledProviderIds(), "interactive") : claimNextQueuedRun(input.connection, taskExecution.cooledProviderIds(), "any"); if (!claimed) return; + if (slot === "interactive" && claimed.provider_id) taskExecution.noteInteractiveProbe(claimed.provider_id); inFlight[slot] = true; notifyDataChanged(); void taskExecution.executeRun(claimed.org_id, claimed.id) diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 023a1328..9f87d49f 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -66,7 +66,7 @@ export class TaskExecutionService { /** Providers currently parked on their monthly quota, for transition logging. */ private quotaBlockedProviders = new Map(); /** Providers parked after a transient failure (usage/rate limit, overload): id -> park state. */ - private providerCooldowns = new Map(); + private providerCooldowns = new Map(); /** Tasks already logged as budget-spent, so the park is announced once, not every 5s tick. */ private spentTasksLogged = new Set(); @@ -184,6 +184,7 @@ export class TaskExecutionService { ); providerRequestId = result.lastRequestId; this.completeRun(orgId, runId, result.lastRequestId); + this.clearProviderCooldown(provider); this.audit.record({ orgId, entityType: "task_run", @@ -385,7 +386,13 @@ export class TaskExecutionService { private parkProvider(provider: TaskExecutionProviderRow, message: string): void { const active = this.providerCooldowns.get(provider.id); const wasParked = active !== undefined && active.until > this.now(); - this.providerCooldowns.set(provider.id, { until: this.now() + PROVIDER_TRANSIENT_FAILURE_COOLDOWN_MS, label: provider.label }); + this.providerCooldowns.set(provider.id, { + until: this.now() + PROVIDER_TRANSIENT_FAILURE_COOLDOWN_MS, + label: provider.label, + // A failed probe must not unlock endless probes: re-arming the park + // keeps the spent-probe mark until the window expires naturally. + interactiveProbedAt: wasParked ? active?.interactiveProbedAt : undefined, + }); if (!wasParked) { const minutes = Math.round(PROVIDER_TRANSIENT_FAILURE_COOLDOWN_MS / 60_000); console.warn(`[task-queue] provider "${provider.label}" reported a transient failure ("${message}"); its tasks are parked for ${minutes} minutes.`); @@ -409,6 +416,31 @@ export class TaskExecutionService { return active; } + /** + * Cooled providers that even the interactive lane must skip: their one + * founder-probe for this park window is already spent. A cooled provider + * not yet probed stays interactively claimable — one human message is + * allowed through to learn whether the outage was a blip, instead of the + * founder sitting out the whole cooldown for a single failed background + * request. + */ + interactiveCooledProviderIds(): string[] { + return this.cooledProviderIds().filter((id) => this.providerCooldowns.get(id)?.interactiveProbedAt !== undefined); + } + + /** Marks the cooled provider's founder-probe as spent (no-op when healthy). */ + noteInteractiveProbe(providerId: string): void { + const park = this.providerCooldowns.get(providerId); + if (park) park.interactiveProbedAt = this.now(); + } + + /** A served request is proof of health: the park ends for everyone at once. */ + private clearProviderCooldown(provider: TaskExecutionProviderRow): void { + if (this.providerCooldowns.delete(provider.id)) { + console.info(`[task-queue] provider "${provider.label}" served a request; cooldown lifted early.`); + } + } + /** * Logs monthly-quota transitions (blocked <-> unblocked) exactly once per * state change so a parked queue is diagnosable from the main-process log diff --git a/tests/integration/interactive-lane.test.ts b/tests/integration/interactive-lane.test.ts index 93ad6d82..d5b01246 100644 --- a/tests/integration/interactive-lane.test.ts +++ b/tests/integration/interactive-lane.test.ts @@ -10,7 +10,7 @@ import { ProviderService } from "../../src/main/services/provider-service"; import { providerAdapters } from "../../src/main/services/provider-adapters"; import { TaskExecutionService } from "../../src/main/services/task-execution-service"; import { dispatchAgentTool } from "../../src/main/services/agent-tools"; -import { startTaskQueueWorker } from "../../src/main/app/task-queue-worker"; +import { claimNextQueuedRun, startTaskQueueWorker } from "../../src/main/app/task-queue-worker"; import type { AgentTurnResult, ProviderToolTurnInput, ToolCall } from "../../src/main/services/provider-tool-turn"; const testVault = { @@ -126,4 +126,130 @@ describe("interactive execution lane", () => { connection.close(); } }); + + test("a founder message probes through a provider cooldown and a served request lifts the park", async () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-probe-${randomUUID()}.db`)); + const organization = new OrganizationService(connection.db); + let orgId = ""; + let ceoId = ""; + let providerId = ""; + connection.transaction(() => { + orgId = organization.createOrganization({ organizationName: "OrbitOps", employeeName: "Yang", mission: "m", goal: "g" }); + providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'").get(orgId) as { id: string }).id; + connection.db.prepare("update providers set enabled = 1 where id = ? and org_id = ?").run(providerId, orgId); + ceoId = organization.createAgent({ + orgId, name: "CEO", role: "CEO", department: "Executive", + modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 25000, managerAgentId: null, roleLevel: "c_suite", + }); + organization.createAgent({ + orgId, name: "Dana", role: "Engineer", department: "Engineering", + modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: ceoId, + }); + }); + const employeeId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id; + const chat = new ChatService(connection.db); + + let ceoReplied = false; + const script = async (input: ProviderToolTurnInput): Promise => { + const prompt = input.messages.find((message) => message.role === "user")?.text ?? ""; + if (prompt.includes("Grind through the migration backlog")) { + throw new Error("The engine is currently overloaded, please try again later"); + } + const threadId = /thread_id="([^"]+)"/.exec(prompt)?.[1]; + if (prompt.includes("Are we blocked") && threadId && !ceoReplied) { + ceoReplied = true; + return turn("Answering.", [call("send_message", { to: "Yang", body: "Not blocked — one hiccup, already moving.", thread_id: threadId })]); + } + return turn("done"); + }; + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, script); + providers.saveCredential({ orgId, providerId, secret: "sk-test-key", model: "gpt-5.4" }); + const taskExecution = new TaskExecutionService(connection.db, providers); + + // Autonomous work that will trip the transient failure and park the provider. + dispatchAgentTool({ db: connection.db, organization, chat }, { + orgId, agent: { id: ceoId, modelProviderId: null, modelId: null }, + taskProjectId: null, taskGoalId: null, workspace: null, + }, call("send_message", { to: "Dana", body: "Grind through the migration backlog." })); + + const stopWorker = startTaskQueueWorker({ + connection, + mainWindow: { isDestroyed: () => true } as unknown as BrowserWindow, + intervalMs: 25, + taskExecution, + }); + try { + await vi.waitFor(() => { + expect(taskExecution.cooledProviderIds()).toEqual([providerId]); + }, { timeout: 3000, interval: 25 }); + + // The founder speaks INTO the cooldown; the interactive lane probes through. + const spaceId = chat.ensureDirectSpace(orgId, { type: "employee", id: employeeId }, { type: "agent", id: ceoId }); + chat.sendMessage({ + orgId, spaceId, threadId: null, senderType: "employee", senderId: employeeId, + bodyMd: "Are we blocked on anything?", attachmentFileIds: [], + }); + await vi.waitFor(() => { + const reply = connection.db.prepare( + "select id from chat_messages where org_id = ? and sender_type = 'agent' and sender_id = ? and space_id = ?", + ).get(orgId, ceoId, spaceId); + expect(reply).toBeTruthy(); + }, { timeout: 3000, interval: 25 }); + + // The served probe is proof of health: the park lifted for everyone. + expect(taskExecution.cooledProviderIds()).toEqual([]); + } finally { + stopWorker(); + connection.close(); + } + }); + + test("a spent probe blocks further interactive claims until the window expires", async () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-probe-spent-${randomUUID()}.db`)); + const organization = new OrganizationService(connection.db); + let orgId = ""; + let ceoId = ""; + let providerId = ""; + connection.transaction(() => { + orgId = organization.createOrganization({ organizationName: "OrbitOps", employeeName: "Yang", mission: "m", goal: "g" }); + providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'").get(orgId) as { id: string }).id; + connection.db.prepare("update providers set enabled = 1 where id = ? and org_id = ?").run(providerId, orgId); + ceoId = organization.createAgent({ + orgId, name: "CEO", role: "CEO", department: "Executive", + modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 25000, managerAgentId: null, roleLevel: "c_suite", + }); + organization.createAgent({ + orgId, name: "Dana", role: "Engineer", department: "Engineering", + modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: ceoId, + }); + }); + const chat = new ChatService(connection.db); + const alwaysOverloaded = async (): Promise => { + throw new Error("The engine is currently overloaded, please try again later"); + }; + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, alwaysOverloaded); + providers.saveCredential({ orgId, providerId, secret: "sk-test-key", model: "gpt-5.4" }); + const taskExecution = new TaskExecutionService(connection.db, providers); + const wakeDana = () => dispatchAgentTool({ db: connection.db, organization, chat }, { + orgId, agent: { id: ceoId, modelProviderId: null, modelId: null }, + taskProjectId: null, taskGoalId: null, workspace: null, + }, call("send_message", { to: "Dana", body: "Grind through the migration backlog." })); + + wakeDana(); + const first = claimNextQueuedRun(connection)!; + await taskExecution.executeRun(first.org_id, first.id).catch(() => {}); + expect(taskExecution.cooledProviderIds()).toEqual([providerId]); + // The park window opens with one probe available to the interactive lane. + expect(taskExecution.interactiveCooledProviderIds()).toEqual([]); + + taskExecution.noteInteractiveProbe(providerId); + expect(taskExecution.interactiveCooledProviderIds()).toEqual([providerId]); + + // A failed probe re-arms the park WITHOUT restoring the probe: no hammering. + wakeDana(); + const second = claimNextQueuedRun(connection)!; + await taskExecution.executeRun(second.org_id, second.id).catch(() => {}); + expect(taskExecution.interactiveCooledProviderIds()).toEqual([providerId]); + connection.close(); + }); });