From 84e6fd1112cce32c27b44feadaf5d656d8fa1593 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Darko=20Miji=C4=87?= Date: Sat, 24 Oct 2026 22:07:24 +0200 Subject: [PATCH 1/2] backends: bound controls and require provider recovery evidence Add capability-tested silent-turn pings, early account pause signals, and ZCode duplicate/cancel guards. Coalesce RPC input fragments to bound retained transport memory. Co-Authored-By: GPT-6.1 Sol --- .claude-plugin/marketplace.json | 4 +- README.md | 23 +- package-lock.json | 4 +- package.json | 2 +- plugins/relay/.claude-plugin/plugin.json | 2 +- plugins/relay/CHANGELOG.md | 11 + .../scripts/lib/backends/codex/connection.mjs | 7 +- .../relay/scripts/lib/backends/codex/turn.mjs | 14 +- .../scripts/lib/backends/grok/connection.mjs | 14 +- .../scripts/lib/backends/kimi/connection.mjs | 15 +- .../relay/scripts/lib/backends/kimi/wire.mjs | 2 + .../scripts/lib/backends/transport/acp.mjs | 5 +- .../lib/backends/transport/stdio-rpc.mjs | 60 ++++- plugins/relay/scripts/lib/backends/types.d.ts | 9 + .../scripts/lib/backends/zcode/connection.mjs | 37 ++- plugins/relay/scripts/lib/worker/liveness.mjs | 41 ++++ plugins/relay/scripts/lib/worker/worker.mjs | 71 ++++-- plugins/relay/skills/jobs/internals.md | 21 ++ tests/admission-worker.test.mjs | 215 +++++++++++++++++- tests/backend-liveness.test.mjs | 100 ++++++++ tests/io-bounds.test.mjs | 21 +- tests/provider-evidence.test.mjs | 81 +++++++ tests/zcode.test.mjs | 64 +++++- 23 files changed, 754 insertions(+), 69 deletions(-) create mode 100644 plugins/relay/scripts/lib/worker/liveness.mjs create mode 100644 tests/backend-liveness.test.mjs create mode 100644 tests/provider-evidence.test.mjs 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..59edfd88 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 against a live quota fixture | +| 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..1f540e23 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,11 @@ Changed: Fixed: +- 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..89f721de 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,10 @@ export interface InboxMessage { } export interface TurnInput { + /** 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..a920b59a 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) 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/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..35b9ffdf 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,18 @@ 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({ + 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 +784,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 +844,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 +897,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 +914,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 +923,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 +972,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 +990,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..31e905f5 100644 --- a/plugins/relay/skills/jobs/internals.md +++ b/plugins/relay/skills/jobs/internals.md @@ -131,3 +131,24 @@ 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 and +silent turns. 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-worker.test.mjs b/tests/admission-worker.test.mjs index c25c92b9..ca285baf 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,208 @@ 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 close the server and use the one-recovery path", async t => { + const { TransportTimeout } = await import("../plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs"); + const s = seed(t); s.unblock(); + let now = 0, connections = 0; + 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) { + let pings = 0; + t.mock.method(connection, "ping", async () => { if (pings++ > 0) throw new TransportTimeout("codex", "account/read", 30000); }); + t.mock.method(connection, "run", async turn => { + turn.onPending(); turn.onTurnId("silent"); now = 10 * 60000; + await waitFor(() => readJob(s.dir).backendLiveness.misses === 1); + now += 5 * 60000; + return new Promise(() => {}); + }); + } + 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.ok(job.attempts[0].servers[0].endedAt); +}); + +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); +}); 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); +}); From 6ed24d8bf577c07fd19c51013228fda0ff31109c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Darko=20Miji=C4=87?= Date: Sat, 24 Oct 2026 23:11:11 +0200 Subject: [PATCH 2/2] Preserve probe fencing through retries and cancel pending ZCode input Keep live probe claims when refusals arrive, fold deferred refusals before probe success, and expose claimed cancellation before ZCode may resend discarded input. Add inbox and full transport-recovery regressions, and reconcile the documentation. Co-Authored-By: GPT-6.1 Sol --- README.md | 2 +- plugins/relay/CHANGELOG.md | 3 + plugins/relay/scripts/lib/backends/types.d.ts | 2 + .../scripts/lib/backends/zcode/connection.mjs | 2 +- .../relay/scripts/lib/worker/admission.mjs | 7 +- plugins/relay/scripts/lib/worker/worker.mjs | 1 + plugins/relay/skills/jobs/internals.md | 7 +- tests/admission-pauses.test.mjs | 32 ++++++ tests/admission-worker.test.mjs | 98 +++++++++++++++++-- 9 files changed, 140 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 59edfd88..de01ac55 100644 --- a/README.md +++ b/README.md @@ -381,7 +381,7 @@ These pauses gate new starts and later turns on other jobs while a running turn | Backend | Paused on | When | |---|---|---| | Codex | usage or rate | at first retry, including subagent signals | -| Grok | weekly credits or rate | at first retry, unverified against a live quota fixture | +| 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 | diff --git a/plugins/relay/CHANGELOG.md b/plugins/relay/CHANGELOG.md index 1f540e23..da3c30e3 100644 --- a/plugins/relay/CHANGELOG.md +++ b/plugins/relay/CHANGELOG.md @@ -27,6 +27,9 @@ 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 diff --git a/plugins/relay/scripts/lib/backends/types.d.ts b/plugins/relay/scripts/lib/backends/types.d.ts index 89f721de..51d084c7 100644 --- a/plugins/relay/scripts/lib/backends/types.d.ts +++ b/plugins/relay/scripts/lib/backends/types.d.ts @@ -250,6 +250,8 @@ 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. */ diff --git a/plugins/relay/scripts/lib/backends/zcode/connection.mjs b/plugins/relay/scripts/lib/backends/zcode/connection.mjs index a920b59a..472768dc 100644 --- a/plugins/relay/scripts/lib/backends/zcode/connection.mjs +++ b/plugins/relay/scripts/lib/backends/zcode/connection.mjs @@ -501,7 +501,7 @@ class ZcodeConnection { 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) return { turnId, turnStatus: "interrupted", finalMessage: "", ...receipts() }; + 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; 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/worker.mjs b/plugins/relay/scripts/lib/worker/worker.mjs index 35b9ffdf..e26fd6af 100644 --- a/plugins/relay/scripts/lib/worker/worker.mjs +++ b/plugins/relay/scripts/lib/worker/worker.mjs @@ -766,6 +766,7 @@ async function runWorkerRound(jobDir, options, held) { try { let providerAnswered = false; const running = connection.run({ + isCancelled: () => cancelRequested, onProviderResponse() { if (!providerAnswered) { providerAnswered = true; completeProbe(); } }, diff --git a/plugins/relay/skills/jobs/internals.md b/plugins/relay/skills/jobs/internals.md index 31e905f5..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 @@ -145,8 +145,9 @@ disables the watchdog for that connection with an event note, keeping control de `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 and -silent turns. A turn ending without model evidence releases the claim without clearing the pause. +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 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 ca285baf..d633a1ad 100644 --- a/tests/admission-worker.test.mjs +++ b/tests/admission-worker.test.mjs @@ -596,22 +596,51 @@ for (const [name, error] of [["codex", { codexErrorInfo: { usageLimitExceeded: { }); } -test("two unanswered watchdog pings close the server and use the one-recovery path", async t => { - const { TransportTimeout } = await import("../plugins/relay/scripts/lib/backends/transport/stdio-rpc.mjs"); +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(); - let now = 0, connections = 0; + // 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) { - let pings = 0; - t.mock.method(connection, "ping", async () => { if (pings++ > 0) throw new TransportTimeout("codex", "account/read", 30000); }); + const run = connection.run.bind(connection), close = connection.close.bind(connection); t.mock.method(connection, "run", async turn => { - turn.onPending(); turn.onTurnId("silent"); now = 10 * 60000; + 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 new Promise(() => {}); + 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; @@ -621,7 +650,9 @@ test("two unanswered watchdog pings close the server and use the one-recovery pa 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.ok(job.attempts[0].servers[0].endedAt); + 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 => { @@ -708,3 +739,54 @@ test("a worker leaves a silent 25-minute sleep alive while the backend answers p 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"); +});