diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 6b3c05c9..f2b90fb3 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -5,13 +5,13 @@ }, "metadata": { "description": "Run jobs on Codex, Grok Build, ZCode and Kimi Code from Claude Code like Claude agents.", - "version": "3.0.0-dev.24" + "version": "3.0.0-dev.26" }, "plugins": [ { "name": "relay", "description": "Run jobs on Codex, Grok Build, ZCode and Kimi Code from Claude Code like Claude agents.", - "version": "3.0.0-dev.24", + "version": "3.0.0-dev.26", "author": { "name": "Libar" }, diff --git a/README.md b/README.md index 8bc495d9..de01ac55 100644 --- a/README.md +++ b/README.md @@ -372,11 +372,20 @@ lock stays queued and cancellable; a stuck-lock diagnostic identifies the holder Recognized usage refusals pause Codex, Grok, Kimi or ZCode for 30 minutes, or a definite reset within six hours. Rate pauses start at 60 seconds and double to 15 minutes; Kimi concurrency refusals -pause for 10 minutes. After a pause, one generation-fenced probe runs; its success clears the -pause. Refused jobs remain failed for explicit resume. Pauses follow the selected backend account home; default homes retain the existing pause files. +pause for 10 minutes. After a pause, one job probes the provider; the pause ends only when that job gets an actual answer from the +provider, not when it merely starts. Refused jobs remain failed for explicit resume. Pauses follow the selected backend account home; default homes retain the existing pause files. Grok recognizes explicit weekly/usage/credit exhaustion, rate-limit text and provider 429 errors; its usage classifier has not been verified against a live exhausted account. A live probe is never replaced by age. -These pauses gate new starts, not requests or retries inside running turns. +These pauses gate new starts and later turns on other jobs while a running turn keeps its own retries. + +| Backend | Paused on | When | +|---|---|---| +| Codex | usage or rate | at first retry, including subagent signals | +| Grok | weekly credits or rate | at first retry, unverified with a live exhausted account | +| Kimi | usage, rate or concurrency | at first retry | +| ZCode | usage or rate | when the turn fails; runtime retries are not exposed | + +Grok's web search cannot be turned off. With the guard effective, Kimi's internal retry default is 3 unless explicitly configured. Turn it off for an invocation with `RELAY_ADMISSION=off`, or domain-wide with @@ -384,6 +393,14 @@ Turn it off for an invocation with `RELAY_ADMISSION=off`, or domain-wide with `~/.relay-jobs/admission/config.json` (or `RELAY_ADMISSION_HOME/config.json`), shared across jobs homes. A custom jobs home cannot disable the guard. Old weight/class controls are ignored. +Backend startup has a 2-minute deadline; session setup has 5 minutes, and a steer or cancel request has +30 seconds. ACP cancellation is a notification and has no reply to wait for. Turns have no time limit. +After 10 minutes without notifications, relay checks the backend every 5 minutes. Two unanswered +30-second pings restart it once, using the normal server recovery path. A backend answering pings can +keep a silent build or sleep running indefinitely. Relay tests ping support on each connection; an +unanswered initial ping disables that watchdog and records a note, while control deadlines still apply. +Job JSON and events record whether the backend is answering or relay is checking it. + After a job starts, relay samples memory used by the job and its tools. Job status shows `memory:`, and guard status shows `relay jobs use`. On Linux, when a user systemd manager is available, every worker runs in its own scope under `relay-jobs.slice`; its tools inherit `oom_score_adj=500`, making diff --git a/package-lock.json b/package-lock.json index a54569b0..dedf8f68 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@libar-dev/relay-plugin-cc", - "version": "3.0.0-dev.24", + "version": "3.0.0-dev.26", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@libar-dev/relay-plugin-cc", - "version": "3.0.0-dev.24", + "version": "3.0.0-dev.26", "license": "Apache-2.0", "devDependencies": { "@types/node": "^25.5.0", diff --git a/package.json b/package.json index b14d6920..e0e400f5 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@libar-dev/relay-plugin-cc", - "version": "3.0.0-dev.24", + "version": "3.0.0-dev.26", "private": true, "type": "module", "description": "Run jobs on Codex, Grok Build, ZCode and Kimi Code from Claude Code like Claude agents.", diff --git a/plugins/relay/.claude-plugin/plugin.json b/plugins/relay/.claude-plugin/plugin.json index 9f9a6327..c7af9f0d 100644 --- a/plugins/relay/.claude-plugin/plugin.json +++ b/plugins/relay/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "relay", - "version": "3.0.0-dev.24", + "version": "3.0.0-dev.26", "description": "Run jobs on Codex, Grok Build, ZCode and Kimi Code from Claude Code like Claude agents.", "author": { "name": "Libar" diff --git a/plugins/relay/CHANGELOG.md b/plugins/relay/CHANGELOG.md index 5d3edc40..da3c30e3 100644 --- a/plugins/relay/CHANGELOG.md +++ b/plugins/relay/CHANGELOG.md @@ -8,6 +8,12 @@ its entries here. Changed: +- Backend controls have request deadlines; silent turns use a capability-tested ping watchdog. + Responsive long-running tools have no turn time limit. Unresponsive backends follow the existing + one-restart recovery path. +- Provider probes require model-response evidence and slow startup grants are checked again before + sending. Codex, Grok and Kimi retry signals pause new work immediately for the selected account. + - Foreground `task` workers run independently of the caller; Ctrl-C leaves a dispatched job running. - Jobs own their execution groups and stop their tools at backend loss, cancellation and completion. Linux workers use per-job scopes whenever a user manager is available, and set OOM score 500. @@ -21,6 +27,14 @@ Changed: Fixed: +- Provider retry refusals preserve a live probe’s claim and generation, including deferred refusals + replayed before a healthy answer clears the pause. + +- RPC input coalesces tiny fragments so per-fragment object overhead cannot escape the line-memory bound. + +- ZCode duplicate waits cancel promptly and report uncertain starts accurately. Duplicate steer + acknowledgements stay owed until inclusion is proved, and cancellation prevents discarded-input resends. + - Repeated cleanup preserves a predecessor's first observed death, keeping the recorded ordering of predecessor exit and replacement startup intact. - Worker heartbeats keep process accounting active with the load guard off without creating diff --git a/plugins/relay/scripts/lib/backends/codex/connection.mjs b/plugins/relay/scripts/lib/backends/codex/connection.mjs index e3912402..0c0d233a 100644 --- a/plugins/relay/scripts/lib/backends/codex/connection.mjs +++ b/plugins/relay/scripts/lib/backends/codex/connection.mjs @@ -1,3 +1,4 @@ +import { CONTROL_MS } from "../transport/stdio-rpc.mjs"; /** @typedef {import("../types").Connection} Connection */ import { CodexAppServerClient } from "./client.mjs"; import { clientIdsInThread, openThread, runTurnOnThread } from "./turn.mjs"; @@ -64,7 +65,7 @@ class CodexConnection { }; this.steerTracker.turnSteerHandler?.("requested", params); try { - const response = await this.client.request("turn/steer", params); + const response = await this.client.request("turn/steer", params, { timeoutMs: CONTROL_MS }); this.steerTracker.turnSteerHandler?.("settled", params, true); return response; } catch (error) { @@ -73,7 +74,9 @@ class CodexConnection { } } - async interrupt(turnId) { await this.client.request("turn/interrupt", { threadId: this.sessionId, turnId }); } + async interrupt(turnId) { await this.client.request("turn/interrupt", { threadId: this.sessionId, turnId }, { timeoutMs: CONTROL_MS }); } + get pingMethod() { return "account/read"; } + async ping() { await this.client.request("account/read", { refreshToken: false }, { timeoutMs: CONTROL_MS, wedgeOnTimeout: false }); } async close() { await this.client.close(); } } diff --git a/plugins/relay/scripts/lib/backends/codex/turn.mjs b/plugins/relay/scripts/lib/backends/codex/turn.mjs index f0532387..303ac0ee 100644 --- a/plugins/relay/scripts/lib/backends/codex/turn.mjs +++ b/plugins/relay/scripts/lib/backends/codex/turn.mjs @@ -1,3 +1,4 @@ +import { SESSION_MS } from "../transport/stdio-rpc.mjs"; /** * @typedef {import("./protocol").AppServerNotification} AppServerNotification * @typedef {import("./protocol").ThreadResumeParams} ThreadResumeParams @@ -31,6 +32,8 @@ * reasoningSummary: string[], * error: unknown, * onTurnId: ((turnId: string) => void) | null, + * onProviderResponse?: () => void, + * onProviderLimit?: (error: unknown) => void, * responseReceived: boolean, * workSeen: boolean * }} TurnCaptureState @@ -200,6 +203,8 @@ function createTurnCaptureState(threadId, options = {}) { reasoningSummary: [], error: null, onTurnId: options.onTurnId ?? null, + onProviderResponse: options.onProviderResponse, + onProviderLimit: options.onProviderLimit, responseReceived: false, // Any item beyond the user's own message: a failed turn that did work is never resent. workSeen: false @@ -383,6 +388,8 @@ function applyTurnNotification(state, message) { (message.method.startsWith("item/") || message.method === "turn/started")) { invalidateInference(state); } + if (["item/agentMessage/delta", "item/reasoning/summaryTextDelta", "item/reasoning/textDelta"].includes(message.method) || + (["item/started", "item/completed"].includes(message.method) && ["agentMessage", "reasoning", "commandExecution", "fileChange", "mcpToolCall", "webSearch", "collabAgentToolCall"].includes(message.params.item?.type))) state.onProviderResponse?.(); switch (message.method) { case "thread/started": trackThread(state, message.params.thread.id); @@ -414,6 +421,7 @@ function applyTurnNotification(state, message) { break; case "error": state.error = message.params.error; + if (message.params.willRetry) state.onProviderLimit?.(message.params.error); break; case "turn/completed": if ((message.params.threadId ?? null) !== state.threadId) { @@ -541,7 +549,7 @@ export function connectOptions(options) { export async function startThread(client, cwd, options = {}) { - const response = await client.request("thread/start", buildThreadParams(cwd, options)); + const response = await client.request("thread/start", buildThreadParams(cwd, options), { timeoutMs: SESSION_MS }); const threadId = response.thread.id; if (options.threadName) { try { @@ -564,7 +572,7 @@ export async function startThread(client, cwd, options = {}) { async function resumeThread(client, threadId, cwd, options = {}) { - return client.request("thread/resume", buildResumeParams(threadId, cwd, options)); + return client.request("thread/resume", buildResumeParams(threadId, cwd, options), { timeoutMs: SESSION_MS }); } @@ -608,7 +616,7 @@ export async function runTurnOnThread(client, threadId, options = {}) { ...(options.approvalPolicy ? { approvalPolicy: options.approvalPolicy } : {}), outputSchema: options.outputSchema ?? null }), - { onTurnId: options.onTurnId, steerTracker: options.steerTracker } + { onTurnId: options.onTurnId, steerTracker: options.steerTracker, onProviderResponse: options.onProviderResponse, onProviderLimit: options.onProviderLimit } ); return { diff --git a/plugins/relay/scripts/lib/backends/grok/connection.mjs b/plugins/relay/scripts/lib/backends/grok/connection.mjs index 3b489c21..537b941a 100644 --- a/plugins/relay/scripts/lib/backends/grok/connection.mjs +++ b/plugins/relay/scripts/lib/backends/grok/connection.mjs @@ -1,3 +1,4 @@ +import { CONTROL_MS, SESSION_MS } from "../transport/stdio-rpc.mjs"; /** @typedef {import("../types").Connection} Connection */ import { ACP_INITIALIZE_PARAMS, ACP_CHUNKS, acpUpdate, openSession, setConfigOption, promptSession, cancelSession, promptCancelled, promptResult, answerPermission } from "../transport/acp.mjs"; import { heldMessage, evidenceKey, countHeld, grokSessionMessages } from "./history.mjs"; @@ -96,6 +97,11 @@ class GrokConnection { } get server() { return this.client; } + get pingMethod() { return "relay/ping"; } + async ping() { + try { await this.client.request("relay/ping", {}, { timeoutMs: CONTROL_MS, wedgeOnTimeout: false }); } + catch (error) { if (error.rpcCode !== -32601) throw error; } + } async close() { await this.client.close(); } saveState(patch) { @@ -235,14 +241,14 @@ class GrokConnection { const promptId = this.running; this.steers = (this.steers ?? 0) + 1; try { - const response = await client.request("_x.ai/interject", { sessionId, text: message.text }); + const response = await client.request("_x.ai/interject", { sessionId, text: message.text }, { timeoutMs: CONTROL_MS }); const status = response.result?.status ?? response.status; if (status !== "queued") throw new Error(`Grok did not queue the interjection: ${status ?? "unknown"}`); // Queued is not durable: Grok can lose the interjection before it persists it. if (promptId) this.saveState({ promptIds: { ...this.state.promptIds, [message.id]: promptId } }); return { turnId, promptId, unconfirmed: true }; } catch (error) { - if (!client.exitSeen) this.saveDeliveries(this.settle(delivery, false)); + if (!client.exitSeen && !client.wedged) this.saveDeliveries(this.settle(delivery, false)); throw error; } finally { this.steers -= 1; @@ -287,6 +293,8 @@ class GrokConnection { check(); } const update = params.update; + if (notification.method === "session/update" && ["agent_message_chunk", "agent_thought_chunk", "tool_call"].includes(update?.sessionUpdate)) options.onProviderResponse?.(); + if (notification.method === "_x.ai/session/update" && update?.sessionUpdate === "retry_state") options.onProviderLimit?.({ message: update.reason ?? update.retryState?.reason, data: { retryState: update.retryState ?? update } }); const standard = acpUpdate(notification); if (standard.didWork || update?.sessionUpdate === "tool_call_delta_chunk") didWork = true; if (standard.startsTool) text = ""; @@ -354,6 +362,6 @@ export async function connect(request, hooks) { } }); try { await client.initialize(); } - catch (error) { await client.close(); throw new Error(client.exitInfo ? grokEndMessage(client.exitInfo) : error.message); } + catch (error) { await client.close(); if (error.name === "TransportTimeout") throw error; throw new Error(client.exitInfo ? grokEndMessage(client.exitInfo) : error.message); } return new GrokConnection(client, request, { ...hooks, state: { ...hooks.state, binary } }); } diff --git a/plugins/relay/scripts/lib/backends/kimi/connection.mjs b/plugins/relay/scripts/lib/backends/kimi/connection.mjs index 583874af..fd282ec0 100644 --- a/plugins/relay/scripts/lib/backends/kimi/connection.mjs +++ b/plugins/relay/scripts/lib/backends/kimi/connection.mjs @@ -1,3 +1,4 @@ +import { CONTROL_MS, SESSION_MS } from "../transport/stdio-rpc.mjs"; import { admissionEnabled, DEFAULTS } from "../../worker/admission.mjs"; import { shellCommandMarkdown } from "../../util/shell.mjs"; /** @typedef {import("../types").Connection} Connection */ @@ -137,6 +138,11 @@ export class KimiConnection { get server() { return this.client; } + get pingMethod() { return "relay/ping"; } + async ping() { + try { await this.client.request("relay/ping", {}, { timeoutMs: CONTROL_MS, wedgeOnTimeout: false }); } + catch (error) { if (error.rpcCode !== -32601) throw error; } + } async close() { await this.stopObserver("the connection closed"); await this.reads?.close(); @@ -173,7 +179,7 @@ export class KimiConnection { sessionId = sessionId ?? response.sessionId; this.sessionId = sessionId; // Kimi keeps model and effort across a load but not the mode, so all three go every time. - await client.request("session/set_mode", { sessionId, modeId: KIMI_MODES[request.mode] }); + await client.request("session/set_mode", { sessionId, modeId: KIMI_MODES[request.mode] }, { timeoutMs: SESSION_MS }); for (const [configId, value] of [["model", request.model], ["thinking", request.effort]]) { if (value != null) await setConfigOption(client, sessionId, configId, value); } @@ -215,9 +221,13 @@ export class KimiConnection { const wire = this.wire.read(); // Where this prompt's records can start: a record at or after it came after the send. const from = wire.consumed; + wire.onRecord = record => { + if (record.type === "turn.step.retrying") options.onProviderLimit?.(record.error ?? { message: record.errorMessage }); + }; const run = { turnId: null, didWork: false, from }; client.setNotificationHandler(notification => { if (notification.params?.sessionId !== sessionId) return; + if (notification.method === "session/update" && ["agent_message_chunk", "agent_thought_chunk", "tool_call"].includes(notification.params?.update?.sessionUpdate)) options.onProviderResponse?.(); if (acpUpdate(notification).didWork) run.didWork = true; }); const ended = client.exitPromise.then(() => { throw new Error(kimiEndMessage(client.exitInfo)); }); @@ -238,6 +248,7 @@ export class KimiConnection { return { turnId: run.turnId, turnStatus: "failed", finalMessage: "", didWork: run.didWork, retryable: false, error: { message: error.message } }; } finally { await this.stopObserver(client.exitSeen || client.exitInfo ? "Kimi's server ended" : "the run ended"); + wire.onRecord = null; client.setNotificationHandler(null); } } @@ -377,6 +388,6 @@ export async function connect(request, hooks) { } }); try { await client.initialize(); } - catch (error) { await reads?.close(); await client.close(); throw new Error(client.exitInfo ? kimiEndMessage(client.exitInfo) : error.message); } + catch (error) { await reads?.close(); await client.close(); if (error.name === "TransportTimeout") throw error; throw new Error(client.exitInfo ? kimiEndMessage(client.exitInfo) : error.message); } return new KimiConnection(client, request, { ...hooks, state: { ...hooks.state, binary } }, reads); } diff --git a/plugins/relay/scripts/lib/backends/kimi/wire.mjs b/plugins/relay/scripts/lib/backends/kimi/wire.mjs index f913e91f..c3b8b851 100644 --- a/plugins/relay/scripts/lib/backends/kimi/wire.mjs +++ b/plugins/relay/scripts/lib/backends/kimi/wire.mjs @@ -65,6 +65,7 @@ export class KimiWire { */ constructor(locate, { io = fs, retainPrompt = () => false } = {}) { this.locate = locate; + this.onRecord = null; this.io = io; this.retainPrompt = retainPrompt; /** @type {string | null} */ @@ -123,6 +124,7 @@ export class KimiWire { if (!line.trim()) continue; let record; try { record = JSON.parse(line); } catch { continue; } + this.onRecord?.(record); this.apply(record, at); this.prune(); } diff --git a/plugins/relay/scripts/lib/backends/transport/acp.mjs b/plugins/relay/scripts/lib/backends/transport/acp.mjs index 9d47d818..64ee0ddb 100644 --- a/plugins/relay/scripts/lib/backends/transport/acp.mjs +++ b/plugins/relay/scripts/lib/backends/transport/acp.mjs @@ -1,3 +1,4 @@ +import { SESSION_MS } from "./stdio-rpc.mjs"; // The ACP messages Grok and Kimi use. Grok's queue and interjections, and Kimi's session record, // stay in their connections. export const ACP_INITIALIZE_PARAMS = { @@ -26,11 +27,11 @@ export function acpUpdate(notification) { export function openSession(client, cwd, sessionId) { return client.request(sessionId ? "session/load" : "session/new", { ...(sessionId ? { sessionId } : {}), cwd, mcpServers: [] - }); + }, { timeoutMs: SESSION_MS }); } export function setConfigOption(client, sessionId, configId, value) { - return client.request("session/set_config_option", { sessionId, configId, value }); + return client.request("session/set_config_option", { sessionId, configId, value }, { timeoutMs: SESSION_MS }); } // A client that serves file writes and terminals itself declares them, and the agent sends them. diff --git a/plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs b/plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs index ebc26209..97d25cbd 100644 --- a/plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs +++ b/plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs @@ -71,11 +71,25 @@ function createProtocolError(message, data) { } +export class TransportTimeout extends Error { + constructor(backend, method, timeoutMs) { + super(`${backend} stopped answering (${method}, ${timeoutMs / 1000} s)`); + this.name = "TransportTimeout"; + this.method = method; + this.timeoutMs = timeoutMs; + } +} + +export const CONTROL_MS = 30000; +export const SESSION_MS = 300000; + class AppServerClientBase { constructor(cwd, options = {}) { this.cwd = cwd; this.options = options; this.pending = new Map(); + this.wedged = null; + this.wedgedPromise = new Promise(resolve => { this.resolveWedged = resolve; }); this.nextId = 1; this.stderr = ""; this.closed = false; @@ -86,7 +100,7 @@ class AppServerClientBase { /** Called with every notification before the handler; never replaced by turn capture. */ this.notificationTap = options.notificationTap ?? null; this.exitInfo = null; - this.lineChunks = []; + this.lineBuffer = null; this.lineBytes = 0; this.transport = "unknown"; @@ -104,9 +118,11 @@ class AppServerClientBase { * @template {AppServerMethod} M * @param {M} method * @param {import("../codex/protocol").AppServerRequestParams} params + * @param {{ timeoutMs?: number, wedgeOnTimeout?: boolean }} [options] * @returns {Promise>} */ - request(method, params) { + request(method, params, { timeoutMs, wedgeOnTimeout = true } = {}) { + if (this.wedged) return Promise.reject(this.wedged); if (this.closed) { throw new Error(`${this.options.backendName ?? "codex"} server client is closed.`); } @@ -120,12 +136,27 @@ class AppServerClientBase { this.nextId += 1; return new Promise((resolve, reject) => { - this.pending.set(id, { resolve, reject, method }); - this.sendMessage({ id, method, params }); + let timer; + const finish = fn => value => { clearTimeout(timer); fn(value); }; + this.pending.set(id, { resolve: finish(resolve), reject: finish(reject), method }); + if (timeoutMs != null) timer = setTimeout(() => { + const error = new TransportTimeout(this.options.backendName ?? "codex", method, timeoutMs); + this.pending.delete(id); + if (wedgeOnTimeout) this.markWedged(error); + reject(error); + }, timeoutMs); + try { this.sendMessage({ id, method, params }); } + catch (error) { this.pending.delete(id); clearTimeout(timer); reject(error); } }); } + markWedged(error) { + if (this.wedged) return; + this.wedged = error; + this.resolveWedged(error); + } + notify(method, params = {}) { if (this.closed) { return; @@ -142,16 +173,23 @@ class AppServerClientBase { const piece = bytes.subarray(from, end === -1 ? bytes.length : end); this.lineBytes += piece.length; if (this.lineBytes > MAX_LINE_BYTES) { - this.lineChunks = []; + this.lineBuffer = null; this.lineBytes = 0; this.handleExit(createProtocolError(`${this.options.backendName ?? "codex"} sent a single message over 64 MiB; the connection was closed`)); void this.close(); return; } - this.lineChunks.push(piece); + // Keep one geometrically grown buffer, not one retained object per fragment. A peer + // trickling single bytes cannot amplify the payload bound into millions of Buffers. + if (!this.lineBuffer || this.lineBuffer.length < this.lineBytes) { + const buffer = Buffer.allocUnsafe(Math.min(MAX_LINE_BYTES, Math.max(4096, (this.lineBuffer?.length ?? 0) * 2, this.lineBytes))); + this.lineBuffer?.copy(buffer, 0, 0, this.lineBytes - piece.length); + this.lineBuffer = buffer; + } + piece.copy(this.lineBuffer, this.lineBytes - piece.length); if (end === -1) break; - const line = Buffer.concat(this.lineChunks, this.lineBytes).toString("utf8"); - this.lineChunks = []; + const line = this.lineBuffer.toString("utf8", 0, this.lineBytes); + this.lineBuffer = null; this.lineBytes = 0; this.handleLine(line); if (this.exitResolved) return; @@ -290,7 +328,7 @@ export class SpawnedServerClient extends AppServerClientBase { this.transportEnded = true; this.stdoutClosedBeforeExit = !this.exitSeen; this.partialStdoutBytes = this.lineBytes; - this.lineChunks = []; this.lineBytes = 0; + this.lineBuffer = null; this.lineBytes = 0; // EOF can arrive while the process is still alive. Stop it so pending requests // get the actual exit details, even if it left stderr or its event loop open. if (!this.exitSeen) void this.close(); @@ -313,7 +351,7 @@ export class SpawnedServerClient extends AppServerClientBase { await this.request(this.options.initializeMethod ?? "initialize", this.options.initializeParams ?? { clientInfo: this.options.clientInfo ?? DEFAULT_CLIENT_INFO, capabilities: this.options.capabilities ?? DEFAULT_CAPABILITIES - }); + }, { timeoutMs: 120000 }); if (!this.options.initializeParams) this.notify("initialized", {}); } @@ -330,7 +368,7 @@ export class SpawnedServerClient extends AppServerClientBase { if (this.exitResolved) return; if (!this.transportEnded) { this.partialStdoutBytes = this.lineBytes; - this.lineChunks = []; this.lineBytes = 0; + this.lineBuffer = null; this.lineBytes = 0; } this.proc.stdout.destroy(); this.proc.stderr.destroy(); diff --git a/plugins/relay/scripts/lib/backends/types.d.ts b/plugins/relay/scripts/lib/backends/types.d.ts index 6adbee77..51d084c7 100644 --- a/plugins/relay/scripts/lib/backends/types.d.ts +++ b/plugins/relay/scripts/lib/backends/types.d.ts @@ -159,6 +159,8 @@ export interface ConnectHooks { */ export interface Connection { readonly server: ServerState; + readonly pingMethod?: string; + ping?(): Promise; /** Open a new session, or resume `sessionId`. */ open(options: OpenOptions): Promise; /** Whether a work item reached the session before a previous worker died. */ @@ -181,6 +183,9 @@ export interface Connection { } export interface ServerState { + readonly wedged?: Error; + readonly wedgedPromise?: Promise; + markWedged?(error: Error): void; readonly exitInfo: ExitInfo | null; /** The process exited; its pipes may still be draining. */ readonly exitSeen: boolean; @@ -245,6 +250,12 @@ export interface InboxMessage { } export interface TurnInput { + /** Cancellation the worker has claimed, including while an initial acknowledgement waits. */ + isCancelled?(): boolean; + /** Evidence of model output, never a local acknowledgement or replay. */ + onProviderResponse?(): void; + /** A provider refusal while the harness continues its own retries. */ + onProviderLimit?(error: unknown): void; prompt: string; clientId: string; source: WorkItem["source"]; diff --git a/plugins/relay/scripts/lib/backends/zcode/connection.mjs b/plugins/relay/scripts/lib/backends/zcode/connection.mjs index 2430c969..472768dc 100644 --- a/plugins/relay/scripts/lib/backends/zcode/connection.mjs +++ b/plugins/relay/scripts/lib/backends/zcode/connection.mjs @@ -1,3 +1,4 @@ +import { CONTROL_MS, SESSION_MS } from "../transport/stdio-rpc.mjs"; /** @typedef {import("../types").Connection} Connection */ // ZCode's app-server protocol, isolated from the worker's message and process accounting. // Sessions and turns go through ZCode Protocol v4 commands (`v4/command`); a turn's start and @@ -146,12 +147,12 @@ function live(connection, promise) { return Promise.race([promise, connection.server.exitPromise.then(() => { throw new Error(zcodeEndMessage(connection.server.exitInfo)); }), ...(connection.session?.closedPromise ? [connection.session.closedPromise] : [])]); } -function call(connection, method, params) { - return live(connection, Promise.resolve().then(() => connection.client.request(method, params))); +function call(connection, method, params, options = { timeoutMs: SESSION_MS }) { + return live(connection, Promise.resolve().then(() => connection.client.request(method, params, options))); } -function command(connection, sessionId, type, payload, commandId = randomUUID()) { - return call(connection, "v4/command", { commandId, clientId: CLIENT_ID, sessionId, type, payload, issuedAt: Date.now() }); +function command(connection, sessionId, type, payload, commandId = randomUUID(), options = { timeoutMs: SESSION_MS }) { + return call(connection, "v4/command", { commandId, clientId: CLIENT_ID, sessionId, type, payload, issuedAt: Date.now() }, options); } // A stop with steers still queued pauses ZCode's queue, durably, and ZCode then refuses every @@ -163,7 +164,7 @@ function sendText(connection, commandId, text, requestedDelivery) { return command(connection, z.sessionId, "sendText", { text, requestedDelivery, modelSelection: z.selection, mode: zcodeSessionMode(z.request), heldQueueDisposition: "clearQueueAndSend", ...(denied.length ? { toolDisallowlist: denied } : {}) - }, commandId); + }, commandId, { timeoutMs: requestedDelivery === "guide" ? CONTROL_MS : undefined }); } function refused(ack, what) { @@ -194,11 +195,15 @@ function trackSession(connection, notification) { const z = connection.session; if (notification.method !== "session/event" || notification.params?.sessionId !== z.sessionId) return; const { type, turnId, payload = {} } = notification.params; + if ((type === "model.streaming" && ["text_delta", "reasoning_delta", "tool_call"].includes(payload.kind)) || + (type === "tool.updated" && payload.kind === "started") || + (type === "part.delta" && ["text", "reasoning", "tool-call", "tool_call"].includes(payload.type ?? payload.part?.type)) || + (type === "turn.completed" && payload.resultType === "success")) z.onProviderResponse?.(); if (type === "turn.started") { z.running.set(turnId, payload.inputId ?? null); if (payload.inputId) z.joined.add(baseId(payload.inputId)); z.outstanding.delete(payload.inputId); - if (z.cancelRequested) void command(connection, z.sessionId, "stop", {}).catch(() => {}); + if (z.cancelRequested) void command(connection, z.sessionId, "stop", {}, undefined, { timeoutMs: CONTROL_MS }).catch(() => {}); } else if (type === "turn.steerDrained") { for (const id of inputIdsOf(payload)) { z.joined.add(baseId(id)); @@ -285,6 +290,8 @@ class ZcodeConnection { } get server() { return this.client; } + get pingMethod() { return "runtime/capabilities"; } + async ping() { await this.client.request("runtime/capabilities", {}, { timeoutMs: CONTROL_MS, wedgeOnTimeout: false }); } async close() { await this.client.close(); } /** @param {import("../types").OpenOptions} options */ @@ -305,6 +312,7 @@ class ZcodeConnection { * closed: string | null, * selection: ReturnType | null, * changed: (() => void) | null, + * onProviderResponse?: (() => void) | null, * closedPromise?: Promise, * rejectClosed?: (reason: Error) => void * }} */ @@ -410,7 +418,7 @@ class ZcodeConnection { z.changed?.(); try { const ack = await sendText(this, commandId, message.text, "guide"); - if (ack.status !== "accepted") throw new Error(refused(ack, "steer")); + if (!["accepted", "duplicate"].includes(ack.status)) throw new Error(refused(ack, "steer")); // Accepted while idle, ZCode starts it as its own turn, and this run waits for it. return { turnId, unconfirmed: true }; } catch (error) { @@ -426,7 +434,8 @@ class ZcodeConnection { const client = this.client; const z = this.session; z.cancelRequested = true; - const ack = await command(this, z.sessionId, "stop", {}); + z.changed?.(); + const ack = await command(this, z.sessionId, "stop", {}, undefined, { timeoutMs: CONTROL_MS }); if (!["accepted", "noop", "duplicate"].includes(ack.status)) throw new Error(refused(ack, "stop")); z.changed?.(); } @@ -449,6 +458,7 @@ class ZcodeConnection { const ours = end => end.inputId === commandId; const settle = () => { clearTimeout(settleTimer); settleTimer = null; }; const check = () => { + if (z.cancelRequested) duplicateStarted?.(false); if (!turnId) { const started = [...z.running].find(([, inputId]) => inputId === commandId)?.[0] ?? z.ended.slice(first).find(ours)?.turnId; if (started) { turnId = started; options.onTurnId(started); } @@ -485,17 +495,22 @@ class ZcodeConnection { droppedSteers: [...dropped].map(([input, reason]) => ({ id: baseId(input), reason })) }); z.cancelRequested = false; + z.onProviderResponse = options.onProviderResponse; z.changed = check; try { for (let resend = 0; ; resend += 1) { // Each discarded answer is immediate, and suffixes an earlier process used // are unknown here, so the bound is generous. + if (z.cancelRequested || options.isCancelled?.()) return { turnId, turnStatus: "interrupted", finalMessage: "", ...receipts() }; z.outstanding.add(commandId); const ack = await sendText(this, commandId, options.prompt, "startNow"); if (ack.status === "accepted") break; z.outstanding.delete(commandId); // A duplicate may still start: adopt only that id, as soon as its session event arrives. if (ack.status === "duplicate") { + // The worker must be able to stop a duplicate wait even before a turn id arrives. + z.pending = true; + options.onPending?.(); let timer; let started; try { @@ -509,8 +524,9 @@ class ZcodeConnection { duplicateStarted = null; } if (started) break; + if (z.cancelRequested) return { turnId, turnStatus: "interrupted", finalMessage: "", ...receipts() }; return { turnId: null, turnStatus: "failed", finalMessage: "", teardown: true, - error: { message: `ZCode already holds this message (duplicate id) and did not start it within ${Number((duplicateWaitMs / 1000).toFixed(1))} s; it was not sent again, so it can't run twice.` }, ...receipts() }; + error: { message: `ZCode already holds this message (duplicate id) but relay could not confirm its start within ${Number((duplicateWaitMs / 1000).toFixed(1))} s; it was not sent again, so it can't run twice.` }, ...receipts() }; } // Only positive evidence that the old input was discarded permits a fresh id. const discarded = ack.status === "failed" && ack.reasonCode === DISCARDED_ON_RESTART; @@ -532,6 +548,7 @@ class ZcodeConnection { z.outstanding.clear(); z.pending = false; z.changed = null; + z.onProviderResponse = null; } } } @@ -551,7 +568,7 @@ export async function connect(request, hooks) { onServerRequest(message, transport) { answerServerRequest(request, message, transport); } }); try { await client.initialize(); } - catch (error) { await client.close(); throw new Error(client.exitInfo ? zcodeEndMessage(client.exitInfo) : protocolMessage(error)); } + catch (error) { await client.close(); if (error.name === "TransportTimeout") throw error; throw new Error(client.exitInfo ? zcodeEndMessage(client.exitInfo) : protocolMessage(error)); } return new ZcodeConnection(client, request); } // Where `zcode app-server` finds its sign-in and default model, read without a model call and diff --git a/plugins/relay/scripts/lib/worker/admission.mjs b/plugins/relay/scripts/lib/worker/admission.mjs index 46c78c51..e65f6275 100644 --- a/plugins/relay/scripts/lib/worker/admission.mjs +++ b/plugins/relay/scripts/lib/worker/admission.mjs @@ -690,7 +690,9 @@ function applyRefusal(old, refusal, jobId, now, backend, account) { const keepOld = Number.isFinite(kept) && kept > fresh; const until = keepOld ? kept : fresh; const resetSource = keepOld && !old.damaged ? old.resetSource ?? "default" : provider ? "provider" : "default"; - return { v: 1, backend, account, ...refusal, since: same ? old.since : iso(now), updatedAt: iso(now), until: iso(until), strikes, resetSource, lastJobId: jobId, probe: null }; + // pauseRead already discarded a dead owner. A refusal never releases a live probe; + // its owner keeps the same generation through inner retries until evidence or release. + return { v: 1, backend, account, ...refusal, since: same ? old.since : iso(now), updatedAt: iso(now), until: iso(until), strikes, resetSource, lastJobId: jobId, probe: old?.probe ?? null }; } function foldPending(home, backend, account, deps) { const dir = path.join(home, "pauses", "pending"), now = deps.now(); @@ -751,6 +753,9 @@ export function probeSucceeded(home, backend, jobId, injected = {}, generation = if (!admissionEnabled(deps.env, undefined, deps.platform)) return false; const account = jobAccount(home, jobId, deps); return retryLock(home, deps, () => { + // Apply older refusals before the answer, so a deferred retry cannot resurrect the + // pause after this fenced owner succeeds. Damaged pending records still hold. + if (foldPending(home, backend, account, deps)) return false; const p = pauseRead(home, backend, deps, false, true, account); if (!p?.probe || p.probe.jobId !== jobId || !generation || p.probe.generation !== generation) return false; fs.rmSync(pauseFile(home, backend, account), { force: true }); return true; diff --git a/plugins/relay/scripts/lib/worker/liveness.mjs b/plugins/relay/scripts/lib/worker/liveness.mjs new file mode 100644 index 00000000..1dc6c0bb --- /dev/null +++ b/plugins/relay/scripts/lib/worker/liveness.mjs @@ -0,0 +1,41 @@ +// A silent turn is allowed to run indefinitely while its backend answers requests. +// The first ping discovers support on this connection; failure disables only this watchdog. +import { TransportTimeout } from "../backends/transport/stdio-rpc.mjs"; +export const STALLED_AFTER_MS = 10 * 60 * 1000; +export const PING_INTERVAL_MS = 5 * 60 * 1000; + +export function backendLiveness(connection, { now = Date.now, note = (_info) => {} } = {}) { + let enabled = false, stopped = false, pending = false, misses = 0, lastPing = -Infinity; + let lastNotification = now(); + const method = connection.pingMethod; + const ping = async () => { + try { await connection.ping(); return true; } + catch (error) { + // An RPC error is an answer too (ACP normally answers -32601). + if (Number.isInteger(error.rpcCode)) return true; + return false; + } + }; + return { + touch() { lastNotification = now(); }, + stop() { stopped = true; }, + async open() { + if (!connection.ping) return; + enabled = await ping(); + if (!stopped) note({ enabled, method, state: enabled ? "backend answering" : "watchdog disabled", reason: enabled ? null : "open-time ping was not answered; control deadlines remain active" }); + }, + async tick(active) { + if (!active || !enabled || stopped || pending || now() - lastNotification < STALLED_AFTER_MS || now() - lastPing < PING_INTERVAL_MS) return; + pending = true; + lastPing = now(); + note({ enabled, method, state: "checking the backend" }); + try { + const answered = await ping(); + if (stopped) return; + misses = answered ? 0 : misses + 1; + note({ enabled, method, state: answered ? "backend answering" : "checking the backend", misses }); + if (misses >= 2) connection.server.markWedged(new TransportTimeout(connection.request.backend, method, 30000)); + } finally { pending = false; } + } + }; +} diff --git a/plugins/relay/scripts/lib/worker/worker.mjs b/plugins/relay/scripts/lib/worker/worker.mjs index 4b75716a..e26fd6af 100644 --- a/plugins/relay/scripts/lib/worker/worker.mjs +++ b/plugins/relay/scripts/lib/worker/worker.mjs @@ -1,3 +1,4 @@ +import { backendLiveness } from "./liveness.mjs"; // The job worker. One worker per job at a time, holding the job's lease. It starts its // backend's server process, opens the session, runs turns, acts on inbox messages, and is // the only writer of job.json while it lives. It stops the server when the job ends. @@ -241,7 +242,7 @@ async function runWorkerRound(jobDir, options, held) { job = { ...job, jobDir, execution, executions: [...previousExecutions, execution].slice(-20), launches, launch: readJsonOrNull(executionEnv.RELAY_WORKER_LAUNCH ?? "") ?? job.launch }; const backend = backendFor(job.request); - const admissionDeps = { env: executionEnv, platform: options.platform ?? process.platform, ...options.admission }; + const admissionDeps = { request: job.request, env: executionEnv, platform: options.platform ?? process.platform, ...options.admission }; const home = admissionHome(admissionDeps.env); let slot = null; const guardOn = admissionEnabled(admissionDeps.env, undefined, admissionDeps.platform ?? process.platform); @@ -269,6 +270,7 @@ async function runWorkerRound(jobDir, options, held) { const cancelMessages = []; /** @type {import("../backends/types").Connection | null} */ let connection = null; + let liveness = null; let sessionId = job.sessionId ?? null; let activeTurnId = null; let turnActive = false; @@ -504,7 +506,7 @@ async function runWorkerRound(jobDir, options, held) { }; async function serviceTurnControls() { - if (!connection || !sessionId || !turnActive || !connection.controlsReady(activeTurnId)) { + if (!connection || connection.server.wedged || !sessionId || !turnActive || !connection.controlsReady(activeTurnId)) { return; } if (cancelRequested && !interruptSent) { @@ -569,6 +571,12 @@ async function runWorkerRound(jobDir, options, held) { slot = { ...slot, probe: false, probeGeneration: null, probeAt: null }; } }; + const noteRefusal = refusal => { + if (!guardOn || !refusal) return; + const pause = recordRefusal(home, backend.name, refusal, job.id, admissionDeps); + save({ refusal: { backend: backend.name, kind: refusal.kind, code: refusal.code, + message: refusal.message, at: nowIso(), pausedUntil: pause?.until ?? null } }); + }; const sample = (fresh = false) => { if (!["linux", "darwin"].includes(options.platform ?? process.platform)) return; try { @@ -583,11 +591,7 @@ async function runWorkerRound(jobDir, options, held) { }; const heartbeat = setInterval(() => { sample(); - if (slot?.probeAt && (admissionDeps.now ?? Date.now)() - Date.parse(slot.probeAt) >= DEFAULTS.probeMs) { - if (activeTurnId) completeProbe(); - else if (releaseProbe(home, backend.name, job.id, admissionDeps, slot.probeGeneration)) - slot = { ...slot, probe: false, probeGeneration: null, probeAt: null }; - } + void liveness?.tick(turnActive).catch(() => {}); }, config.heartbeatMs); heartbeat.unref?.(); const poller = setInterval(() => { @@ -603,7 +607,7 @@ async function runWorkerRound(jobDir, options, held) { const request = job.request; const writeEvent = notification => fs.appendFileSync(paths.events, `${JSON.stringify({ at: lastEventAt, ...notification })}\n`, "utf8"); - const touchEvents = () => { lastEventAt = nowIso(); }; + const touchEvents = () => { lastEventAt = nowIso(); liveness?.touch(); }; const eventRecorder = backend.eventRecorder(writeEvent, touchEvents); const connect = async () => { const attemptIndex = job.attempts.length - 1; @@ -633,7 +637,7 @@ async function runWorkerRound(jobDir, options, held) { }); }; - let granted = false; + let grantedAt = null; let admitted = wasAdmitted(job); const waitForAdmission = async (turn = false) => { for (;;) { @@ -646,7 +650,7 @@ async function runWorkerRound(jobDir, options, held) { throw Object.assign(new Error(decision.reason), { admissionPermanent: true }); } if (decision.admit) { - granted = true; + grantedAt = (admissionDeps.now ?? Date.now)(); admitted = true; touchEvents(); save({ state: "starting", firstAdmittedAt: job.firstAdmittedAt ?? decision.record.admittedAt, admission: decision.record }); @@ -693,6 +697,15 @@ async function runWorkerRound(jobDir, options, held) { // Whether a turn ever ran on this thread, so a backend can tell a lost draft from lost work. sessionUsed: Boolean(sessionId && job.attempts.some((attempt) => attempt.turnId)) }); + liveness?.stop(); + liveness = backendLiveness(connection, { + now: options.now ?? Date.now, + note(info) { + save({ backendLiveness: info }); + writeEvent({ method: "relay/backendLiveness", params: info }); + } + }); + await liveness.open(); sessionId = opened.sessionId; save({ sessionId, serviceTier: opened.serviceTier ?? null }); // What the backend had to decide while opening is a `note:` in the state block, once. @@ -745,13 +758,19 @@ async function runWorkerRound(jobDir, options, held) { result = { ...result, turnStatus: "interrupted" }; break; } - if (!granted && !(await waitForAdmission(true))) { + if ((grantedAt === null || (admissionDeps.now ?? Date.now)() - Number(grantedAt) > 10000) && !(await waitForAdmission(true))) { result = { ...result, turnStatus: "interrupted" }; break; } - granted = false; + grantedAt = null; try { - result = await connection.run({ + let providerAnswered = false; + const running = connection.run({ + isCancelled: () => cancelRequested, + onProviderResponse() { + if (!providerAnswered) { providerAnswered = true; completeProbe(); } + }, + onProviderLimit(error) { noteRefusal(classifyRefusal(backend.name, error)); }, prompt: dispatch.prompt, onPending() { save({ state: "running" }); void serviceTurnControls(); }, clientId: dispatch.clientId, @@ -766,14 +785,18 @@ async function runWorkerRound(jobDir, options, held) { void serviceTurnControls(); } }); + result = await Promise.race([running, ...(connection.server.wedgedPromise ? [connection.server.wedgedPromise.then(error => { throw error; })] : [])]); + if (connection.server.wedged) throw connection.server.wedged; } catch (error) { + if (connection.server.wedged) throw connection.server.wedged; result = { turnStatus: "failed", error: { ...error.data, message: error.message, data: error.data }, finalMessage: "" }; } refusal = guardOn && result.turnStatus === "failed" ? classifyRefusal(backend.name, result.error) : null; - if (refusal) { - const pause = recordRefusal(home, backend.name, refusal, job.id, admissionDeps); - save({ refusal: { backend: backend.name, kind: refusal.kind, code: refusal.code, - message: refusal.message, at: nowIso(), pausedUntil: pause?.until ?? null } }); + if (refusal) noteRefusal(refusal); + if (slot?.probe) { + const releasedSlot = slot; + releaseProbe(home, backend.name, job.id, admissionDeps, slot.probeGeneration); + slot = { ...releasedSlot, probe: false, probeGeneration: null, probeAt: null }; } // A backend that retries model calls itself marks its failures `retryable: false`. // Commentary does not stop a retry; work done decides how the retry is sent. @@ -822,7 +845,6 @@ async function runWorkerRound(jobDir, options, held) { let state; let errorMessage = null; if (result.turnStatus === "completed") { - completeProbe(); state = "completed"; } else if (result.turnStatus === "interrupted" && cancelRequested) { state = "cancelled"; @@ -876,8 +898,15 @@ async function runWorkerRound(jobDir, options, held) { break; } catch (error) { - granted = false; + grantedAt = null; if (error.admissionPermanent) throw error; + const wedged = connection?.server.wedged; + if (wedged) { + liveness?.stop(); + turnActive = false; + await connection.close(); + await settleInflight(); + } accept(claimInboxMessages(jobDir)); // The process exited and the connection is still draining its pipes: wait for the end // record, which the connection completes within its drain time. @@ -886,7 +915,7 @@ async function runWorkerRound(jobDir, options, held) { // connect can fail before it returns a connection. Its onEnd record still exists. const end = connection?.server.exitInfo ?? job.attempts.at(-1).servers?.at(-1); if (!end?.endedAt) throw error; - const transportErrors = [...(job.attempts.at(-1).transportErrors ?? []), backend.endMessage(end)]; + const transportErrors = [...(job.attempts.at(-1).transportErrors ?? []), wedged?.message ?? (error.name === "TransportTimeout" ? error.message : backend.endMessage(end))]; updateAttempt({ transportErrors }); if (!sessionId || transportErrors.length > 1 || recoveryForEpoch(job)) throw new Error(transportErrors.join("\n")); const recovery = { id: newMessageId(), epoch: job.recoveryEpoch ?? "start", at: nowIso(), @@ -895,6 +924,7 @@ async function runWorkerRound(jobDir, options, held) { updateAttempt({ state: "starting" }); await connection?.close(); connection = null; + liveness?.stop(); const sweep = await stopRecordedAppServers(job, { jobDir, ...options.predecessorCleanup }); if (sweep.survivors.length || sweep.cause) { sweepFailed = true; @@ -943,6 +973,7 @@ async function runWorkerRound(jobDir, options, held) { } finally { clearInterval(heartbeat); clearInterval(poller); + liveness?.stop(); // The connection reads the server's output to its end as it closes. A turn can end on the // backend's record before the worker has read the updates the server sent ahead of it, so // the events are flushed only once the connection has closed. @@ -960,6 +991,7 @@ async function runWorkerRound(jobDir, options, held) { } eventRecorder?.flush(); } finally { + if (slot?.probe) releaseProbe(home, backend.name, job.id, admissionDeps, slot.probeGeneration); if (slot && !sweepFailed) releaseSlot(home, job.id, me, admissionDeps); slot = null; } diff --git a/plugins/relay/skills/jobs/internals.md b/plugins/relay/skills/jobs/internals.md index 8cf3846c..a5151a22 100644 --- a/plugins/relay/skills/jobs/internals.md +++ b/plugins/relay/skills/jobs/internals.md @@ -69,7 +69,7 @@ Backend loss and job completion stop owned processes with TERM/CONT, then KILL, slot release. An unresolved survivor prevents reconnect; at job end it preserves the terminal state, records `cleanup: unresolved`, and keeps its seat. The backend close itself sends TERM after 50 ms, KILL at 5 seconds and returns within 10 seconds, recording `closeTimedOut` -if process end remains unobserved. Control RPC and turn deadlines are unchanged. +if process end remains unobserved. Control-request deadlines and the silent-turn watchdog are described below. Heartbeats write `treeBytes` to the job and running slot: Linux scope `memory.current`, otherwise summed member RSS. Active-job text shows this live measurement; terminal result text omits it so @@ -131,3 +131,25 @@ saved for the next admission check. Future startup times remain reserved across Only a currently held seat can be reused; earlier completed admissions do not reserve a place. Memory later used by tools/tests, native Claude agents, other harnesses and other users is outside admission control. Recognized provider pauses and recovery are described in [orchestration.md](orchestration.md). + + +### Backend liveness and provider evidence + +Relay gives a backend 2 minutes to start, 5 minutes to open or configure a session, and 30 seconds +to take a steer or a cancel request. ACP cancel is a notification without a reply. A turn has no time +limit. After 10 minutes without notifications, relay pings every 5 minutes with a 30-second deadline; +two consecutive unanswered pings use the normal one-restart server recovery path. Responsive long +builds and sleeps keep running. An open-time ping tests this capability on each connection; no answer +disables the watchdog for that connection with an event note, keeping control deadlines. +`backendLiveness` in job JSON and `relay/backendLiveness` events show `backend answering`, +`checking the backend`, or `watchdog disabled`, and name the ping method. + +After a pause, one job probes the provider; the pause ends only when that job gets an actual answer +from the provider, not when it merely starts. A live probe retains its claim through slow startup, +silent turns, and provider retries. A turn ending without model evidence releases the claim without +clearing the pause. +A start grant older than 10 seconds is checked again before sending. Codex and Grok retry notifications +and Kimi's `turn.step.retrying` record pause new starts immediately while the active attempt retries. +ZCode exposes no runtime retry signal, so its coverage is terminal failures only. Pauses follow the +backend account home. Grok retry classification has not been verified with a live exhausted account; +Grok's web search cannot be disabled. diff --git a/tests/admission-pauses.test.mjs b/tests/admission-pauses.test.mjs index f2953176..db315731 100644 --- a/tests/admission-pauses.test.mjs +++ b/tests/admission-pauses.test.mjs @@ -352,3 +352,35 @@ test("guard status identifies each paused account", t => { const lines = renderGuard({ enabled: true, availableBytes: null, pauses: listPauses(s.env, s.deps) }); assert.ok(lines.some(line => line.includes(`kimi (account ${account}) paused:`))); }); + +for (const pending of [false, true]) test(`a retry refusal preserves a live probe's owner and generation (${pending ? 'pending replay' : 'direct'})`, t => { + const s = setup(t); + recordRefusal(s.root, "codex", { kind: "rate" }, "previous", s.deps); s.tick(D.ratePauseMs); + const owner = s.ctx(1), first = admit(owner, s.deps), generation = first.slot.probeGeneration; + const file = path.join(s.root, "admission.lock"); + if (pending) acquireOwnership(file, { pid: process.pid }); + try { + const refusal = recordRefusal(s.root, "codex", { kind: "rate" }, owner.jobId, { ...s.deps, lockSleep() {} }); + if (pending) assert.deepEqual(refusal, { pending: true }); + } finally { if (pending) releaseOwnership(file); } + s.tick(D.maxRatePauseMs + D.probeMs); + assert.equal(admit(s.ctx(2), s.deps).waitKind, "probe", "no second probe while the retrying owner lives"); + const pause = JSON.parse(fs.readFileSync(path.join(s.root, "pauses", "codex.json"))); + assert.equal(pause.probe.jobId, owner.jobId); assert.equal(pause.probe.generation, generation); + assert.equal(probeSucceeded(s.root, "codex", "job-2", s.deps, generation), false); + assert.equal(probeSucceeded(s.root, "codex", owner.jobId, s.deps, "stale-generation"), false); + assert.equal(probeSucceeded(s.root, "codex", owner.jobId, s.deps, generation), true); + assert.equal(listPauses(s.env, s.deps).length, 0); +}); + +test("a retry refusal preserves recovery when its pending record is folded by probe success", t => { + const s = setup(t); + recordRefusal(s.root, "codex", { kind: "rate" }, "previous", s.deps); s.tick(D.ratePauseMs); + const owner = s.ctx(1), first = admit(owner, s.deps), file = path.join(s.root, "admission.lock"); + acquireOwnership(file, { pid: process.pid }); + try { assert.deepEqual(recordRefusal(s.root, "codex", { kind: "rate" }, owner.jobId, { ...s.deps, lockSleep() {} }), { pending: true }); } + finally { releaseOwnership(file); } + assert.equal(probeSucceeded(s.root, "codex", owner.jobId, s.deps, first.slot.probeGeneration), true); + assert.deepEqual(fs.readdirSync(path.join(s.root, "pauses", "pending")), []); + s.tick(D.spacingMs); assert.equal(admit(s.ctx(2), s.deps).admit, true, "an older pending retry must not resurrect the recovered pause"); +}); diff --git a/tests/admission-worker.test.mjs b/tests/admission-worker.test.mjs index c25c92b9..d633a1ad 100644 --- a/tests/admission-worker.test.mjs +++ b/tests/admission-worker.test.mjs @@ -268,7 +268,7 @@ test("dormant workerless jobs stop deferring the repository review gate after fi assert.equal(findActiveJob(s.ctx.work, now + 15 * 60000 + 1).id, job.id, "an adopter makes it active again"); }); -test("a heartbeat during slow probe startup releases the claim without provider success", async t => { +test("a heartbeat during slow probe startup retains the claim until provider evidence", async t => { const { recordRefusal, listPauses } = await import("../plugins/relay/scripts/lib/worker/admission.mjs"); const s = seed(t); s.unblock(); s.options.env.RELAY_JOBS_HEARTBEAT_MS = "10"; @@ -279,14 +279,16 @@ test("a heartbeat during slow probe startup releases the claim without provider let paused; t.mock.method(backend, "connect", async (...args) => { assert.ok(listPauses(s.ctx.env, deps)[0].probing); - s.tick(D.probeMs); - await waitFor(() => { paused = JSON.parse(fs.readFileSync(path.join(home, "pauses", "codex.json"))); return paused && !paused.probe; }); + s.tick(D.probeMs + 10000); + await new Promise(resolve => setTimeout(resolve, 30)); + paused = JSON.parse(fs.readFileSync(path.join(home, "pauses", "codex.json"))); + assert.ok(paused.probe); assert.equal(paused.strikes, 1); return connect(...args); }); await runWorker(s.dir, s.options); assert.equal(readJob(s.dir).state, "completed"); - assert.equal(listPauses(s.ctx.env, deps)[0].strikes, 1, "completion of a released generation cannot clear the pause"); + assert.equal(listPauses(s.ctx.env, deps).length, 0, "actual model output clears the probe"); }); test("server-death replacement waits for critical states, scope, provider pause and pacing before reconnecting", async t => { @@ -501,3 +503,290 @@ test("a reparented worker acknowledges even after its caller has gone", async t assert.ok(ack.callerGoneAt); assert.equal(readJob(s.dir).state, "completed"); }); + +test("a silent probe retains its fence beyond two minutes and clears only on provider evidence", async t => { + const { recordRefusal, listPauses, admit } = await import("../plugins/relay/scripts/lib/worker/admission.mjs"); + const s = seed(t); s.unblock(); + s.options.env.RELAY_JOBS_HEARTBEAT_MS = "10"; + const home = s.ctx.env.RELAY_ADMISSION_HOME, deps = { ...s.options.admission, env: s.ctx.env }; + recordRefusal(home, "codex", { kind: "rate" }, "previous", deps); s.tick(D.ratePauseMs); + const backend = backendNamed("codex"), connect = backend.connect; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args); + t.mock.method(connection, "run", async turn => { + turn.onPending(); turn.onTurnId("silent-turn"); s.tick(D.probeMs + 10000); + await new Promise(resolve => setTimeout(resolve, 30)); + assert.ok(listPauses(s.ctx.env, deps)[0].probing); + const decision = admit({ home, jobId: "other-job", request: readJob(s.dir).request, + me: { pid: process.pid, startTime: readJob(s.dir).worker.startTime } }, deps); + assert.equal(decision.waitKind, "probe"); + turn.onProviderResponse(); + assert.equal(listPauses(s.ctx.env, deps).length, 0); + return { turnStatus: "completed", finalMessage: "Answer" }; + }); + return connection; + }); + await runWorker(s.dir, s.options); assert.equal(readJob(s.dir).state, "completed"); +}); + +test("a slow connection rechecks a grant before sending into a newly recorded provider pause", async t => { + const { recordRefusal } = await import("../plugins/relay/scripts/lib/worker/admission.mjs"); + const s = seed(t); s.unblock(); + const home = s.ctx.env.RELAY_ADMISSION_HOME, deps = { ...s.options.admission, env: s.ctx.env }; + const backend = backendNamed("codex"), connect = backend.connect; + let held = false; + s.options.admissionSleep = async () => { + assert.equal(readJob(s.dir).admission.waitKind, "pause"); held = true; + fs.rmSync(path.join(home, "pauses", "codex.json")); + }; + t.mock.method(backend, "connect", async (...args) => { + s.tick(15000); recordRefusal(home, "codex", { kind: "usage" }, "other-job", deps); + const connection = await connect(...args), run = connection.run.bind(connection); + t.mock.method(connection, "run", (...turn) => { assert.equal(held, true); return run(...turn); }); + return connection; + }); + await runWorker(s.dir, s.options); assert.equal(readJob(s.dir).state, "completed"); +}); + +test("a completed local turn without provider evidence releases its probe and preserves the pause", async t => { + const { recordRefusal, listPauses } = await import("../plugins/relay/scripts/lib/worker/admission.mjs"); + const s = seed(t); s.unblock(); + const home = s.ctx.env.RELAY_ADMISSION_HOME, deps = { ...s.options.admission, env: s.ctx.env }; + recordRefusal(home, "codex", { kind: "rate" }, "previous", deps); s.tick(D.ratePauseMs); + const backend = backendNamed("codex"), connect = backend.connect; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args); + t.mock.method(connection, "run", async () => ({ turnStatus: "completed", finalMessage: "" })); + return connection; + }); + await runWorker(s.dir, s.options); + const pause = listPauses(s.ctx.env, deps)[0]; + assert.equal(readJob(s.dir).state, "completed"); assert.equal(pause.strikes, 1); assert.equal(pause.probing, false); +}); + +for (const [name, error] of [["codex", { codexErrorInfo: { usageLimitExceeded: {} }, message: "Quota" }], + ["grok", { message: "HTTP 429 Too Many Requests" }], ["kimi", { message: "403 You have reached your weekly usage limit" }]]) { + test(`${name} early retry signal pauses new work before the current turn completes`, async t => { + const { admit } = await import("../plugins/relay/scripts/lib/worker/admission.mjs"); + const s = seed(t); s.unblock(); + if (name !== "codex") { + const { installFakeGrok } = await import("./fakes/grok.mjs"); + const { installFakeKimi } = await import("./fakes/kimi.mjs"); + const fake = name === "grok" ? installFakeGrok(s.ctx.bin) : installFakeKimi(s.ctx.bin); + s.ctx.env[`${name.toUpperCase()}_BINARY`] = fake.binary; + const request = buildLaunchRequest({ backend: name, model: "default", effort: "default", cd: s.ctx.work }, { baseDir: s.ctx.work, env: s.ctx.env }); + writeJob(s.dir, { ...readJob(s.dir), request }); + } + const backend = backendNamed(name), connect = backend.connect; + let signalled = false; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args); + t.mock.method(connection, "run", async turn => { + turn.onPending(); turn.onTurnId("still-running"); turn.onProviderLimit(error); signalled = true; + const decision = admit({ home: s.ctx.env.RELAY_ADMISSION_HOME, jobId: "queued-job", request: readJob(s.dir).request, + me: readJob(s.dir).worker }, { ...s.options.admission, env: s.ctx.env }); + assert.equal(decision.waitKind, "pause"); assert.equal(readJob(s.dir).state, "running"); + return { turnStatus: "completed", finalMessage: "Eventually answered" }; + }); + return connection; + }); + await runWorker(s.dir, s.options); + assert.equal(signalled, true); assert.equal(readJob(s.dir).state, "completed"); + assert.equal(readJob(s.dir).refusal.kind, name === "grok" ? "rate" : "usage"); + }); +} + +test("two unanswered watchdog pings bound a stubborn server's close and recover exactly once", async t => { + const { writeExecutable } = await import("./support/helpers.mjs"); + const { installFakeCodex } = await import("./fakes/codex.mjs"); + const s = seed(t); s.unblock(); + // A real RPC peer: its first ping answers, later pings and the turn stay silent. EOF + // and TERM leave it running, so recovery must wait for the transport's bounded KILL. + writeExecutable(path.join(s.ctx.bin, "codex"), `#!/usr/bin/env node +const readline = require('node:readline'); +process.on('SIGTERM', () => {}); setInterval(() => {}, 1000); +let pings = 0; +readline.createInterface({ input: process.stdin }).on('line', line => { + const m = JSON.parse(line); if (m.id == null) return; + const reply = result => process.stdout.write(JSON.stringify({ id: m.id, result }) + '\\n'); + if (m.method === 'account/read') { if (++pings === 1) reply({}); return; } + if (m.method === 'turn/start') { reply({ turn: { id: 'silent', status: 'inProgress', items: [] } }); return; } + if (m.method === 'thread/start' || m.method === 'thread/resume') { reply({ thread: { id: 'thr_silent' } }); return; } + reply({}); +}); +`); + const setTimer = globalThis.setTimeout; + let pingDeadlines = 0; + let deadlineTimer; + let now = 0, connections = 0, closeMs; + s.options.now = () => now; s.options.env.RELAY_JOBS_HEARTBEAT_MS = "10"; + const backend = backendNamed("codex"), connect = backend.connect; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args); connections++; + if (connections === 1) { + const run = connection.run.bind(connection), close = connection.close.bind(connection); + t.mock.method(connection, "run", async turn => { + assert.equal(readJob(s.dir).backendLiveness.enabled, true, "the real open-time ping answered"); + deadlineTimer = t.mock.method(globalThis, "setTimeout", (callback, ms, ...args) => { + if (ms === 30000) { pingDeadlines++; return setTimer(callback, 20, ...args); } + return setTimer(callback, ms, ...args); + }); + const running = run(turn); now = 10 * 60000; + await waitFor(() => readJob(s.dir).backendLiveness.misses === 1); + now += 5 * 60000; + return running; + }); + t.mock.method(connection, "close", async () => { + if (closeMs != null) return close(); + const started = Date.now(); await close(); closeMs = Date.now() - started; + deadlineTimer.mock.restore(); + installFakeCodex(s.ctx.bin, "job", { turnMs: 30 }); + }); + } + return connection; + }); + await runWorker(s.dir, s.options); + const job = readJob(s.dir); + assert.equal(job.state, "completed"); assert.equal(connections, 2); + assert.equal(job.automaticResumes.length, 1); assert.equal(job.automaticResumes[0].reason, "app-server"); + assert.match(job.attempts[0].transportErrors[0], /codex stopped answering \(account\/read, 30 s\)/); + assert.equal(job.attempts[0].servers[0].signal, "SIGKILL"); + assert.ok(closeMs >= 4500 && closeMs < 12000, `bounded close took ${closeMs} ms`); + assert.equal(pingDeadlines, 2, "both missed pings use the real 30-second request deadline"); +}); + +test("initialize with no reply fails the job at its control deadline", async t => { + const { writeExecutable } = await import("./support/helpers.mjs"); + const s = seed(t); s.unblock(); + writeExecutable(path.join(s.ctx.bin, "codex"), '#!/usr/bin/env node\nprocess.stdin.resume();\n'); + const setTimer = globalThis.setTimeout; + let deadlineSeen = false; + t.mock.method(globalThis, "setTimeout", (callback, ms, ...args) => { + if (ms === 120000) { deadlineSeen = true; return setTimer(callback, 10, ...args); } + return setTimer(callback, ms, ...args); + }); + await runWorker(s.dir, s.options); + const job = readJob(s.dir); + assert.equal(deadlineSeen, true); assert.equal(job.state, "failed"); + assert.match(job.errorMessage, /codex stopped answering \(initialize, 120 s\)/); + assert.ok(job.attempts[0].servers[0].endedAt); +}); + +test("an unanswered open-time ping disables the watchdog and the job still completes", async t => { + const { TransportTimeout } = await import("../plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs"); + const s = seed(t); s.unblock(); + const backend = backendNamed("codex"), connect = backend.connect; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args); + t.mock.method(connection, "ping", async () => { throw new TransportTimeout("codex", "account/read", 30000); }); + return connection; + }); + await runWorker(s.dir, s.options); + assert.equal(readJob(s.dir).state, "completed"); + assert.equal(readJob(s.dir).backendLiveness.enabled, false); + assert.match(fs.readFileSync(jobPaths(s.dir).events, "utf8"), /watchdog disabled/); +}); + +test("a steer that never answers is queued with recovery evidence and cannot hang settleInflight", async t => { + const s = seed(t, "[held] Wait for the steer."); s.unblock(); + const backend = backendNamed("codex"), connect = backend.connect; + const setTimer = globalThis.setTimeout; + t.mock.method(globalThis, "setTimeout", (callback, ms, ...args) => setTimer(callback, ms === 30000 ? 20 : ms, ...args)); + let connections = 0, steer; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args); connections++; + if (connections === 1) { + const send = connection.client.sendMessage.bind(connection.client); + t.mock.method(connection.client, "sendMessage", message => { if (message.method !== "turn/steer") send(message); }); + const run = connection.run.bind(connection); + t.mock.method(connection, "run", turn => { + steer = postInboxMessage(s.dir, { kind: "steer", text: "[quick] Next turn." }); + return run(turn); + }); + } + return connection; + }); + await runWorker(s.dir, s.options); + const job = readJob(s.dir); + assert.equal(job.state, "completed"); assert.equal(connections, 2); + assert.ok(job.messages.some(m => m.id === steer.id && m.action === "queued-next-turn" && /stopped answering/.test(m.reason))); + assert.ok(["turn-started", "delivered-before"].includes(readOutcome(s.dir, steer.id).action)); + assert.equal(job.automaticResumes.length, 1); +}); + +test("a worker leaves a silent 25-minute sleep alive while the backend answers pings", async t => { + const s = seed(t); s.unblock(); + let now = 0, pings = 0, turns = 0, finished = false; + s.options.now = () => now; s.options.env.RELAY_JOBS_HEARTBEAT_MS = "10"; + const backend = backendNamed("codex"), connect = backend.connect; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args), close = connection.close.bind(connection); + t.mock.method(connection, "ping", async () => { pings++; }); + t.mock.method(connection, "close", async () => { assert.equal(finished, true); await close(); }); + t.mock.method(connection, "run", async turn => { + turns++; turn.onPending(); turn.onTurnId("long-sleep"); + for (let minute = 1; minute <= 25; minute++) { + now = minute * 60000; + await new Promise(resolve => setTimeout(resolve, 25)); + assert.equal(connection.server.wedged, null); + } + finished = true; + return { turnStatus: "completed", finalMessage: "Sleep finished" }; + }); + return connection; + }); + await runWorker(s.dir, s.options); + assert.equal(readJob(s.dir).state, "completed"); assert.equal(turns, 1); assert.equal(pings, 5); + assert.equal(readJob(s.dir).automaticResumes?.length ?? 0, 0); +}); + + +test("an inbox cancel claimed during ZCode's initial acknowledgement prevents discarded-input resend", async t => { + const { installFakeZcode } = await import("./fakes/zcode.mjs"); + const s = seed(t); s.unblock(); s.ctx.env.ZCODE_BINARY = installFakeZcode(s.ctx.bin).binary; + writeJob(s.dir, { ...readJob(s.dir), request: buildLaunchRequest({ backend: "zcode", model: "default", effort: "default", cd: s.ctx.work }, { baseDir: s.ctx.work, env: s.ctx.env }) }); + const backend = backendNamed("zcode"), connect = backend.connect; + let resolveAck, params, sends = 0; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args), request = connection.client.request.bind(connection.client); + t.mock.method(connection.client, "request", (method, input, options) => { + if (method === "v4/command" && input.type === "sendText") { + sends++; params = input; + if (sends === 1) return new Promise(resolve => { resolveAck = resolve; }); + } + return request(method, input, options); + }); + return connection; + }); + const running = runWorker(s.dir, s.options); + await waitFor(() => resolveAck); + const cancel = postInboxMessage(s.dir, { kind: "cancel" }); + await waitFor(() => fs.existsSync(path.join(s.dir, "inbox", "claimed", `${cancel.id}.json`))); + resolveAck({ commandId: params.commandId, status: "failed", reasonCode: "fault.command.inputDiscardedOnRestart" }); + await running; + assert.equal(readJob(s.dir).state, "cancelled"); assert.equal(sends, 1); + assert.equal(readOutcome(s.dir, cancel.id).action, "cancelled"); +}); + +test("an in-turn retry preserves probe ownership until a later provider answer restores service", async t => { + const { recordRefusal, admit, listPauses } = await import("../plugins/relay/scripts/lib/worker/admission.mjs"); + const s = seed(t); s.unblock(); + const home = s.ctx.env.RELAY_ADMISSION_HOME, deps = { ...s.options.admission, env: s.ctx.env }; + recordRefusal(home, "codex", { kind: "rate" }, "previous", deps); s.tick(D.ratePauseMs); + const backend = backendNamed("codex"), connect = backend.connect; + t.mock.method(backend, "connect", async (...args) => { + const connection = await connect(...args); + t.mock.method(connection, "run", async turn => { + turn.onPending(); turn.onTurnId("probe-retrying"); + const file = path.join(home, "pauses", "codex.json"), before = JSON.parse(fs.readFileSync(file)).probe; + turn.onProviderLimit({ codexErrorInfo: { rateLimitExceeded: {} }, message: "Rate limit, retrying" }); + assert.deepEqual(JSON.parse(fs.readFileSync(file)).probe, before); + s.tick(D.maxRatePauseMs + D.probeMs); + assert.equal(admit({ home, jobId: "second-probe", request: readJob(s.dir).request, me: readJob(s.dir).worker }, deps).waitKind, "probe"); + turn.onProviderResponse(); assert.equal(listPauses(s.ctx.env, deps).length, 0); + return { turnStatus: "completed", finalMessage: "Recovered" }; + }); + return connection; + }); + await runWorker(s.dir, s.options); assert.equal(readJob(s.dir).state, "completed"); +}); diff --git a/tests/backend-liveness.test.mjs b/tests/backend-liveness.test.mjs new file mode 100644 index 00000000..e5b5db1b --- /dev/null +++ b/tests/backend-liveness.test.mjs @@ -0,0 +1,100 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { SpawnedServerClient, TransportTimeout } from "../plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs"; +import { backendLiveness, STALLED_AFTER_MS, PING_INTERVAL_MS } from "../plugins/relay/scripts/lib/worker/liveness.mjs"; +import { scriptedConnection } from "./support/backend-connection.mjs"; + +const flush = async () => { for (let i = 0; i < 20; i++) await Promise.resolve(); }; +function silentClient(backend = "codex") { + const client = new SpawnedServerClient("/tmp", { backendName: backend }); + client.proc = { stdin: { write() {} } }; + return client; +} + +for (const [method, ms] of [["initialize", 120000], ["runtime/capabilities", 120000], ["session/new", 300000], ["turn/steer", 30000]]) { + test(`${method} deadline rejects, forgets the pending call and marks the transport wedged`, async t => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + const client = silentClient(); + let failure; + const request = client.request(method, {}, { timeoutMs: ms }).catch(error => { failure = error; }); + t.mock.timers.tick(ms - 1); await flush(); assert.equal(failure, undefined); + t.mock.timers.tick(1); await request; + assert.ok(failure instanceof TransportTimeout); + assert.equal(failure.message, `codex stopped answering (${method}, ${ms / 1000} s)`); + assert.equal(client.pending.size, 0); assert.equal(client.wedged, failure); + assert.equal(await client.wedgedPromise, failure); + }); +} + +test("turn requests have no deadline and replies/exits clear request timers", async t => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + const client = silentClient(); + let done = false; + const turn = client.request("turn/start", {}).then(() => { done = true; }); + const control = client.request("turn/steer", {}, { timeoutMs: 30000 }); + client.handleLine(JSON.stringify({ id: 2, result: { turnId: "t" } })); + await control; + t.mock.timers.tick(25 * 60000); await flush(); + assert.equal(done, false); assert.equal(client.wedged, null); + client.handleLine(JSON.stringify({ id: 1, result: {} })); await turn; + const pending = client.request("session/load", {}, { timeoutMs: 300000 }); + client.handleExit(new Error("ended")); await assert.rejects(pending, /ended/); + t.mock.timers.tick(300000); assert.equal(client.wedged, null); +}); + +for (const backend of ["codex", "grok", "kimi", "zcode"]) test(`${backend} uses its read-only liveness method`, async () => { + const calls = []; + const client = { request: async (...args) => { + calls.push(args); + if (["grok", "kimi"].includes(backend)) throw Object.assign(new Error("Method not found"), { rpcCode: -32601 }); + return {}; + } }; + const connection = await scriptedConnection(backend, client, { cwd: "/tmp" }); + await connection.ping(); + const [method, params, options] = calls[0]; + assert.equal(method, backend === "codex" ? "account/read" : backend === "zcode" ? "runtime/capabilities" : "relay/ping"); + assert.deepEqual(params, backend === "codex" ? { refreshToken: false } : {}); + assert.deepEqual(options, { timeoutMs: 30000, wedgeOnTimeout: false }); +}); + +function watchdog() { + let now = 0, pings = 0, answering = true; + const notes = [], server = silentClient(); + const connection = { server, request: { backend: "codex" }, pingMethod: "account/read", async ping() { + pings++; + if (!answering) throw new TransportTimeout("codex", "account/read", 30000); + } }; + const watch = backendLiveness(connection, { now: () => now, note: value => notes.push(value) }); + return { watch, server, notes, pings: () => pings, advance: ms => now += ms, silent: () => answering = false, answer: () => answering = true }; +} + +test("a fake backend sleeping silently for 25 minutes while answering pings is never killed", async () => { + const s = watchdog(); await s.watch.open(); + for (let minute = 1; minute <= 25; minute++) { s.advance(60000); await s.watch.tick(true); } + assert.equal(s.server.wedged, null); + assert.equal(s.pings(), 5, "open, then minutes 10, 15, 20 and 25"); + assert.equal(s.notes.at(-1).state, "backend answering"); +}); + +test("two unanswered silent-turn pings wedge the backend; a reply resets the count", async () => { + const s = watchdog(); await s.watch.open(); s.silent(); + s.advance(STALLED_AFTER_MS); await s.watch.tick(true); assert.equal(s.server.wedged, null); + s.answer(); s.advance(PING_INTERVAL_MS); await s.watch.tick(true); + s.silent(); s.advance(PING_INTERVAL_MS); await s.watch.tick(true); assert.equal(s.server.wedged, null); + s.advance(PING_INTERVAL_MS); await s.watch.tick(true); assert.ok(s.server.wedged instanceof TransportTimeout); +}); + +test("an unanswered open-time ping disables only the watchdog", async () => { + const s = watchdog(); s.silent(); await s.watch.open(); + s.advance(25 * 60000); await s.watch.tick(true); + assert.equal(s.pings(), 1); assert.equal(s.server.wedged, null); + assert.equal(s.notes[0].enabled, false); assert.match(s.notes[0].reason, /control deadlines remain/); +}); + +test("notifications, inactive turns and concurrent heartbeats never open extra pings", async () => { + const s = watchdog(); await s.watch.open(); s.advance(STALLED_AFTER_MS); s.watch.touch(); + await s.watch.tick(true); assert.equal(s.pings(), 1); + s.advance(STALLED_AFTER_MS); await s.watch.tick(false); assert.equal(s.pings(), 1); + await Promise.all([s.watch.tick(true), s.watch.tick(true)]); assert.equal(s.pings(), 2); + s.watch.stop(); s.advance(STALLED_AFTER_MS); await s.watch.tick(true); assert.equal(s.pings(), 2); +}); diff --git a/tests/io-bounds.test.mjs b/tests/io-bounds.test.mjs index 44a0a7b8..b54d2b7e 100644 --- a/tests/io-bounds.test.mjs +++ b/tests/io-bounds.test.mjs @@ -3,7 +3,7 @@ import test from "node:test"; import assert from "node:assert/strict"; import fs from "node:fs"; import path from "node:path"; -import { SpawnedServerClient } from "../plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs"; +import { SpawnedServerClient, MAX_LINE_BYTES } from "../plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs"; import { KimiWire } from "../plugins/relay/scripts/lib/backends/kimi/wire.mjs"; import { makeTempDir, writeExecutable } from "./support/helpers.mjs"; @@ -22,7 +22,7 @@ else {const chunk='x'.repeat(1024*1024); let n=70; function write(){while(n-->0) await client.close(); assert.equal(client.exitSeen, true); assert.equal(client.lineBytes, 0); - assert.equal(client.lineChunks.length, 0); + assert.equal(client.lineBuffer, null); }); test("bounded close escalates TERM to KILL even after ChildProcess.killed is true", { timeout: 12000 }, async t => { @@ -117,3 +117,20 @@ for (const replay of [false, true]) test(`Kimi historical deliveries keep prompt assert.equal(wire.prompts.size, 65); assert.equal(wire.promptSince(0, "prompt-0").delivered, true); }); + + +test("many tiny RPC fragments retain one bounded buffer and preserve UTF-8 and line boundaries", () => { + const client = new SpawnedServerClient("/tmp"); + const received = []; + client.setNotificationHandler(message => received.push(message)); + const message = { method: "tiny/chunks", params: { text: "é🙂".repeat(20000) } }; + const bytes = Buffer.from(JSON.stringify(message)); + for (let i = 0; i < bytes.length; i++) client.handleChunk(bytes.subarray(i, i + 1)); + assert.equal(client.lineBytes, bytes.length); + assert.ok(client.lineBuffer.length < bytes.length * 2); + assert.ok(client.lineBuffer.length <= MAX_LINE_BYTES); + assert.equal(client.lineChunks, undefined, "no retained slice per input fragment"); + client.handleChunk(Buffer.from('\n{"method":"next"}\n')); + assert.deepEqual(received, [message, { method: "next" }]); + assert.equal(client.lineBuffer, null); assert.equal(client.lineBytes, 0); +}); diff --git a/tests/provider-evidence.test.mjs b/tests/provider-evidence.test.mjs new file mode 100644 index 00000000..fa9a7df5 --- /dev/null +++ b/tests/provider-evidence.test.mjs @@ -0,0 +1,81 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import path from "node:path"; +import { setImmediate as tick } from "node:timers/promises"; +import { SpawnedServerClient } from "../plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs"; +import { scriptedConnection } from "./support/backend-connection.mjs"; +import { makeTempDir } from "./support/helpers.mjs"; + +function scriptedClient() { + const client = new SpawnedServerClient("/tmp"); + let prompt; + client.request = async (method, params) => { + if (method === "turn/start") return { turn: { id: "t1", status: "inProgress" } }; + if (method === "session/prompt") return new Promise(resolve => { prompt = resolve; }); + return {}; + }; + return { client, finish: value => prompt(value), emit: (method, params) => client.handleLine(JSON.stringify({ method, params })) }; +} +const input = { prompt: "Task", clientId: "m1", source: "brief", onPending() {}, onTurnId() {} }; + +test("Codex model evidence and inner retry signals are scoped to active and subagent turns", async () => { + const { client, emit } = scriptedClient(); + const connection = await scriptedConnection("codex", client, {}); connection.sessionId = "s1"; + let responses = 0; const limits = []; + const running = connection.run({ ...input, onProviderResponse: () => responses++, onProviderLimit: e => limits.push(e) }); + await tick(); + emit("item/started", { threadId: "s1", turnId: "t1", item: { type: "userMessage", id: "u" } }); + emit("item/agentMessage/delta", { threadId: "other", turnId: "t1", delta: "Unrelated" }); + assert.equal(responses, 0); + emit("item/reasoning/textDelta", { threadId: "s1", turnId: "t1", delta: "Thought" }); + assert.equal(responses, 1); + emit("thread/started", { thread: { id: "subagent" } }); + emit("turn/started", { threadId: "subagent", turn: { id: "child-turn" } }); + const error = { message: "Usage exhausted", codexErrorInfo: { usageLimitExceeded: {} } }; + emit("error", { threadId: "subagent", turnId: "child-turn", willRetry: true, error }); + assert.deepEqual(limits, [error], "refusal reaches the worker before completion"); + emit("item/started", { threadId: "s1", turnId: "t1", item: { type: "commandExecution", id: "tool" } }); + assert.equal(responses, 2); + emit("turn/completed", { threadId: "s1", turn: { id: "t1", status: "completed", items: [] } }); + await running; +}); + +test("Grok ACP model updates and retry_state reach hooks before prompt completion", async t => { + const home = makeTempDir("relay-provider-evidence-"); t.after(() => fs.rmSync(home, { recursive: true, force: true })); + const { client, emit, finish } = scriptedClient(); + const connection = await scriptedConnection("grok", client, { cwd: home, grokHome: home }); connection.sessionId = "s1"; + let responses = 0; const limits = []; + const running = connection.run({ ...input, onProviderResponse: () => responses++, onProviderLimit: e => limits.push(e) }); + await tick(); + emit("session/update", { sessionId: "s1", update: { sessionUpdate: "user_message_chunk", content: { text: "User" } } }); + emit("session/update", { sessionId: "other", update: { sessionUpdate: "agent_message_chunk", content: { text: "Other" } } }); + assert.equal(responses, 0); + emit("_x.ai/session/update", { sessionId: "s1", update: { sessionUpdate: "retry_state", type: "retrying", reason: "HTTP 429 Too Many Requests" } }); + assert.equal(limits[0].message, "HTTP 429 Too Many Requests"); + for (const kind of ["agent_message_chunk", "agent_thought_chunk", "tool_call"]) emit("session/update", { sessionId: "s1", update: { sessionUpdate: kind, content: { text: "Answer" } } }); + assert.equal(responses, 3); + finish({ stopReason: "end_turn", _meta: { promptId: "p1" } }); await running; +}); + +test("Kimi observes only new wire retry records and ACP provider output during the turn", async t => { + const home = makeTempDir("relay-provider-evidence-"); t.after(() => fs.rmSync(home, { recursive: true, force: true })); + const dir = path.join(home, "sessions", "work", "s1", "agents", "main"); fs.mkdirSync(dir, { recursive: true }); + const file = path.join(dir, "wire.jsonl"); + const retry = { type: "turn.step.retrying", turnId: 1, errorName: "APIStatusError", errorMessage: "403 You have reached your weekly usage limit" }; + fs.writeFileSync(file, JSON.stringify(retry) + "\n"); + const { client, emit } = scriptedClient(); + const connection = await scriptedConnection("kimi", client, { cwd: home, kimiHome: home, mode: "yolo" }); + await connection.open({ sessionId: "s1" }); + let responses = 0; const limits = []; + const running = connection.run({ ...input, onProviderResponse: () => responses++, onProviderLimit: e => limits.push(e) }); + await tick(); assert.equal(limits.length, 0, "old retries cannot pause a new turn"); + fs.appendFileSync(file, JSON.stringify(retry) + "\n"); + connection.wire.read(); assert.equal(limits[0].message, retry.errorMessage); + emit("session/update", { sessionId: "s1", update: { sessionUpdate: "user_message_chunk" } }); + assert.equal(responses, 0); + emit("session/update", { sessionId: "s1", update: { sessionUpdate: "agent_message_chunk", content: { text: "Answer" } } }); + assert.equal(responses, 1); + await connection.close(); await running; + assert.equal(connection.wire.onRecord, null); +}); diff --git a/tests/zcode.test.mjs b/tests/zcode.test.mjs index 26508b27..10f218db 100644 --- a/tests/zcode.test.mjs +++ b/tests/zcode.test.mjs @@ -830,7 +830,7 @@ test("a duplicate with no session evidence tears down after the default 60 secon assert.equal(result.turnStatus, "failed"); assert.equal(result.teardown, true); assert.equal(result.retryable, false); - assert.equal(result.error.message, "ZCode already holds this message (duplicate id) and did not start it within 60 s; it was not sent again, so it can't run twice."); + assert.equal(result.error.message, "ZCode already holds this message (duplicate id) but relay could not confirm its start within 60 s; it was not sent again, so it can't run twice."); assert.deepEqual(client.requests.filter(r => r.params?.type === "sendText").map(r => r.params.commandId), ["m1"]); }); @@ -1008,3 +1008,65 @@ test("the GLM docs: ZCode is the only route, GLM-5.3 and GLM-5.3-Flash run, Flas const design = read("docs/design.md").replace(/\s+/g, " "); assert.match(design, /read its model list, and close it unsaved, with no message/); }); + + +test("cancel wakes a duplicate wait immediately, without waiting for its 60-second bound", async () => { + const { client } = await opened({ sendText: params => Promise.resolve(ack(params, { status: "duplicate" })) }); + const connection = connections.get(client); + const running = runOf(client); + await sleep(10); + await connection.interrupt(null); + const result = await running; + assert.equal(result.turnStatus, "interrupted"); + assert.equal(client.requests.filter(r => r.params?.type === "sendText").length, 1); +}); + +test("a duplicate steer remains uncertain until session inclusion proves it", async () => { + const { client } = await opened({ sendText: params => Promise.resolve(ack(params, { status: params.payload.requestedDelivery === "guide" ? "duplicate" : "accepted" })) }); + const connection = connections.get(client), running = runOf(client); + await waitFor(() => connection.controlsReady()); + client.emit("turn.started", "t1", { inputId: "m1" }); + assert.equal((await connection.steer("t1", { id: "guide", text: "Guide" })).unconfirmed, true); + assert.equal(connection.steerHeld("guide"), false); + client.emit("turn.steerDrained", "t1", { drainedInputs: [{ intent: { sourceCommandId: "guide" } }] }); + assert.equal(connection.steerHeld("guide"), true); + client.emit("turn.completed", "t1", { inputId: "m1", resultType: "success", response: "Done" }); + await running; +}); + +test("a cancel received with a discarded acknowledgement prevents the final resend", async () => { + let connection; + const { client } = await opened({ sendText: params => { + connection.session.cancelRequested = true; + return Promise.resolve(ack(params, { status: "failed", reasonCode: "fault.command.inputDiscardedOnRestart" })); + } }); + connection = connections.get(client); + assert.equal((await runOf(client)).turnStatus, "interrupted"); + assert.equal(client.requests.filter(r => r.params?.type === "sendText").length, 1); +}); + +test("ZCode only treats model output or a successful completion as provider evidence", async () => { + const { client } = await opened(); const connection = connections.get(client); + let responses = 0; + const running = connection.run({ prompt: "main", clientId: "m1", source: "brief", onPending() {}, onTurnId() {}, onProviderResponse: () => responses++ }); + await waitFor(() => connection.controlsReady()); + client.emit("turn.started", "t1", { inputId: "m1" }); + client.emit("turn.steerQueued", "t1", { inputId: "guide" }); + assert.equal(responses, 0); + client.emit("model.streaming", "t1", { kind: "text_delta", delta: "Answer" }); + assert.equal(responses, 1); + client.emit("turn.completed", "t1", { inputId: "m1", resultType: "success", response: "Done" }); + await running; assert.equal(responses, 2); +}); + + +test("the worker can cancel while ZCode is waiting for a duplicate with no turn id", async t => { + const ctx = fixture(t, { duplicateFirst: true }); ctx.env.RELAY_JOBS_ZCODE_DUPLICATE_MS = "60000"; + const { id } = startJob(ctx, HEADER, "Duplicate."); + await waitFor(() => readJob(ctx, id).state === "running"); + assert.equal(readJob(ctx, id).turnId, null); + companion(ctx, ["cancel", id]); await finishJob(ctx, id, 5); + assert.equal(readJob(ctx, id).state, "cancelled"); + assert.equal(commands(ctx, "stop").length, 1); + assert.equal(commands(ctx, "sendText").length, 1); +});