The founder probes through provider cooldowns; a served request lifts the park
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.
This commit is contained in:
1 parent
9246e7e312
commit
75c73ce6f0
3 files changed
+166
-4
No files matched your search
@@ -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)
|
||||
|
||||
@@ -66,7 +66,7 @@ export class TaskExecutionService {
|
||||
/** Providers currently parked on their monthly quota, for transition logging. */
|
||||
private quotaBlockedProviders = new Map<string, string>();
|
||||
/** Providers parked after a transient failure (usage/rate limit, overload): id -> park state. */
|
||||
private providerCooldowns = new Map<string, { until: number; label: string }>();
|
||||
private providerCooldowns = new Map<string, { until: number; label: string; interactiveProbedAt?: number }>();
|
||||
/** Tasks already logged as budget-spent, so the park is announced once, not every 5s tick. */
|
||||
private spentTasksLogged = new Set<string>();
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<AgentTurnResult> => {
|
||||
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<AgentTurnResult> => {
|
||||
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();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user