fix(server): stop short start gates from undercutting provider startup budgets (#3621)
* fix(server): stop short start gates from undercutting provider startup budgets Agent creation still failed on contended cold starts after #3578, because two shorter gates on the same path fire before the event-stream readiness budget can apply. A traced creation reached session.create twelve seconds in, past the 10s session.create gate, and the agent was then killed by a bare "timeout" at fifteen seconds from AGENT_RUN_START_TIMEOUT_MS while the SSE connect was still running. A run-start bound of 15s wraps a path that is deliberately allowed 30s, so the outer gate always wins and the inner budget is unreachable. Give session.create the shared OPENCODE_SERVER_STARTUP_TIMEOUT_MS budget, since creating the first session for a directory is part of OpenCode coming up, and derive its error message from the constant instead of hardcoding "10s". Raise AGENT_RUN_START_TIMEOUT_MS to 60s with a comment recording that it must stay above the slowest provider startup budget; it stays a literal because that module is provider-agnostic. Related to #3571. * test(server): drive the run start budget through a real AgentManager The run start boundary tests replaced AgentManager.waitForAgentRunStart with a bare timer, so they never exercised the production run-state transition or the agent_state subscription the real wait depends on. That path could regress while the tests stayed green. Build the scenario on a real AgentManager with a provider session that holds its turn start open for a configurable span, so pendingRun.started, the lifecycle change to "running", and the agent_state notification are what the wait actually observes. Budgets stay independent literals and the settled-before/after assertions are unchanged. * fix(server): describe agent run start timeouts --------- Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>
This commit is contained in:
co-authored by
Mohamed Boudra
parent
8f0c35e924
commit
60a5344cc6
@@ -1,5 +1,9 @@
|
||||
import { expect, it, test, vi } from "vitest";
|
||||
import pino, { type Logger } from "pino";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { createTestLogger } from "../../test-utils/test-logger.js";
|
||||
import { AgentManager } from "./agent-manager.js";
|
||||
@@ -8,8 +12,15 @@ import {
|
||||
formatSystemNotificationPrompt,
|
||||
isSystemInjectedEnvelope,
|
||||
setupFinishNotification,
|
||||
waitForAgentRunStartWithTimeout,
|
||||
} from "./agent-prompt.js";
|
||||
import type { AgentManagerEvent, ManagedAgent } from "./agent-manager.js";
|
||||
import type {
|
||||
AgentClient,
|
||||
AgentRunResult,
|
||||
AgentSession,
|
||||
AgentStreamEvent,
|
||||
} from "./agent-sdk-types.js";
|
||||
|
||||
interface CapturedLogger {
|
||||
logger: Logger;
|
||||
@@ -520,3 +531,234 @@ it("does not notify archived callers", async () => {
|
||||
expect(streamAgentSpy).not.toHaveBeenCalled();
|
||||
expect(replaceAgentRunSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// Deliberately independent literals rather than the production constants these tests
|
||||
// guard: deriving the boundaries from AGENT_RUN_START_TIMEOUT_MS would keep the tests
|
||||
// green if that constant were shortened back under a provider's startup budget.
|
||||
const EXPECTED_RUN_START_BUDGET_MS = 60_000;
|
||||
// The slowest provider startup budget the run-start wait has to sit outside of today
|
||||
// (OpenCode's OPENCODE_SERVER_STARTUP_TIMEOUT_MS).
|
||||
const SLOWEST_PROVIDER_STARTUP_BUDGET_MS = 30_000;
|
||||
|
||||
const RUN_START_TEST_CAPABILITIES = {
|
||||
supportsStreaming: false,
|
||||
supportsSessionPersistence: false,
|
||||
supportsSessionListing: true,
|
||||
supportsDynamicModes: false,
|
||||
supportsMcpServers: false,
|
||||
supportsReasoningStream: false,
|
||||
supportsToolInvocations: false,
|
||||
} as const;
|
||||
|
||||
/**
|
||||
* Provider session whose turn start is held open for a configurable span, so the real
|
||||
* AgentManager run-state transition (pendingRun.started -> lifecycle "running" ->
|
||||
* agent_state) is what the run-start wait observes. `startDelayMs: null` never starts.
|
||||
*/
|
||||
class SlowStartAgentSession implements AgentSession {
|
||||
readonly provider = "codex" as const;
|
||||
readonly capabilities = RUN_START_TEST_CAPABILITIES;
|
||||
readonly id = randomUUID();
|
||||
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
|
||||
private releaseStartTurn!: () => void;
|
||||
private readonly released = new Promise<void>((resolve) => {
|
||||
this.releaseStartTurn = resolve;
|
||||
});
|
||||
|
||||
constructor(private readonly startDelayMs: number | null) {}
|
||||
|
||||
async run(): Promise<AgentRunResult> {
|
||||
return { sessionId: this.id, finalText: "", timeline: [] };
|
||||
}
|
||||
|
||||
/** Teardown hook so a never-starting turn cannot wedge the suite. */
|
||||
release(): void {
|
||||
this.releaseStartTurn();
|
||||
}
|
||||
|
||||
async startTurn(): Promise<{ turnId: string }> {
|
||||
await new Promise<void>((resolve) => {
|
||||
if (this.startDelayMs !== null) {
|
||||
setTimeout(resolve, this.startDelayMs);
|
||||
}
|
||||
void this.released.then(resolve);
|
||||
});
|
||||
const turnId = "turn-1";
|
||||
setTimeout(() => {
|
||||
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
|
||||
this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
|
||||
}, 0);
|
||||
return { turnId };
|
||||
}
|
||||
|
||||
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
|
||||
this.subscribers.add(callback);
|
||||
return () => {
|
||||
this.subscribers.delete(callback);
|
||||
};
|
||||
}
|
||||
|
||||
pushEvent(event: AgentStreamEvent): void {
|
||||
for (const callback of this.subscribers) {
|
||||
callback(event);
|
||||
}
|
||||
}
|
||||
|
||||
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {}
|
||||
|
||||
async getRuntimeInfo() {
|
||||
return { provider: this.provider, sessionId: this.id, model: null, modeId: null };
|
||||
}
|
||||
|
||||
async getAvailableModes() {
|
||||
return [];
|
||||
}
|
||||
|
||||
async getCurrentMode() {
|
||||
return null;
|
||||
}
|
||||
|
||||
async setMode(): Promise<void> {}
|
||||
|
||||
getPendingPermissions() {
|
||||
return [];
|
||||
}
|
||||
|
||||
async respondToPermission(): Promise<void> {}
|
||||
|
||||
describePersistence() {
|
||||
return { provider: this.provider, sessionId: this.id };
|
||||
}
|
||||
|
||||
async interrupt(): Promise<void> {}
|
||||
|
||||
async close(): Promise<void> {}
|
||||
}
|
||||
|
||||
class SlowStartAgentClient implements AgentClient {
|
||||
readonly provider = "codex" as const;
|
||||
readonly capabilities = RUN_START_TEST_CAPABILITIES;
|
||||
readonly sessions: SlowStartAgentSession[] = [];
|
||||
|
||||
constructor(private readonly startDelayMs: number | null) {}
|
||||
|
||||
async isAvailable(): Promise<boolean> {
|
||||
return true;
|
||||
}
|
||||
|
||||
async createSession(): Promise<AgentSession> {
|
||||
const session = new SlowStartAgentSession(this.startDelayMs);
|
||||
this.sessions.push(session);
|
||||
return session;
|
||||
}
|
||||
|
||||
async fetchCatalog() {
|
||||
return { models: [], modes: [] };
|
||||
}
|
||||
|
||||
async resumeSession(): Promise<AgentSession> {
|
||||
return await this.createSession();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Real AgentManager driving a real agent, so the run-start wait exercises the production
|
||||
* run-state and agent_state subscription path rather than a replaced method.
|
||||
*/
|
||||
async function createRunStartScenario(startDelayMs: number | null): Promise<{
|
||||
agentManager: AgentManager;
|
||||
agentId: string;
|
||||
startRun: () => Promise<void>;
|
||||
cleanup: () => Promise<void>;
|
||||
}> {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-run-start-budget-"));
|
||||
const client = new SlowStartAgentClient(startDelayMs);
|
||||
const agentManager = new AgentManager({
|
||||
clients: { codex: client },
|
||||
logger: createTestLogger(),
|
||||
});
|
||||
const snapshot = await agentManager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
|
||||
let drained: Promise<void> = Promise.resolve();
|
||||
return {
|
||||
agentManager,
|
||||
agentId: snapshot.id,
|
||||
// streamAgent registers the pending run synchronously, so the wait always observes it.
|
||||
startRun: async () => {
|
||||
const run = agentManager.streamAgent(snapshot.id, "start the run");
|
||||
drained = (async () => {
|
||||
for await (const _event of run) {
|
||||
// Drain whatever the turn produces.
|
||||
}
|
||||
})().catch(() => undefined);
|
||||
},
|
||||
cleanup: async () => {
|
||||
// Release any turn still held open, then close. The drain is deliberately not
|
||||
// awaited: depending on how far the turn got, the stream ends either from the
|
||||
// release or from the close, and teardown must not depend on which.
|
||||
for (const session of client.sessions) {
|
||||
session.release();
|
||||
}
|
||||
await agentManager.closeAgent(snapshot.id).catch(() => undefined);
|
||||
void drained;
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
test("waiting for a run start outlasts the slowest provider startup budget", async () => {
|
||||
// A provider is still allowed to be starting here, so the outer wait must not abort it.
|
||||
const scenario = await createRunStartScenario(SLOWEST_PROVIDER_STARTUP_BUDGET_MS + 5_000);
|
||||
vi.useFakeTimers();
|
||||
|
||||
try {
|
||||
await scenario.startRun();
|
||||
const wait = waitForAgentRunStartWithTimeout(scenario.agentManager, scenario.agentId);
|
||||
let settled = false;
|
||||
const markSettled = () => {
|
||||
settled = true;
|
||||
};
|
||||
void wait.then(markSettled, markSettled);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(SLOWEST_PROVIDER_STARTUP_BUDGET_MS);
|
||||
expect(settled).toBe(false);
|
||||
expect(scenario.agentManager.getAgent(scenario.agentId)?.lifecycle).not.toBe("running");
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
await expect(wait).resolves.toBeUndefined();
|
||||
expect(scenario.agentManager.getAgent(scenario.agentId)?.lifecycle).toBe("running");
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
await scenario.cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
test("waiting for a run start still gives up at the run start budget", async () => {
|
||||
const scenario = await createRunStartScenario(null);
|
||||
vi.useFakeTimers();
|
||||
|
||||
try {
|
||||
await scenario.startRun();
|
||||
const wait = waitForAgentRunStartWithTimeout(scenario.agentManager, scenario.agentId);
|
||||
const rejection = expect(wait).rejects.toThrow(
|
||||
"codex run did not start within 60 seconds (phase: run start)",
|
||||
);
|
||||
let settled = false;
|
||||
const markSettled = () => {
|
||||
settled = true;
|
||||
};
|
||||
void wait.then(markSettled, markSettled);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(EXPECTED_RUN_START_BUDGET_MS - 1);
|
||||
expect(settled).toBe(false);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await rejection;
|
||||
expect(settled).toBe(true);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
await scenario.cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -212,14 +212,35 @@ export interface StartCreatedAgentInitialPromptParams {
|
||||
logger: Logger;
|
||||
}
|
||||
|
||||
const AGENT_RUN_START_TIMEOUT_MS = 15_000;
|
||||
/**
|
||||
* Outer bound on a run reaching "started" after dispatch.
|
||||
*
|
||||
* This wraps provider startup, so it MUST stay larger than the slowest provider's own
|
||||
* startup budget — otherwise it aborts a start the provider was still allowed to be
|
||||
* working on, and the provider's budget can never apply. OpenCode is the slowest today:
|
||||
* up to 30s for the server to boot (OPENCODE_SERVER_STARTUP_TIMEOUT_MS) and then a
|
||||
* session.create on the same budget, so this is deliberately set well above 30s.
|
||||
*
|
||||
* Not derived from the provider constant on purpose: this module is provider-agnostic
|
||||
* and must not depend on a specific provider's internals.
|
||||
*/
|
||||
const AGENT_RUN_START_TIMEOUT_MS = 60_000;
|
||||
|
||||
export async function waitForAgentRunStartWithTimeout(
|
||||
agentManager: AgentManager,
|
||||
agentId: string,
|
||||
): Promise<void> {
|
||||
const provider = agentManager.getAgent(agentId)?.provider ?? "provider";
|
||||
const startAbort = new AbortController();
|
||||
const startTimeout = setTimeout(() => startAbort.abort("timeout"), AGENT_RUN_START_TIMEOUT_MS);
|
||||
const startTimeout = setTimeout(
|
||||
() =>
|
||||
startAbort.abort(
|
||||
new Error(
|
||||
`${provider} run did not start within ${AGENT_RUN_START_TIMEOUT_MS / 1000} seconds (phase: run start)`,
|
||||
),
|
||||
),
|
||||
AGENT_RUN_START_TIMEOUT_MS,
|
||||
);
|
||||
|
||||
try {
|
||||
await agentManager.waitForAgentRunStart(agentId, { signal: startAbort.signal });
|
||||
|
||||
@@ -343,6 +343,81 @@ describe("OpenCodeAgentClient adapter smoke tests", () => {
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}, 60_000);
|
||||
|
||||
test("creates a session when session.create needs more than ten seconds", async () => {
|
||||
vi.useFakeTimers();
|
||||
const cwd = tmpCwd();
|
||||
const runtime = new TestOpenCodeHarness();
|
||||
const openCode = new TestOpenCodeClient();
|
||||
const slowCreate = createTestDeferred<void>();
|
||||
openCode.sessionCreateImplementation = async () => {
|
||||
await slowCreate.promise;
|
||||
return { data: { id: "session-1" } };
|
||||
};
|
||||
runtime.enqueueClient(openCode);
|
||||
const client = new OpenCodeAgentClient(logger, undefined, {
|
||||
serverManager: runtime,
|
||||
createClient: runtime.createClient,
|
||||
});
|
||||
|
||||
try {
|
||||
const creation = client.createSession(buildConfig(cwd));
|
||||
let settled = false;
|
||||
const markSettled = () => {
|
||||
settled = true;
|
||||
};
|
||||
void creation.then(markSettled, markSettled);
|
||||
|
||||
// Under CPU contention a per-directory bootstrap has been measured well past ten
|
||||
// seconds, which the previous budget failed outright.
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(settled).toBe(false);
|
||||
|
||||
slowCreate.resolve();
|
||||
const session = await creation;
|
||||
expect(session.provider).toBe("opencode");
|
||||
expect(openCode.calls.sessionCreate).toHaveLength(1);
|
||||
await session.close();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
slowCreate.resolve();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("bounds session.create at the server startup budget", async () => {
|
||||
vi.useFakeTimers();
|
||||
const cwd = tmpCwd();
|
||||
const runtime = new TestOpenCodeHarness();
|
||||
const openCode = new TestOpenCodeClient();
|
||||
openCode.sessionCreateImplementation = () => new Promise<never>(() => undefined);
|
||||
runtime.enqueueClient(openCode);
|
||||
const client = new OpenCodeAgentClient(logger, undefined, {
|
||||
serverManager: runtime,
|
||||
createClient: runtime.createClient,
|
||||
});
|
||||
|
||||
try {
|
||||
const creation = client.createSession(buildConfig(cwd));
|
||||
const rejection = expect(creation).rejects.toThrow("OpenCode session.create timed out");
|
||||
let settled = false;
|
||||
const markSettled = () => {
|
||||
settled = true;
|
||||
};
|
||||
void creation.then(markSettled, markSettled);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(EXPECTED_STARTUP_BUDGET_MS - 1);
|
||||
expect(settled).toBe(false);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await rejection;
|
||||
expect(settled).toBe(true);
|
||||
expect(runtime.acquisitions.at(-1)?.releaseCount).toBe(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("archives and unarchives the durable native session through client hooks", async () => {
|
||||
const cwd = tmpCwd();
|
||||
const runtime = new TestOpenCodeHarness();
|
||||
|
||||
@@ -1396,10 +1396,15 @@ export class OpenCodeAgentClient implements AgentClient {
|
||||
});
|
||||
|
||||
try {
|
||||
// Creating the first session for a directory is part of OpenCode coming up, so it
|
||||
// shares the server startup budget instead of a shorter one that fails agent
|
||||
// creation on contended cold starts.
|
||||
const response = await withTimeout(
|
||||
client.session.create({ directory: openCodeConfig.cwd }),
|
||||
10_000,
|
||||
"OpenCode session.create timed out after 10s",
|
||||
OPENCODE_SERVER_STARTUP_TIMEOUT_MS,
|
||||
`OpenCode session.create timed out after ${Math.round(
|
||||
OPENCODE_SERVER_STARTUP_TIMEOUT_MS / 1000,
|
||||
)}s`,
|
||||
);
|
||||
|
||||
if (response.error) {
|
||||
|
||||
+4
@@ -145,6 +145,7 @@ export class TestOpenCodeClient {
|
||||
sessionCommandEvents: unknown[] = [idleEvent()];
|
||||
sessionCommandResponse: OpenCodeResponse = {};
|
||||
sessionCreateResponse: OpenCodeResponse = { data: { id: "session-1" } };
|
||||
sessionCreateImplementation: ((parameters: unknown) => Promise<OpenCodeResponse>) | null = null;
|
||||
sessionDeleteResponse: OpenCodeResponse = {};
|
||||
sessionChildrenResponses: OpenCodeResponse[] = [];
|
||||
sessionChildrenImplementation:
|
||||
@@ -294,6 +295,9 @@ export class TestOpenCodeClient {
|
||||
},
|
||||
create: async (parameters: unknown) => {
|
||||
this.calls.sessionCreate.push(parameters);
|
||||
if (this.sessionCreateImplementation) {
|
||||
return await this.sessionCreateImplementation(parameters);
|
||||
}
|
||||
return this.sessionCreateResponse;
|
||||
},
|
||||
delete: async (parameters: unknown) => {
|
||||
|
||||
Reference in New Issue
Block a user