From c3e8fbc3cdc93fbc13909aea93c0f1a0ee471079 Mon Sep 17 00:00:00 2001 From: Daniel Saldarriaga Date: Tue, 4 Aug 2026 12:03:19 -0500 Subject: [PATCH] Bound automatic continuation retries Co-authored-by: Sam Ruiz --- README.md | 4 +- dist/server.js | 359 ++++++++++++++++-- src/server.ts | 339 +++++++++++++++-- src/state.ts | 89 ++++- test/server.test.ts | 884 +++++++++++++++++++++++++++++++++++++++++++- test/state.test.ts | 159 ++++++++ 6 files changed, 1765 insertions(+), 69 deletions(-) diff --git a/README.md b/README.md index bbc8369..5609c26 100644 --- a/README.md +++ b/README.md @@ -106,8 +106,8 @@ Defaults: - `defer_while_tasks_active`: `true`; when enabled, goal auto-continuation waits for active OpenCode Task child sessions and their orchestrator reconciliation before sending the next goal prompt. - `max_auto_turns`: `25` - `min_continue_interval_seconds`: `3` -- `max_turn_time`: unset by default; set a positive number of seconds to retry one active-goal continuation prompt when a model turn remains busy for that long. Each new busy event resets the watchdog. Idle, built-in retry, session deletion, active Task children, and restricted agents suppress the retry. Watchdog retries are independent of `min_continue_interval_seconds` and do not consume auto-turn, no-progress, or prompt-failure budgets. -- `max_prompt_failures`: `3` +- `max_turn_time`: unset by default; set a positive number of seconds to retry one active-goal continuation prompt when a model turn remains busy for that long. Each new busy event resets the watchdog. Idle, built-in retry, session deletion, active Task children, and restricted agents suppress the retry. Watchdog retries are independent of `min_continue_interval_seconds` and never consume auto-turn or no-progress budgets, but recognized transport failures still count toward the `max_prompt_failures` ceiling. +- `max_prompt_failures`: `3`; consecutive transport or no-response continuation failures pause the goal at this ceiling. Prompt delivery alone does not reset the count; substantive assistant or tool progress, a new goal, or an explicit resume does. - `default_token_budget`: unset by default; when set, new goals inherit this token budget. - `max_goal_duration_seconds`: unset by default; when set, new goals inherit this elapsed-time safety limit. - `no_progress_token_threshold`: `50`; output-token floor used to judge whether a goal continuation turn made progress. diff --git a/dist/server.js b/dist/server.js index 79a11af..b53ea88 100644 --- a/dist/server.js +++ b/dist/server.js @@ -50,6 +50,8 @@ var GoalSchema = Schema.Struct({ autoTurns: Schema.Number, lastContinuationAt: NullableNumber, continuationFailures: Schema.optionalWith(Schema.Number, { default: () => 0 }), + pendingContinuationStart: Schema.optionalWith(NullableNumber, { default: () => null }), + pendingContinuationStarted: Schema.optionalWith(Schema.Boolean, { default: () => false }), lastStatus: Schema.optionalWith(NullableString, { default: () => null }), maxAutoTurns: Schema.optionalWith(NullableNumber, { default: () => null }), maxDurationSeconds: Schema.optionalWith(NullableNumber, { default: () => null }), @@ -172,6 +174,9 @@ function normalizeGoal(goal) { goal.lastAssistantMessageID ??= ""; goal.lastPromptAgent ??= null; goal.awaitingContinuationProgress = goal.awaitingContinuationProgress === true; + goal.lastContinuationAt = typeof goal.lastContinuationAt === "number" && Number.isFinite(goal.lastContinuationAt) ? Math.floor(goal.lastContinuationAt >= 1000000000000 ? goal.lastContinuationAt / 1000 : goal.lastContinuationAt) : null; + goal.pendingContinuationStart = typeof goal.pendingContinuationStart === "number" && Number.isFinite(goal.pendingContinuationStart) ? goal.pendingContinuationStart : null; + goal.pendingContinuationStarted = goal.pendingContinuationStarted === true; goal.continuationBaselineMessageID ??= ""; goal.continuationBaselineSummary ??= ""; goal.noProgressTurns = nonNegativeInteger(goal.noProgressTurns, 0); @@ -239,6 +244,8 @@ function snapshot(goal) { blocker: goal.blocker ?? null, closedAt: goal.closedAt ?? null, continuationFailures: goal.continuationFailures, + pendingContinuationStart: goal.pendingContinuationStart, + pendingContinuationStarted: goal.pendingContinuationStarted, lastStatus: goal.lastStatus, maxAutoTurns: goal.maxAutoTurns, maxDurationSeconds: goal.maxDurationSeconds, @@ -293,6 +300,8 @@ async function createGoal(sessionID, objective, options) { autoTurns: 0, lastContinuationAt: null, continuationFailures: 0, + pendingContinuationStart: null, + pendingContinuationStarted: false, lastStatus: paused ? "Goal recorded from Plan mode; execution paused until resumed from Build mode." : "Goal set.", maxAutoTurns: normalizedOptions.maxAutoTurns, maxDurationSeconds: normalizedOptions.maxDurationSeconds, @@ -336,6 +345,12 @@ async function updateGoalObjective(sessionID, objective, status = "active", opti goal.closedAt = null; goal.stopReason = planModePause ? PLAN_MODE_STOP_REASON : null; goal.budgetWrapupSent = false; + if (goal.status === "active") { + goal.continuationFailures = 0; + goal.pendingContinuationStart = null; + goal.pendingContinuationStarted = false; + goal.awaitingContinuationProgress = false; + } if (agent) goal.lastPromptAgent = agent; goal.lastStatus = planModePause ? "Goal objective updated; execution paused while the session is in Plan mode." : goal.status === "active" ? "Goal objective updated and resumed." : "Goal objective updated and paused."; @@ -387,6 +402,8 @@ async function setGoalStatus(sessionID, status, agent) { goal.updatedAt = nowSeconds(); goal.lastAccountedAt = status === "active" ? goal.updatedAt : null; goal.continuationFailures = status === "active" ? 0 : goal.continuationFailures; + goal.pendingContinuationStart = status === "active" ? null : goal.pendingContinuationStart; + goal.pendingContinuationStarted = status === "active" ? false : goal.pendingContinuationStarted; goal.noProgressTurns = status === "active" ? 0 : goal.noProgressTurns; goal.stopReason = status === "active" ? null : "paused"; goal.budgetWrapupSent = status === "active" ? false : goal.budgetWrapupSent; @@ -462,6 +479,7 @@ async function recordAssistantProgress(sessionID, input) { const threshold = positiveIntegerOrNull(input.noProgressTokenThreshold) ?? goal.noProgressTokenThreshold; const maxNoProgressTurns = positiveIntegerOrNull(input.maxNoProgressTurns) ?? goal.maxNoProgressTurns; const summary = summarizeText(text); + const substantive = /[\p{L}\p{N}]/u.test(text); const previousSummary = summarizeText(goal.lastAssistantText); const repeatedMessage = Boolean(messageID && messageID === goal.lastAssistantMessageID); const changed = Boolean(summary && summary !== previousSummary); @@ -471,9 +489,16 @@ async function recordAssistantProgress(sessionID, input) { goal.lastAssistantText = text; if (messageID) goal.lastAssistantMessageID = messageID; + if (substantive && summary && (!repeatedMessage || changed)) { + goal.continuationFailures = 0; + goal.pendingContinuationStart = null; + goal.pendingContinuationStarted = false; + } const continuationTurnCompleted = input.evaluateContinuation === true && goal.awaitingContinuationProgress && Boolean(messageID) && messageID !== goal.continuationBaselineMessageID; if (continuationTurnCompleted) { goal.awaitingContinuationProgress = false; + goal.pendingContinuationStart = null; + goal.pendingContinuationStarted = false; const lowOutput = outputTokens > 0 && outputTokens < (threshold ?? DEFAULT_NO_PROGRESS_TOKEN_THRESHOLD); const changedSinceContinuation = Boolean(summary && summary !== goal.continuationBaselineSummary); if (lowOutput && !changedSinceContinuation) { @@ -523,7 +548,7 @@ async function reserveContinuation(sessionID, maxAutoTurns, minIntervalSeconds) return snapshot(goal); }); } -async function recordContinuationResult(sessionID, result, maxFailures) { +async function recordContinuationResult(sessionID, result, maxFailures, options) { return mutate((state) => { const goal = state.goals[sessionID]; if (!goal || isClosed(goal.status)) @@ -531,15 +556,21 @@ async function recordContinuationResult(sessionID, result, maxFailures) { const now = nowSeconds(); goal.updatedAt = now; if (result === "success") { - goal.continuationFailures = 0; if (goal.status === "active") { + goal.pendingContinuationStart = Date.now(); + goal.pendingContinuationStarted = options?.started === true; goal.lastStatus = "Auto-continue prompt sent."; - goal.awaitingContinuationProgress = true; + if (options?.armNoProgress !== false) + goal.awaitingContinuationProgress = true; } return snapshot(goal); } + if (options?.requirePending && goal.pendingContinuationStart == null) + return null; goal.continuationFailures += 1; goal.awaitingContinuationProgress = false; + goal.pendingContinuationStart = null; + goal.pendingContinuationStarted = false; goal.lastStatus = `Auto-continue failed ${goal.continuationFailures} time(s).`; pushHistory(goal, "error", goal.lastStatus); if (goal.continuationFailures >= maxFailures) { @@ -554,6 +585,37 @@ async function recordContinuationResult(sessionID, result, maxFailures) { return snapshot(goal); }); } +async function markPendingContinuationStarted(sessionID) { + return mutate((state) => { + const goal = state.goals[sessionID]; + if (!goal || goal.status !== "active") + return goal ? snapshot(goal) : null; + if (goal.pendingContinuationStart == null || goal.pendingContinuationStarted) + return snapshot(goal); + goal.pendingContinuationStarted = true; + goal.updatedAt = nowSeconds(); + return snapshot(goal); + }); +} +async function recordToolProgress(sessionID, text) { + return mutate((state) => { + const goal = state.goals[sessionID]; + if (!goal || goal.status !== "active") + return goal ? snapshot(goal) : null; + const value = text?.trim() ?? ""; + if (!value) + return snapshot(goal); + if (goal.continuationFailures === 0 && goal.pendingContinuationStart == null) + return snapshot(goal); + goal.continuationFailures = 0; + goal.pendingContinuationStart = null; + goal.pendingContinuationStarted = false; + goal.awaitingContinuationProgress = false; + goal.noProgressTurns = 0; + goal.updatedAt = nowSeconds(); + return snapshot(goal); + }); +} function reserveWrapup(goal) { if (goal.budgetWrapupSent) return null; @@ -783,6 +845,11 @@ var DEFAULT_RESTRICTED_AGENTS = ["plan"]; var TASK_SETTLE_DELAY_MS = 25; var SNAPSHOT_IDLE_HOLD_MS = 250; var MAX_TIMER_DELAY_MS = 2147483647; +var STALE_PENDING_MS = 30000; +var RETRY_SETTLE_MS = 25; +var TRANSPORT_ERROR_PATTERN = /\b(?:network|fetch|socket|connect|connection|timeout|timed out|ECONNRESET|ECONNREFUSED|ETIMEDOUT|EAI_AGAIN|ENOTFOUND|EPIPE|transport|stream|websocket|offline|internet|request failed|proxy)\b/i; +var NON_TRANSPORT_TERMINAL_PATTERN = /\b(?:abort(?:ed)?|interrupt(?:ed|ion)?)\b/i; +var NON_PROGRESS_TOOLS = new Set(["get_goal", "get_goal_history"]); var TASK_TERMINAL_STATES = new Set(["completed", "error", "cancelled"]); var PLAN_MODE_CREATE_NOTICE = 'Goal recorded while the session is in Plan mode, so execution is paused. Do not start implementation work now. Ask the user to switch to Build mode and resume the goal (for example with "/goal resume") to begin execution.'; var LIMITED_GOAL_NOTICE = "Safety limit reached. Do not start or continue substantive work for this goal. Summarize useful progress, remaining work, and blockers, then wait for the user to resume or edit the goal."; @@ -825,6 +892,9 @@ function commandNameFromOptions(options) { function positiveIntegerOrNull2(value) { return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : null; } +function nonNegativeIntegerOrNull(value) { + return typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : null; +} function timeoutMillisecondsFromSeconds(value) { if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) return null; @@ -994,6 +1064,84 @@ function isIdleEvent(event) { const status = event.properties?.status; return event.type === "session.status" && typeof status === "object" && status !== null && status.type === "idle"; } +function isTransportError(error) { + const message = error instanceof Error ? error.message : typeof error === "string" ? error : ""; + if (!message || NON_TRANSPORT_TERMINAL_PATTERN.test(message)) + return false; + if (TRANSPORT_ERROR_PATTERN.test(message)) + return true; + return false; +} +function transportErrorMessageFromEvent(props) { + for (const candidate of [props.error, props.message, props.reason]) { + if (typeof candidate === "string" && candidate.trim()) + return candidate.trim(); + if (isRecord(candidate)) { + for (const key of ["message", "error", "reason", "description"]) { + const value = candidate[key]; + if (typeof value === "string" && value.trim()) + return value.trim(); + } + } + } + return ""; +} +function continuationRetryDelayMs(minIntervalSeconds, attemptAt, now = Date.now()) { + return Math.max(0, attemptAt + minIntervalSeconds * 1000 - now) + RETRY_SETTLE_MS; +} +function continuationDelayFromSnapshot(minIntervalSeconds, lastContinuationAt, now = Date.now()) { + if (lastContinuationAt == null) + return RETRY_SETTLE_MS; + return Math.max(0, (lastContinuationAt + minIntervalSeconds + 1) * 1000 - now) + RETRY_SETTLE_MS; +} +var TOOL_FAILURE_STATES = new Set([ + "failed", + "failure", + "error", + "cancelled", + "canceled", + "aborted", + "abort", + "interrupted", + "running", + "pending", + "in_progress", + "in-progress", + "incomplete", + "partial", + "timeout", + "timed_out" +]); +function toolOutputFailed(output) { + if (!isRecord(output)) + return true; + if (typeof output.error === "string" && output.error.trim()) + return true; + if (output.success === false) + return true; + const text = typeof output.output === "string" ? output.output.trim() : ""; + const state = output.state ?? output.status; + if (typeof state === "string") { + const normalized = state.trim().toLowerCase(); + if (TOOL_FAILURE_STATES.has(normalized)) + return true; + if (["completed", "complete", "success", "succeeded", "ok", "done"].includes(normalized)) + return false; + } + if (isRecord(output.metadata)) { + const metaState = output.metadata.state ?? output.metadata.status; + if (typeof metaState === "string" && TOOL_FAILURE_STATES.has(metaState.trim().toLowerCase())) + return true; + } + const taskState = parseTaskState(text); + if (taskState) + return taskState !== "completed"; + if (/^state:\s*(failed|failure|error|cancelled|canceled|aborted|abort|interrupted|running|pending|incomplete|partial|timeout|timed_out)\b/im.test(text)) + return true; + if (/^/i.test(text) || /^/i.test(text) || /^error:/i.test(text)) + return true; + return false; +} function sessionIDFromEvent(event) { const direct = event.properties?.sessionID; if (typeof direct === "string") @@ -1277,15 +1425,20 @@ class TaskTracker { } async function recordAssistantMessage(sessionID, message, options, evaluateContinuation = false) { if (!message) - return; - await recordAssistantProgress(sessionID, { - messageID: messageID(message), - text: textFromMessage(message), + return { goal: null, progressed: false }; + const before = await getGoal(sessionID); + const id = messageID(message) ?? ""; + const text = textFromMessage(message); + const progressed = Boolean(/[\p{L}\p{N}]/u.test(text) && (id !== (before?.lastAssistantMessageID ?? "") || text !== (before?.lastAssistantText ?? ""))); + const goal = await recordAssistantProgress(sessionID, { + messageID: id, + text, outputTokens: outputTokensFromMessage(message) ?? null, noProgressTokenThreshold: positiveIntegerOrNull2(options.no_progress_token_threshold), maxNoProgressTurns: positiveIntegerOrNull2(options.max_no_progress_turns), evaluateContinuation }); + return { goal, progressed }; } function mergeSystemReminder(output, reminder) { if (!reminder.trim()) @@ -1311,7 +1464,7 @@ var server = async ({ client }, options) => { const autoContinue = options?.auto_continue ?? true; const deferWhileTasksActive = options?.defer_while_tasks_active ?? true; const maxAutoTurns = positiveIntegerOrNull2(options?.max_auto_turns) ?? DEFAULT_MAX_AUTO_TURNS; - const minInterval = positiveIntegerOrNull2(options?.min_continue_interval_seconds) ?? DEFAULT_CONTINUE_INTERVAL_SECONDS; + const minInterval = nonNegativeIntegerOrNull(options?.min_continue_interval_seconds) ?? DEFAULT_CONTINUE_INTERVAL_SECONDS; const maxTurnTimeMs = timeoutMillisecondsFromSeconds(options?.max_turn_time); const maxPromptFailures = positiveIntegerOrNull2(options?.max_prompt_failures) ?? DEFAULT_MAX_PROMPT_FAILURES; const registerCommand = options?.register_command ?? true; @@ -1321,6 +1474,9 @@ var server = async ({ client }, options) => { const scheduledContinuations = new Map; const turnWatchdogs = new Map; const busySessions = new Set; + const nativeRetrySessions = new Set; + const locallyDeliveredPendingSessions = new Set; + const watchdogRescuedSessions = new Set; const planAgents = restrictedAgentSet(options); const isPlanAgent = (agent) => typeof agent === "string" && planAgents.has(agent.trim().toLowerCase()); async function createGoalFromTool(input, context) { @@ -1355,6 +1511,8 @@ var server = async ({ client }, options) => { function armTurnWatchdog(sessionID) { if (maxTurnTimeMs == null) return; + if (watchdogRescuedSessions.has(sessionID)) + return; clearTurnWatchdog(sessionID); const watchdog = { timer: setTimeout(() => void runTurnWatchdog(sessionID, watchdog), maxTurnTimeMs) @@ -1367,7 +1525,7 @@ var server = async ({ client }, options) => { async function runTurnWatchdog(sessionID, watchdog) { let claimedContinuation = false; try { - if (turnWatchdogs.get(sessionID) !== watchdog || !busySessions.has(sessionID)) + if (turnWatchdogs.get(sessionID) !== watchdog || !busySessions.has(sessionID) || watchdogRescuedSessions.has(sessionID)) return; const goal = await getGoal(sessionID); if (turnWatchdogs.get(sessionID) !== watchdog || !busySessions.has(sessionID)) @@ -1380,6 +1538,7 @@ var server = async ({ client }, options) => { const latestTurnAgent = agentFromMessage(latestAssistant); if (isPlanAgent(latestTurnAgent)) return; + await recordAssistantMessage(sessionID, latestAssistant, options ?? {}); const taskStatus = await taskBlockStatus(sessionID); if (turnWatchdogs.get(sessionID) !== watchdog || !busySessions.has(sessionID)) return; @@ -1393,9 +1552,16 @@ var server = async ({ client }, options) => { turnWatchdogs.delete(sessionID); activeContinuations.add(sessionID); claimedContinuation = true; + watchdogRescuedSessions.add(sessionID); await sendContinuation(client, sessionID, continuationPrompt(current), current.lastPromptAgent ?? latestTurnAgent ?? null); + await recordContinuationResult(sessionID, "success", maxPromptFailures, { armNoProgress: false, started: true }); + locallyDeliveredPendingSessions.add(sessionID); + clearTurnWatchdog(sessionID); } catch (error) { try { + if (claimedContinuation && isTransportError(error)) { + await recordContinuationResult(sessionID, "failure", maxPromptFailures); + } await client.app?.log?.({ body: { service: "opencode-goal-plugin", @@ -1414,19 +1580,44 @@ var server = async ({ client }, options) => { turnWatchdogs.delete(sessionID); } } - function scheduleSettledContinuation(sessionID, delayMs = TASK_SETTLE_DELAY_MS) { - if (scheduledContinuations.has(sessionID)) + function cancelScheduledContinuation(sessionID) { + const scheduled = scheduledContinuations.get(sessionID); + if (scheduled) + clearTimeout(scheduled.timer); + scheduledContinuations.delete(sessionID); + } + function scheduleSettledContinuation(sessionID, delayMs = TASK_SETTLE_DELAY_MS, replace = false, purpose = "settle") { + if (!replace && scheduledContinuations.has(sessionID)) return; - const timer = setTimeout(() => { - scheduledContinuations.delete(sessionID); - runAutoContinue(sessionID, true); + if (replace) + cancelScheduledContinuation(sessionID); + const scheduled = {}; + const timer = setTimeout(async () => { + try { + if (scheduledContinuations.get(sessionID) !== scheduled || nativeRetrySessions.has(sessionID)) + return; + if (purpose === "retry") { + const goal = await getGoal(sessionID); + if (!goal || goal.continuationFailures === 0 && goal.pendingContinuationStart == null) + return; + } + if (scheduledContinuations.get(sessionID) !== scheduled || nativeRetrySessions.has(sessionID)) + return; + await runAutoContinue(sessionID, true, scheduled); + } finally { + if (scheduledContinuations.get(sessionID) === scheduled) + scheduledContinuations.delete(sessionID); + } }, Math.max(0, delayMs)); + scheduled.timer = timer; + scheduled.purpose = purpose; const maybeUnref = timer; if (typeof maybeUnref.unref === "function") maybeUnref.unref(); - scheduledContinuations.set(sessionID, timer); + scheduledContinuations.set(sessionID, scheduled); } - async function runAutoContinue(sessionID, fromTaskDeferral = false) { + async function runAutoContinue(sessionID, fromTaskDeferral = false, scheduled) { + let reservedAt = null; if (busySessions.has(sessionID)) return; if (activeContinuations.has(sessionID)) @@ -1438,13 +1629,19 @@ var server = async ({ client }, options) => { const taskStatus = await taskBlockStatus(sessionID); if (taskStatus && taskStatus.blocked) { taskDeferredSessions.add(sessionID); - if (taskStatus.retryAt != null) - scheduleSettledContinuation(sessionID, taskStatus.retryAt - Date.now()); + if (taskStatus.retryAt != null) { + scheduleSettledContinuation(sessionID, taskStatus.retryAt - Date.now(), scheduled != null); + } return; } if (busySessions.has(sessionID)) return; - await recordAssistantMessage(sessionID, latestAssistant, options ?? {}, true); + const observed = await recordAssistantMessage(sessionID, latestAssistant, options ?? {}, true); + const queued = scheduledContinuations.get(sessionID); + if (observed.progressed && queued?.purpose !== "settle") + cancelScheduledContinuation(sessionID); + if (scheduled && scheduledContinuations.get(sessionID) !== scheduled) + return; const current = await getGoal(sessionID); if (!current) return; @@ -1461,13 +1658,45 @@ var server = async ({ client }, options) => { return; } taskDeferredSessions.delete(sessionID); + if (current.status === "active" && current.pendingContinuationStart != null) { + const pendingAgeMs = Date.now() - current.pendingContinuationStart; + const deliveredLocally = locallyDeliveredPendingSessions.has(sessionID); + if (!current.pendingContinuationStarted && (deliveredLocally || pendingAgeMs < STALE_PENDING_MS)) { + return; + } + const afterFailure = await recordContinuationResult(sessionID, "failure", maxPromptFailures, { + requirePending: true + }); + if (afterFailure) + locallyDeliveredPendingSessions.delete(sessionID); + if (autoContinue && afterFailure?.status === "active") { + scheduleSettledContinuation(sessionID, continuationRetryDelayMs(minInterval, current.pendingContinuationStart), true, "retry"); + } + return; + } + const queuedBeforeReserve = scheduledContinuations.get(sessionID); + if (queuedBeforeReserve && queuedBeforeReserve !== scheduled) + return; + if (!autoContinue) + return; + if (nativeRetrySessions.has(sessionID)) + return; const goal = await reserveContinuation(sessionID, maxAutoTurns, minInterval); if (!goal) return; + if (nativeRetrySessions.has(sessionID)) + return; + if (scheduled && scheduledContinuations.get(sessionID) !== scheduled) + return; + reservedAt = Date.now(); await sendContinuation(client, sessionID, goal.status === "active" ? continuationPrompt(goal) : limitPrompt(goal), goal.lastPromptAgent ?? latestTurnAgent ?? null); await recordContinuationResult(sessionID, "success", maxPromptFailures); + locallyDeliveredPendingSessions.add(sessionID); } catch (error) { - await recordContinuationResult(sessionID, "failure", maxPromptFailures); + const afterFailure = reservedAt == null ? null : await recordContinuationResult(sessionID, "failure", maxPromptFailures); + if (isTransportError(error) && autoContinue && reservedAt != null && afterFailure?.status === "active") { + scheduleSettledContinuation(sessionID, continuationRetryDelayMs(minInterval, reservedAt), true, "retry"); + } await client.app?.log?.({ body: { service: "opencode-goal-plugin", @@ -1482,12 +1711,15 @@ var server = async ({ client }, options) => { } return { async dispose() { - for (const timer of scheduledContinuations.values()) - clearTimeout(timer); + for (const scheduled of scheduledContinuations.values()) + clearTimeout(scheduled.timer); scheduledContinuations.clear(); for (const watchdog of turnWatchdogs.values()) clearTimeout(watchdog.timer); turnWatchdogs.clear(); + watchdogRescuedSessions.clear(); + locallyDeliveredPendingSessions.clear(); + nativeRetrySessions.clear(); }, async config(config) { if (!registerCommand) @@ -1598,6 +1830,27 @@ var server = async ({ client }, options) => { }, async "tool.execute.after"(input, output) { taskTracker.noteTaskOutput(input, output); + const sessionID = typeof input?.sessionID === "string" ? input.sessionID : undefined; + if (!sessionID) + return; + if (typeof input?.tool === "string" && NON_PROGRESS_TOOLS.has(input.tool.toLowerCase())) + return; + const toolResult = output; + if (toolOutputFailed(toolResult)) + return; + const text = typeof toolResult.output === "string" ? toolResult.output : undefined; + if (!text) + return; + const before = await getGoal(sessionID); + const scheduled = scheduledContinuations.get(sessionID); + const hasFailureEpisode = Boolean(before && (before.continuationFailures > 0 || before.pendingContinuationStart != null)); + if (!before || !hasFailureEpisode && scheduled?.purpose !== "recovery") + return; + const progressed = await recordToolProgress(sessionID, text); + if (progressed?.continuationFailures === 0 && progressed.pendingContinuationStart == null) { + locallyDeliveredPendingSessions.delete(sessionID); + cancelScheduledContinuation(sessionID); + } }, async "chat.message"(input, output) { const sessionID = typeof input?.sessionID === "string" ? input.sessionID : output.message?.sessionID; @@ -1612,7 +1865,10 @@ var server = async ({ client }, options) => { if (!sessionID) return; await accountUsage(sessionID, tokensFromMessages(output.messages)); - await recordAssistantMessage(sessionID, latestAssistantMessage(output.messages), options ?? {}); + const observed = await recordAssistantMessage(sessionID, latestAssistantMessage(output.messages), options ?? {}); + const scheduled = scheduledContinuations.get(sessionID); + if (observed.progressed && scheduled?.purpose !== "settle") + cancelScheduledContinuation(sessionID); }, async "experimental.chat.system.transform"(input, output) { if (typeof input.sessionID !== "string") @@ -1639,31 +1895,67 @@ var server = async ({ client }, options) => { if (sessionID && eventType === "session.status") { const status = event.properties?.status; if (isRecord(status) && typeof status.type === "string") { - if (status.type === "busy") + if (status.type === "busy") { busySessions.add(sessionID); + nativeRetrySessions.delete(sessionID); + } if (status.type === "busy") armTurnWatchdog(sessionID); + if (status.type === "busy") + await markPendingContinuationStarted(sessionID); if (status.type === "idle") { busySessions.delete(sessionID); + nativeRetrySessions.delete(sessionID); clearTurnWatchdog(sessionID); + watchdogRescuedSessions.delete(sessionID); } - if (status.type === "retry") + if (status.type === "retry") { + nativeRetrySessions.add(sessionID); clearTurnWatchdog(sessionID); + cancelScheduledContinuation(sessionID); + } taskTracker.observeSessionStatus(sessionID, status.type); } } if (sessionID && eventType === "session.idle") { busySessions.delete(sessionID); + nativeRetrySessions.delete(sessionID); clearTurnWatchdog(sessionID); + watchdogRescuedSessions.delete(sessionID); taskTracker.observeSessionStatus(sessionID, "idle"); } + if (sessionID && eventType === "session.error") { + busySessions.delete(sessionID); + nativeRetrySessions.delete(sessionID); + clearTurnWatchdog(sessionID); + watchdogRescuedSessions.delete(sessionID); + const props = event.properties ?? {}; + const errorMessage = transportErrorMessageFromEvent(props); + if (errorMessage && isTransportError(errorMessage)) { + const goal = await getGoal(sessionID); + if (goal?.status === "active") { + if (goal.pendingContinuationStart != null) { + const afterFailure = await recordContinuationResult(sessionID, "failure", maxPromptFailures, { + requirePending: true + }); + if (afterFailure) + locallyDeliveredPendingSessions.delete(sessionID); + if (autoContinue && afterFailure?.status === "active") { + scheduleSettledContinuation(sessionID, continuationRetryDelayMs(minInterval, goal.pendingContinuationStart), true, "retry"); + } + } else if (autoContinue) { + scheduleSettledContinuation(sessionID, continuationDelayFromSnapshot(minInterval, goal.lastContinuationAt), false, "recovery"); + } + } + } + } if (sessionID && eventType === "session.deleted") { busySessions.delete(sessionID); clearTurnWatchdog(sessionID); - const scheduled = scheduledContinuations.get(sessionID); - if (scheduled) - clearTimeout(scheduled); - scheduledContinuations.delete(sessionID); + watchdogRescuedSessions.delete(sessionID); + locallyDeliveredPendingSessions.delete(sessionID); + nativeRetrySessions.delete(sessionID); + cancelScheduledContinuation(sessionID); taskDeferredSessions.delete(sessionID); taskTracker.observeSessionDeleted(sessionID); } @@ -1671,12 +1963,17 @@ var server = async ({ client }, options) => { const props = event.properties ?? {}; const message = [props.info, props.message].find((value) => value && typeof value === "object"); taskTracker.observeAssistantMessage(sessionID, message); - await recordAssistantMessage(sessionID, message, options ?? {}); + const observed = await recordAssistantMessage(sessionID, message, options ?? {}); + const scheduled = scheduledContinuations.get(sessionID); + if (observed.progressed && scheduled?.purpose !== "settle") + cancelScheduledContinuation(sessionID); } - if (!autoContinue || !isIdleEvent(event)) + if (!isIdleEvent(event)) return; if (!sessionID) return; + if (!autoContinue && (await getGoal(sessionID))?.pendingContinuationStart == null) + return; await runAutoContinue(sessionID); } }; diff --git a/src/server.ts b/src/server.ts index e77f834..2c5380c 100644 --- a/src/server.ts +++ b/src/server.ts @@ -14,6 +14,8 @@ import { recordAssistantProgress, recordContinuationResult, recordPromptAgent, + recordToolProgress, + markPendingContinuationStarted, reserveContinuation, setGoalStatus, updateGoalObjective, @@ -64,6 +66,12 @@ const DEFAULT_RESTRICTED_AGENTS = ["plan"] const TASK_SETTLE_DELAY_MS = 25 const SNAPSHOT_IDLE_HOLD_MS = 250 const MAX_TIMER_DELAY_MS = 2_147_483_647 +const STALE_PENDING_MS = 30_000 +const RETRY_SETTLE_MS = 25 +const TRANSPORT_ERROR_PATTERN = + /\b(?:network|fetch|socket|connect|connection|timeout|timed out|ECONNRESET|ECONNREFUSED|ETIMEDOUT|EAI_AGAIN|ENOTFOUND|EPIPE|transport|stream|websocket|offline|internet|request failed|proxy)\b/i +const NON_TRANSPORT_TERMINAL_PATTERN = /\b(?:abort(?:ed)?|interrupt(?:ed|ion)?)\b/i +const NON_PROGRESS_TOOLS = new Set(["get_goal", "get_goal_history"]) const TASK_TERMINAL_STATES = new Set(["completed", "error", "cancelled"]) const PLAN_MODE_CREATE_NOTICE = 'Goal recorded while the session is in Plan mode, so execution is paused. Do not start implementation work now. Ask the user to switch to Build mode and resume the goal (for example with "/goal resume") to begin execution.' @@ -102,6 +110,11 @@ type TurnWatchdog = { timer: ReturnType } +type ScheduledContinuation = { + timer: ReturnType + purpose: "settle" | "recovery" | "retry" +} + function restrictedAgentSet(options?: Options) { if (options?.allow_goal_execution_from_plan === true) return new Set() const names = Array.isArray(options?.restricted_agents) ? options.restricted_agents : DEFAULT_RESTRICTED_AGENTS @@ -142,6 +155,10 @@ function positiveIntegerOrNull(value: unknown) { return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : null } +function nonNegativeIntegerOrNull(value: unknown) { + return typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : null +} + function timeoutMillisecondsFromSeconds(value: unknown) { if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) return null return Math.min(Math.ceil(value * 1000), MAX_TIMER_DELAY_MS) @@ -305,6 +322,80 @@ function isIdleEvent(event: { type?: string; properties?: Record) { + for (const candidate of [props.error, props.message, props.reason]) { + if (typeof candidate === "string" && candidate.trim()) return candidate.trim() + if (isRecord(candidate)) { + for (const key of ["message", "error", "reason", "description"]) { + const value = candidate[key] + if (typeof value === "string" && value.trim()) return value.trim() + } + } + } + return "" +} + +// Retry after the remaining minimum interval measured from the attempt's +// millisecond anchor. The public lastContinuationAt field remains in seconds. +function continuationRetryDelayMs(minIntervalSeconds: number, attemptAt: number, now = Date.now()) { + return Math.max(0, attemptAt + minIntervalSeconds * 1000 - now) + RETRY_SETTLE_MS +} + +function continuationDelayFromSnapshot(minIntervalSeconds: number, lastContinuationAt: number | null, now = Date.now()) { + if (lastContinuationAt == null) return RETRY_SETTLE_MS + // lastContinuationAt is floor(seconds), so include the remainder of that + // second to guarantee reserveContinuation cannot wake too early and wedge. + return Math.max(0, (lastContinuationAt + minIntervalSeconds + 1) * 1000 - now) + RETRY_SETTLE_MS +} + +const TOOL_FAILURE_STATES = new Set([ + "failed", + "failure", + "error", + "cancelled", + "canceled", + "aborted", + "abort", + "interrupted", + "running", + "pending", + "in_progress", + "in-progress", + "incomplete", + "partial", + "timeout", + "timed_out", +]) + +function toolOutputFailed(output: unknown) { + if (!isRecord(output)) return true + if (typeof output.error === "string" && output.error.trim()) return true + if (output.success === false) return true + const text = typeof output.output === "string" ? output.output.trim() : "" + const state = output.state ?? output.status + if (typeof state === "string") { + const normalized = state.trim().toLowerCase() + if (TOOL_FAILURE_STATES.has(normalized)) return true + if (["completed", "complete", "success", "succeeded", "ok", "done"].includes(normalized)) return false + } + if (isRecord(output.metadata)) { + const metaState = output.metadata.state ?? output.metadata.status + if (typeof metaState === "string" && TOOL_FAILURE_STATES.has(metaState.trim().toLowerCase())) return true + } + const taskState = parseTaskState(text) + if (taskState) return taskState !== "completed" + if (/^state:\s*(failed|failure|error|cancelled|canceled|aborted|abort|interrupted|running|pending|incomplete|partial|timeout|timed_out)\b/im.test(text)) return true + if (/^/i.test(text) || /^/i.test(text) || /^error:/i.test(text)) return true + return false +} + function sessionIDFromEvent(event: { type?: string; properties?: Record }) { const direct = event.properties?.sessionID if (typeof direct === "string") return direct @@ -596,15 +687,22 @@ async function recordAssistantMessage( options: Options, evaluateContinuation = false, ) { - if (!message) return - await recordAssistantProgress(sessionID, { - messageID: messageID(message), - text: textFromMessage(message), + if (!message) return { goal: null, progressed: false } + const before = await getGoal(sessionID) + const id = messageID(message) ?? "" + const text = textFromMessage(message) + const progressed = Boolean( + /[\p{L}\p{N}]/u.test(text) && (id !== (before?.lastAssistantMessageID ?? "") || text !== (before?.lastAssistantText ?? "")), + ) + const goal = await recordAssistantProgress(sessionID, { + messageID: id, + text, outputTokens: outputTokensFromMessage(message) ?? null, noProgressTokenThreshold: positiveIntegerOrNull(options.no_progress_token_threshold), maxNoProgressTurns: positiveIntegerOrNull(options.max_no_progress_turns), evaluateContinuation, }) + return { goal, progressed } } function mergeSystemReminder(output: { system: string[] }, reminder: string) { @@ -629,16 +727,22 @@ const server: Plugin = async ({ client }, options?: Options) => { const autoContinue = options?.auto_continue ?? true const deferWhileTasksActive = options?.defer_while_tasks_active ?? true const maxAutoTurns = positiveIntegerOrNull(options?.max_auto_turns) ?? DEFAULT_MAX_AUTO_TURNS - const minInterval = positiveIntegerOrNull(options?.min_continue_interval_seconds) ?? DEFAULT_CONTINUE_INTERVAL_SECONDS + const minInterval = nonNegativeIntegerOrNull(options?.min_continue_interval_seconds) ?? DEFAULT_CONTINUE_INTERVAL_SECONDS const maxTurnTimeMs = timeoutMillisecondsFromSeconds(options?.max_turn_time) const maxPromptFailures = positiveIntegerOrNull(options?.max_prompt_failures) ?? DEFAULT_MAX_PROMPT_FAILURES const registerCommand = options?.register_command ?? true const commandName = commandNameFromOptions(options) const taskTracker = new TaskTracker() const taskDeferredSessions = new Set() - const scheduledContinuations = new Map>() + const scheduledContinuations = new Map() const turnWatchdogs = new Map() const busySessions = new Set() + const nativeRetrySessions = new Set() + const locallyDeliveredPendingSessions = new Set() + // Sessions whose busy episode already received a watchdog rescue. Cleared + // when the episode ends (idle/deleted), so each busy episode rescues at most + // once and a rescue prompt cannot recursively re-arm the watchdog. + const watchdogRescuedSessions = new Set() const planAgents = restrictedAgentSet(options) const isPlanAgent = (agent: unknown) => typeof agent === "string" && planAgents.has(agent.trim().toLowerCase()) @@ -674,6 +778,7 @@ const server: Plugin = async ({ client }, options?: Options) => { function armTurnWatchdog(sessionID: string) { if (maxTurnTimeMs == null) return + if (watchdogRescuedSessions.has(sessionID)) return clearTurnWatchdog(sessionID) const watchdog: TurnWatchdog = { timer: setTimeout(() => void runTurnWatchdog(sessionID, watchdog), maxTurnTimeMs), @@ -686,7 +791,12 @@ const server: Plugin = async ({ client }, options?: Options) => { async function runTurnWatchdog(sessionID: string, watchdog: TurnWatchdog) { let claimedContinuation = false try { - if (turnWatchdogs.get(sessionID) !== watchdog || !busySessions.has(sessionID)) return + if ( + turnWatchdogs.get(sessionID) !== watchdog || + !busySessions.has(sessionID) || + watchdogRescuedSessions.has(sessionID) + ) + return const goal = await getGoal(sessionID) if (turnWatchdogs.get(sessionID) !== watchdog || !busySessions.has(sessionID)) return if (goal?.status !== "active" || isPlanAgent(goal.lastPromptAgent)) return @@ -694,6 +804,9 @@ const server: Plugin = async ({ client }, options?: Options) => { if (turnWatchdogs.get(sessionID) !== watchdog || !busySessions.has(sessionID)) return const latestTurnAgent = agentFromMessage(latestAssistant) if (isPlanAgent(latestTurnAgent)) return + // Establish the pre-rescue baseline so this same historical message + // cannot later be mistaken for progress from the rescue prompt. + await recordAssistantMessage(sessionID, latestAssistant, options ?? {}) const taskStatus = await taskBlockStatus(sessionID) if (turnWatchdogs.get(sessionID) !== watchdog || !busySessions.has(sessionID)) return if (taskStatus && taskStatus.blocked) return @@ -704,9 +817,24 @@ const server: Plugin = async ({ client }, options?: Options) => { turnWatchdogs.delete(sessionID) activeContinuations.add(sessionID) claimedContinuation = true + watchdogRescuedSessions.add(sessionID) await sendContinuation(client, sessionID, continuationPrompt(current), current.lastPromptAgent ?? latestTurnAgent ?? null) + // Watchdog rescues are untracked retries: a delivered prompt arms the + // pending-continuation window but never consumes an auto-turn budget and + // never arms the no-progress evaluation. The rescue delivers while the + // session is already inside a busy episode, so the pending attempt is + // marked started immediately, and this busy episode rescues only once. + await recordContinuationResult(sessionID, "success", maxPromptFailures, { armNoProgress: false, started: true }) + locallyDeliveredPendingSessions.add(sessionID) + clearTurnWatchdog(sessionID) } catch (error) { try { + // Watchdog rescues share the same prompt-failure ceiling: recognized + // transport errors accumulate toward max_prompt_failures without + // consuming auto-turn budgets. + if (claimedContinuation && isTransportError(error)) { + await recordContinuationResult(sessionID, "failure", maxPromptFailures) + } await client.app?.log?.({ body: { service: "opencode-goal-plugin", @@ -724,18 +852,47 @@ const server: Plugin = async ({ client }, options?: Options) => { } } - function scheduleSettledContinuation(sessionID: string, delayMs = TASK_SETTLE_DELAY_MS) { - if (scheduledContinuations.has(sessionID)) return - const timer = setTimeout(() => { - scheduledContinuations.delete(sessionID) - void runAutoContinue(sessionID, true) + function cancelScheduledContinuation(sessionID: string) { + const scheduled = scheduledContinuations.get(sessionID) + if (scheduled) clearTimeout(scheduled.timer) + scheduledContinuations.delete(sessionID) + } + + function scheduleSettledContinuation( + sessionID: string, + delayMs = TASK_SETTLE_DELAY_MS, + replace = false, + purpose: ScheduledContinuation["purpose"] = "settle", + ) { + if (!replace && scheduledContinuations.has(sessionID)) return + if (replace) cancelScheduledContinuation(sessionID) + const scheduled = {} as ScheduledContinuation + const timer = setTimeout(async () => { + try { + if (scheduledContinuations.get(sessionID) !== scheduled || nativeRetrySessions.has(sessionID)) return + if (purpose === "retry") { + const goal = await getGoal(sessionID) + if (!goal || (goal.continuationFailures === 0 && goal.pendingContinuationStart == null)) return + } + if (scheduledContinuations.get(sessionID) !== scheduled || nativeRetrySessions.has(sessionID)) return + await runAutoContinue(sessionID, true, scheduled) + } finally { + if (scheduledContinuations.get(sessionID) === scheduled) scheduledContinuations.delete(sessionID) + } }, Math.max(0, delayMs)) + scheduled.timer = timer + scheduled.purpose = purpose const maybeUnref = timer as { unref?: () => void } if (typeof maybeUnref.unref === "function") maybeUnref.unref() - scheduledContinuations.set(sessionID, timer) + scheduledContinuations.set(sessionID, scheduled) } - async function runAutoContinue(sessionID: string, fromTaskDeferral = false) { + async function runAutoContinue( + sessionID: string, + fromTaskDeferral = false, + scheduled?: ScheduledContinuation, + ) { + let reservedAt: number | null = null if (busySessions.has(sessionID)) return if (activeContinuations.has(sessionID)) return activeContinuations.add(sessionID) @@ -745,11 +902,16 @@ const server: Plugin = async ({ client }, options?: Options) => { const taskStatus = await taskBlockStatus(sessionID) if (taskStatus && taskStatus.blocked) { taskDeferredSessions.add(sessionID) - if (taskStatus.retryAt != null) scheduleSettledContinuation(sessionID, taskStatus.retryAt - Date.now()) + if (taskStatus.retryAt != null) { + scheduleSettledContinuation(sessionID, taskStatus.retryAt - Date.now(), scheduled != null) + } return } if (busySessions.has(sessionID)) return - await recordAssistantMessage(sessionID, latestAssistant, options ?? {}, true) + const observed = await recordAssistantMessage(sessionID, latestAssistant, options ?? {}, true) + const queued = scheduledContinuations.get(sessionID) + if (observed.progressed && queued?.purpose !== "settle") cancelScheduledContinuation(sessionID) + if (scheduled && scheduledContinuations.get(sessionID) !== scheduled) return const current = await getGoal(sessionID) if (!current) return const latestTurnAgent = agentFromMessage(latestAssistant) @@ -763,8 +925,48 @@ const server: Plugin = async ({ client }, options?: Options) => { return } taskDeferredSessions.delete(sessionID) + + // Pending-continuation resolution. A delivered prompt is armed with + // started=false until a session.status busy event marks it started. + // Paired duplicate idles before any busy must never count a failure or + // send a duplicate, so started=false attempts are left alone until they + // go stale (restart recovery). Once started, the following logical idle + // with no substantive progress counts exactly one unresolved failure and + // schedules a bounded retry at the remaining min interval. + if (current.status === "active" && current.pendingContinuationStart != null) { + const pendingAgeMs = Date.now() - current.pendingContinuationStart + const deliveredLocally = locallyDeliveredPendingSessions.has(sessionID) + if (!current.pendingContinuationStarted && (deliveredLocally || pendingAgeMs < STALE_PENDING_MS)) { + return + } + const afterFailure = await recordContinuationResult(sessionID, "failure", maxPromptFailures, { + requirePending: true, + }) + if (afterFailure) locallyDeliveredPendingSessions.delete(sessionID) + if (autoContinue && afterFailure?.status === "active") { + scheduleSettledContinuation( + sessionID, + continuationRetryDelayMs(minInterval, current.pendingContinuationStart), + true, + "retry", + ) + } + return + } + + // A retry or recovery timer is already scheduled for this session (for + // example from a paired idle after an unresolved failure); let that timer + // drive the next attempt instead of sending a duplicate now. + const queuedBeforeReserve = scheduledContinuations.get(sessionID) + if (queuedBeforeReserve && queuedBeforeReserve !== scheduled) return + if (!autoContinue) return + if (nativeRetrySessions.has(sessionID)) return + const goal = await reserveContinuation(sessionID, maxAutoTurns, minInterval) if (!goal) return + if (nativeRetrySessions.has(sessionID)) return + if (scheduled && scheduledContinuations.get(sessionID) !== scheduled) return + reservedAt = Date.now() await sendContinuation( client, sessionID, @@ -772,8 +974,12 @@ const server: Plugin = async ({ client }, options?: Options) => { goal.lastPromptAgent ?? latestTurnAgent ?? null, ) await recordContinuationResult(sessionID, "success", maxPromptFailures) + locallyDeliveredPendingSessions.add(sessionID) } catch (error) { - await recordContinuationResult(sessionID, "failure", maxPromptFailures) + const afterFailure = reservedAt == null ? null : await recordContinuationResult(sessionID, "failure", maxPromptFailures) + if (isTransportError(error) && autoContinue && reservedAt != null && afterFailure?.status === "active") { + scheduleSettledContinuation(sessionID, continuationRetryDelayMs(minInterval, reservedAt), true, "retry") + } await client.app?.log?.({ body: { service: "opencode-goal-plugin", @@ -789,10 +995,13 @@ const server: Plugin = async ({ client }, options?: Options) => { return { async dispose() { - for (const timer of scheduledContinuations.values()) clearTimeout(timer) + for (const scheduled of scheduledContinuations.values()) clearTimeout(scheduled.timer) scheduledContinuations.clear() for (const watchdog of turnWatchdogs.values()) clearTimeout(watchdog.timer) turnWatchdogs.clear() + watchdogRescuedSessions.clear() + locallyDeliveredPendingSessions.clear() + nativeRetrySessions.clear() }, async config(config) { if (!registerCommand) return @@ -922,6 +1131,27 @@ const server: Plugin = async ({ client }, options?: Options) => { input as { tool?: unknown; sessionID?: unknown; callID?: unknown }, output as { output?: unknown }, ) + const sessionID = typeof input?.sessionID === "string" ? input.sessionID : undefined + if (!sessionID) return + if (typeof input?.tool === "string" && NON_PROGRESS_TOOLS.has(input.tool.toLowerCase())) return + const toolResult = output as { output?: unknown; error?: unknown } + // A successful tool output is real progress: it resolves any pending + // continuation and clears the prompt-failure counter. Failed tool + // outputs leave the failure counter and pending window untouched. + if (toolOutputFailed(toolResult)) return + const text = typeof toolResult.output === "string" ? toolResult.output : undefined + if (!text) return + const before = await getGoal(sessionID) + const scheduled = scheduledContinuations.get(sessionID) + const hasFailureEpisode = Boolean( + before && (before.continuationFailures > 0 || before.pendingContinuationStart != null), + ) + if (!before || (!hasFailureEpisode && scheduled?.purpose !== "recovery")) return + const progressed = await recordToolProgress(sessionID, text) + if (progressed?.continuationFailures === 0 && progressed.pendingContinuationStart == null) { + locallyDeliveredPendingSessions.delete(sessionID) + cancelScheduledContinuation(sessionID) + } }, async "chat.message"(input, output) { const sessionID = typeof input?.sessionID === "string" ? input.sessionID : output.message?.sessionID @@ -937,7 +1167,9 @@ const server: Plugin = async ({ client }, options?: Options) => { : output.messages.find((message) => typeof message.info.sessionID === "string")?.info.sessionID if (!sessionID) return await accountUsage(sessionID, tokensFromMessages(output.messages)) - await recordAssistantMessage(sessionID, latestAssistantMessage(output.messages), options ?? {}) + const observed = await recordAssistantMessage(sessionID, latestAssistantMessage(output.messages), options ?? {}) + const scheduled = scheduledContinuations.get(sessionID) + if (observed.progressed && scheduled?.purpose !== "settle") cancelScheduledContinuation(sessionID) }, async "experimental.chat.system.transform"(input, output) { if (typeof input.sessionID !== "string") return @@ -961,27 +1193,79 @@ const server: Plugin = async ({ client }, options?: Options) => { if (sessionID && eventType === "session.status") { const status = (event as { properties?: Record }).properties?.status if (isRecord(status) && typeof status.type === "string") { - if (status.type === "busy") busySessions.add(sessionID) + if (status.type === "busy") { + busySessions.add(sessionID) + nativeRetrySessions.delete(sessionID) + } if (status.type === "busy") armTurnWatchdog(sessionID) + if (status.type === "busy") await markPendingContinuationStarted(sessionID) if (status.type === "idle") { busySessions.delete(sessionID) + nativeRetrySessions.delete(sessionID) clearTurnWatchdog(sessionID) + watchdogRescuedSessions.delete(sessionID) + } + if (status.type === "retry") { + nativeRetrySessions.add(sessionID) + clearTurnWatchdog(sessionID) + cancelScheduledContinuation(sessionID) } - if (status.type === "retry") clearTurnWatchdog(sessionID) taskTracker.observeSessionStatus(sessionID, status.type) } } if (sessionID && eventType === "session.idle") { busySessions.delete(sessionID) + nativeRetrySessions.delete(sessionID) clearTurnWatchdog(sessionID) + watchdogRescuedSessions.delete(sessionID) taskTracker.observeSessionStatus(sessionID, "idle") } + if (sessionID && eventType === "session.error") { + busySessions.delete(sessionID) + nativeRetrySessions.delete(sessionID) + clearTurnWatchdog(sessionID) + watchdogRescuedSessions.delete(sessionID) + const props = (event as { properties?: Record }).properties ?? {} + const errorMessage = transportErrorMessageFromEvent(props) + if (errorMessage && isTransportError(errorMessage)) { + const goal = await getGoal(sessionID) + if (goal?.status === "active") { + if (goal.pendingContinuationStart != null) { + // The pending attempt failed at the transport level: count one + // failure and retry at the remaining min interval. + const afterFailure = await recordContinuationResult(sessionID, "failure", maxPromptFailures, { + requirePending: true, + }) + if (afterFailure) locallyDeliveredPendingSessions.delete(sessionID) + if (autoContinue && afterFailure?.status === "active") { + scheduleSettledContinuation( + sessionID, + continuationRetryDelayMs(minInterval, goal.pendingContinuationStart), + true, + "retry", + ) + } + } else if (autoContinue) { + // No pending attempt: start the first bounded automatic recovery + // without charging a phantom failure. Duplicate transport events + // dedupe through the scheduled-continuation timer. + scheduleSettledContinuation( + sessionID, + continuationDelayFromSnapshot(minInterval, goal.lastContinuationAt), + false, + "recovery", + ) + } + } + } + } if (sessionID && eventType === "session.deleted") { busySessions.delete(sessionID) clearTurnWatchdog(sessionID) - const scheduled = scheduledContinuations.get(sessionID) - if (scheduled) clearTimeout(scheduled) - scheduledContinuations.delete(sessionID) + watchdogRescuedSessions.delete(sessionID) + locallyDeliveredPendingSessions.delete(sessionID) + nativeRetrySessions.delete(sessionID) + cancelScheduledContinuation(sessionID) taskDeferredSessions.delete(sessionID) taskTracker.observeSessionDeleted(sessionID) } @@ -991,11 +1275,14 @@ const server: Plugin = async ({ client }, options?: Options) => { | { info?: unknown; role?: unknown; id?: unknown; time?: unknown; parts?: unknown[] } | undefined taskTracker.observeAssistantMessage(sessionID, message) - await recordAssistantMessage(sessionID, message, options ?? {}) + const observed = await recordAssistantMessage(sessionID, message, options ?? {}) + const scheduled = scheduledContinuations.get(sessionID) + if (observed.progressed && scheduled?.purpose !== "settle") cancelScheduledContinuation(sessionID) } - if (!autoContinue || !isIdleEvent(event as never)) return + if (!isIdleEvent(event as never)) return if (!sessionID) return + if (!autoContinue && (await getGoal(sessionID))?.pendingContinuationStart == null) return await runAutoContinue(sessionID) }, } diff --git a/src/state.ts b/src/state.ts index a66cbd6..52ea348 100644 --- a/src/state.ts +++ b/src/state.ts @@ -65,6 +65,8 @@ export type Goal = { autoTurns: number lastContinuationAt: number | null continuationFailures: number + pendingContinuationStart: number | null + pendingContinuationStarted: boolean lastStatus: string | null maxAutoTurns: number | null maxDurationSeconds: number | null @@ -148,6 +150,8 @@ const GoalSchema = Schema.Struct({ autoTurns: Schema.Number, lastContinuationAt: NullableNumber, continuationFailures: Schema.optionalWith(Schema.Number, { default: () => 0 }), + pendingContinuationStart: Schema.optionalWith(NullableNumber, { default: () => null }), + pendingContinuationStarted: Schema.optionalWith(Schema.Boolean, { default: () => false }), lastStatus: Schema.optionalWith(NullableString, { default: () => null }), maxAutoTurns: Schema.optionalWith(NullableNumber, { default: () => null }), maxDurationSeconds: Schema.optionalWith(NullableNumber, { default: () => null }), @@ -313,6 +317,15 @@ function normalizeGoal(goal: Goal) { goal.lastAssistantMessageID ??= "" goal.lastPromptAgent ??= null goal.awaitingContinuationProgress = goal.awaitingContinuationProgress === true + goal.lastContinuationAt = + typeof goal.lastContinuationAt === "number" && Number.isFinite(goal.lastContinuationAt) + ? Math.floor(goal.lastContinuationAt >= 1_000_000_000_000 ? goal.lastContinuationAt / 1000 : goal.lastContinuationAt) + : null + goal.pendingContinuationStart = + typeof goal.pendingContinuationStart === "number" && Number.isFinite(goal.pendingContinuationStart) + ? goal.pendingContinuationStart + : null + goal.pendingContinuationStarted = goal.pendingContinuationStarted === true goal.continuationBaselineMessageID ??= "" goal.continuationBaselineSummary ??= "" goal.noProgressTurns = nonNegativeInteger(goal.noProgressTurns, 0) @@ -388,6 +401,8 @@ export function snapshot(goal: Goal): GoalSnapshot { blocker: goal.blocker ?? null, closedAt: goal.closedAt ?? null, continuationFailures: goal.continuationFailures, + pendingContinuationStart: goal.pendingContinuationStart, + pendingContinuationStarted: goal.pendingContinuationStarted, lastStatus: goal.lastStatus, maxAutoTurns: goal.maxAutoTurns, maxDurationSeconds: goal.maxDurationSeconds, @@ -450,6 +465,8 @@ export async function createGoal(sessionID: string, objective: string, options?: autoTurns: 0, lastContinuationAt: null, continuationFailures: 0, + pendingContinuationStart: null, + pendingContinuationStarted: false, lastStatus: paused ? "Goal recorded from Plan mode; execution paused until resumed from Build mode." : "Goal set.", maxAutoTurns: normalizedOptions.maxAutoTurns, maxDurationSeconds: normalizedOptions.maxDurationSeconds, @@ -497,6 +514,12 @@ export async function updateGoalObjective( goal.closedAt = null goal.stopReason = planModePause ? PLAN_MODE_STOP_REASON : null goal.budgetWrapupSent = false + if (goal.status === "active") { + goal.continuationFailures = 0 + goal.pendingContinuationStart = null + goal.pendingContinuationStarted = false + goal.awaitingContinuationProgress = false + } if (agent) goal.lastPromptAgent = agent goal.lastStatus = planModePause ? "Goal objective updated; execution paused while the session is in Plan mode." @@ -548,6 +571,8 @@ export async function setGoalStatus(sessionID: string, status: MutableGoalStatus goal.updatedAt = nowSeconds() goal.lastAccountedAt = status === "active" ? goal.updatedAt : null goal.continuationFailures = status === "active" ? 0 : goal.continuationFailures + goal.pendingContinuationStart = status === "active" ? null : goal.pendingContinuationStart + goal.pendingContinuationStarted = status === "active" ? false : goal.pendingContinuationStarted goal.noProgressTurns = status === "active" ? 0 : goal.noProgressTurns goal.stopReason = status === "active" ? null : "paused" goal.budgetWrapupSent = status === "active" ? false : goal.budgetWrapupSent @@ -637,6 +662,7 @@ export async function recordAssistantProgress(sessionID: string, input: Assistan const threshold = positiveIntegerOrNull(input.noProgressTokenThreshold) ?? goal.noProgressTokenThreshold const maxNoProgressTurns = positiveIntegerOrNull(input.maxNoProgressTurns) ?? goal.maxNoProgressTurns const summary = summarizeText(text) + const substantive = /[\p{L}\p{N}]/u.test(text) const previousSummary = summarizeText(goal.lastAssistantText) const repeatedMessage = Boolean(messageID && messageID === goal.lastAssistantMessageID) const changed = Boolean(summary && summary !== previousSummary) @@ -645,6 +671,15 @@ export async function recordAssistantProgress(sessionID: string, input: Assistan if (text) goal.lastAssistantText = text if (messageID) goal.lastAssistantMessageID = messageID + // Substantive assistant text proves the continuation transport is healthy, + // so a pending continuation is resolved and any accumulated prompt failures + // are cleared. Delivery of a prompt alone never resets the counter. + if (substantive && summary && (!repeatedMessage || changed)) { + goal.continuationFailures = 0 + goal.pendingContinuationStart = null + goal.pendingContinuationStarted = false + } + // No-progress accounting is scoped to goal continuation turns: it only runs // once per reserved continuation, when the completed turn is observed at the // next idle. Generic observation paths (messages.transform, message.updated) @@ -656,6 +691,8 @@ export async function recordAssistantProgress(sessionID: string, input: Assistan messageID !== goal.continuationBaselineMessageID if (continuationTurnCompleted) { goal.awaitingContinuationProgress = false + goal.pendingContinuationStart = null + goal.pendingContinuationStarted = false const lowOutput = outputTokens > 0 && outputTokens < (threshold ?? DEFAULT_NO_PROGRESS_TOKEN_THRESHOLD) const changedSinceContinuation = Boolean(summary && summary !== goal.continuationBaselineSummary) if (lowOutput && !changedSinceContinuation) { @@ -706,22 +743,37 @@ export async function reserveContinuation(sessionID: string, maxAutoTurns: numbe }) } -export async function recordContinuationResult(sessionID: string, result: "success" | "failure", maxFailures: number) { +export async function recordContinuationResult( + sessionID: string, + result: "success" | "failure", + maxFailures: number, + options?: { armNoProgress?: boolean; started?: boolean; requirePending?: boolean }, +) { return mutate((state) => { const goal = state.goals[sessionID] if (!goal || isClosed(goal.status)) return goal ? snapshot(goal) : null const now = nowSeconds() goal.updatedAt = now if (result === "success") { - goal.continuationFailures = 0 + // Successful prompt delivery arms the pending-continuation window but + // never resets the failure counter; only real progress, a successful + // tool output, or an explicit resume/new goal clears it. Delivery alone + // is not "started": a session.status busy event marks the attempt as + // actually started through markPendingContinuationStarted. Watchdog + // rescues deliver while already busy and pass started: true. if (goal.status === "active") { + goal.pendingContinuationStart = Date.now() + goal.pendingContinuationStarted = options?.started === true goal.lastStatus = "Auto-continue prompt sent." - goal.awaitingContinuationProgress = true + if (options?.armNoProgress !== false) goal.awaitingContinuationProgress = true } return snapshot(goal) } + if (options?.requirePending && goal.pendingContinuationStart == null) return null goal.continuationFailures += 1 goal.awaitingContinuationProgress = false + goal.pendingContinuationStart = null + goal.pendingContinuationStarted = false goal.lastStatus = `Auto-continue failed ${goal.continuationFailures} time(s).` pushHistory(goal, "error", goal.lastStatus) if (goal.continuationFailures >= maxFailures) { @@ -737,6 +789,37 @@ export async function recordContinuationResult(sessionID: string, result: "succe }) } +export async function markPendingContinuationStarted(sessionID: string) { + return mutate((state) => { + const goal = state.goals[sessionID] + if (!goal || goal.status !== "active") return goal ? snapshot(goal) : null + if (goal.pendingContinuationStart == null || goal.pendingContinuationStarted) return snapshot(goal) + goal.pendingContinuationStarted = true + goal.updatedAt = nowSeconds() + return snapshot(goal) + }) +} + +export async function recordToolProgress(sessionID: string, text?: string) { + return mutate((state) => { + const goal = state.goals[sessionID] + if (!goal || goal.status !== "active") return goal ? snapshot(goal) : null + const value = text?.trim() ?? "" + if (!value) return snapshot(goal) + if (goal.continuationFailures === 0 && goal.pendingContinuationStart == null) return snapshot(goal) + // A successful tool output is real progress: it resolves any pending + // continuation and clears the prompt-failure counter. Failed tool outputs + // never reach this reset. + goal.continuationFailures = 0 + goal.pendingContinuationStart = null + goal.pendingContinuationStarted = false + goal.awaitingContinuationProgress = false + goal.noProgressTurns = 0 + goal.updatedAt = nowSeconds() + return snapshot(goal) + }) +} + function reserveWrapup(goal: Goal) { if (goal.budgetWrapupSent) return null goal.budgetWrapupSent = true diff --git a/test/server.test.ts b/test/server.test.ts index c4bb71f..f30732a 100644 --- a/test/server.test.ts +++ b/test/server.test.ts @@ -1,8 +1,13 @@ import { afterEach, beforeEach, expect, setSystemTime, test } from "bun:test" -import { mkdtemp, rm } from "node:fs/promises" +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises" import { join } from "node:path" import { tmpdir } from "node:os" import plugin from "../src/server" +import { + getGoal, + recordContinuationResult, + reserveContinuation, +} from "../src/state" function requireTool(tool: T | undefined, name: string): T { if (!tool) throw new Error(`expected ${name} to be registered`) @@ -18,6 +23,15 @@ async function waitFor(predicate: () => boolean) { expect(predicate()).toBe(true) } +async function waitForLong(predicate: () => boolean | Promise, deadlineMs = 3000) { + const deadline = Date.now() + deadlineMs + while (Date.now() < deadline) { + if (await predicate()) return + await new Promise((resolve) => setTimeout(resolve, 10)) + } + expect(await predicate()).toBe(true) +} + async function waitForContinuation(calls: unknown[]) { await waitFor(() => calls.length === 1) await new Promise((resolve) => setTimeout(resolve, 10)) @@ -274,6 +288,20 @@ OpenCode goal mode policy: usageContext, ) await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_usage" } } as never }) + // The continuation turn completes with a real assistant message, which + // resolves the pending continuation; the next idle then consumes the + // auto-turn limit and requests the wrap-up. + await hooks["experimental.chat.messages.transform"]!( + {}, + { + messages: [ + { + info: { id: "msg_usage_turn", role: "assistant", sessionID: "ses_usage" }, + parts: [{ type: "text", text: "USAGE_TURN_SHOULD_NOT_LEAK" }], + }, + ], + } as never, + ) await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_usage" } } as never }) const usageLimited = await requireTool(tools.get_goal, "get_goal").execute({}, usageContext) expect(String(usageLimited)).toContain('"status": "usageLimited"') @@ -616,7 +644,7 @@ test("turn watchdog retries a busy active goal without consuming continuation bu }, }, } as never, - { auto_continue: false, max_turn_time: 0.02, max_auto_turns: 1, max_prompt_failures: 1 }, + { auto_continue: false, max_turn_time: 0.02, max_auto_turns: 1, max_prompt_failures: 5 }, ) const tools = hooks.tool if (!tools) throw new Error("expected goal tools to be registered") @@ -638,11 +666,27 @@ test("turn watchdog retries a busy active goal without consuming continuation bu expect(String(read)).toContain('"autoTurns": 0') expect(String(read)).toContain('"continuationFailures": 0') expect(String(read)).toContain('"awaitingContinuationProgress": false') + // A watchdog-delivered prompt is already inside a busy episode, so the + // pending attempt is marked started immediately. + expect(String(read)).toContain('"pendingContinuationStarted": true') + + // The busy episode ends. Auto-continue is disabled here, so nothing further + // happens on idle; the watchdog-delivered attempt stays pending and started. + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + const afterIdle = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(afterIdle)).toContain('"status": "active"') + expect(String(afterIdle)).toContain('"autoTurns": 0') + expect(String(afterIdle)).toContain('"continuationFailures": 1') + expect(String(afterIdle)).toContain('"pendingContinuationStart": null') + // A new busy episode rescues again, still without auto-turn budgets. await hooks.event!({ event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "busy" } } } as never, }) await waitFor(() => calls.length === 2) + const final = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(final)).toContain('"status": "active"') + expect(String(final)).toContain('"autoTurns": 0') }) test("turn watchdog resets when another busy turn starts", async () => { @@ -787,7 +831,7 @@ test("turn watchdog does not inject while tasks are active, the goal is paused, expect(calls).toHaveLength(0) }) -test("turn watchdog transport failures do not pause or charge the goal", async () => { +test("turn watchdog transport failures share the prompt-failure ceiling without charging auto-turns", async () => { const logs: unknown[] = [] const hooks = await plugin.server( { @@ -800,7 +844,7 @@ test("turn watchdog transport failures do not pause or charge the goal", async ( }, }, } as never, - { auto_continue: false, max_turn_time: 0.02, max_prompt_failures: 1 }, + { auto_continue: false, max_turn_time: 0.02, max_prompt_failures: 2 }, ) const tools = hooks.tool if (!tools) throw new Error("expected goal tools to be registered") @@ -812,11 +856,32 @@ test("turn watchdog transport failures do not pause or charge the goal", async ( }) await waitFor(() => logs.length === 1) + const afterFirst = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(afterFirst)).toContain('"status": "active"') + expect(String(afterFirst)).toContain('"autoTurns": 0') + expect(String(afterFirst)).toContain('"continuationFailures": 1') + expect(JSON.stringify(logs[0])).toContain("Turn watchdog retry failed") + + // Duplicate busy notifications in the same episode cannot re-arm a failed + // rescue. + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "busy" } } } as never, + }) + await new Promise((resolve) => setTimeout(resolve, 80)) + expect(logs).toHaveLength(1) + + // A new busy episode gets one rescue; its failure reaches the ceiling. + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "busy" } } } as never, + }) + await waitFor(() => logs.length === 2) + const read = await requireTool(tools.get_goal, "get_goal").execute({}, context) - expect(String(read)).toContain('"status": "active"') + expect(String(read)).toContain('"status": "paused"') expect(String(read)).toContain('"autoTurns": 0') - expect(String(read)).toContain('"continuationFailures": 0') - expect(JSON.stringify(logs[0])).toContain("Turn watchdog retry failed") + expect(String(read)).toContain('"continuationFailures": 2') + expect(String(read)).toContain("Auto-continue prompt failed repeatedly") }) test("running task defers idle auto-continue", async () => { @@ -1662,3 +1727,808 @@ test("idle handler skips overlapping continuations for the same session", async expect(calls).toHaveLength(1) }) + +test("auto-continue retries are bounded: three failed attempts, no fourth", async () => { + const logs: unknown[] = [] + const hooks = await plugin.server( + { + client: { + app: { log: async (input: unknown) => logs.push(input) }, + session: { + promptAsync: async () => { + throw new Error("network down") + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 10, min_continue_interval_seconds: 0, max_prompt_failures: 3 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, { sessionID: "ses_1" } as never) + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + + await waitForLong(() => logs.length === 3) + await new Promise((resolve) => setTimeout(resolve, 300)) + + const read = await requireTool(tools.get_goal, "get_goal").execute({}, { sessionID: "ses_1" } as never) + expect(String(read)).toContain('"status": "paused"') + expect(String(read)).toContain('"continuationFailures": 3') + expect(String(read)).toContain('"autoTurns": 3') + expect(logs).toHaveLength(3) +}) + +test("failed continuation retries wait for the configured minimum interval", async () => { + const logs: unknown[] = [] + const hooks = await plugin.server( + { + client: { + app: { log: async (input: unknown) => logs.push(input) }, + session: { + promptAsync: async () => { + throw new Error("fetch failed") + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 1, max_prompt_failures: 2 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, { sessionID: "ses_1" } as never) + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + await waitFor(() => logs.length === 1) + + const early = await requireTool(tools.get_goal, "get_goal").execute({}, { sessionID: "ses_1" } as never) + expect(String(early)).toContain('"continuationFailures": 1') + + await new Promise((resolve) => setTimeout(resolve, 400)) + const beforeInterval = await requireTool(tools.get_goal, "get_goal").execute({}, { sessionID: "ses_1" } as never) + expect(String(beforeInterval)).toContain('"continuationFailures": 1') + + await waitForLong(() => logs.length === 2) + const read = await requireTool(tools.get_goal, "get_goal").execute({}, { sessionID: "ses_1" } as never) + expect(String(read)).toContain('"status": "paused"') + expect(String(read)).toContain('"continuationFailures": 2') +}) + +test("recognized transport error strings accumulate as continuation failures", async () => { + const errors = [ + "network down", + "fetch failed", + "ECONNRESET: connection reset by peer", + "request timed out", + "Cannot connect to API: The socket connection was closed unexpectedly.", + "Provider response headers timed out after 10000ms", + ] + for (const [index, message] of errors.entries()) { + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async () => { + throw new Error(message) + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 0, max_prompt_failures: 10 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + await requireTool(tools.create_goal, "create_goal").execute( + { objective: "keep going" }, + { sessionID: `ses_transport_${index}` } as never, + ) + await hooks.event!({ + event: { type: "session.idle", properties: { sessionID: `ses_transport_${index}` } } as never, + }) + + const read = await requireTool(tools.get_goal, "get_goal").execute({}, { sessionID: `ses_transport_${index}` } as never) + expect(String(read)).toContain('"continuationFailures": 1') + } +}) + +test("failed tool output does not reset prompt failures; successful tool output does", async () => { + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async () => {}, + }, + }, + } as never, + { auto_continue: false }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await recordContinuationResult("ses_1", "failure", 5) + await recordContinuationResult("ses_1", "success", 5) + expect((await getGoal("ses_1"))?.continuationFailures).toBe(1) + + await hooks["tool.execute.after"]!( + { tool: "bash", sessionID: "ses_1", callID: "call_1", args: {} } as never, + { title: "bash", output: "command not found", metadata: {} } as never, + ) + const afterFailedTool = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(afterFailedTool)).toContain('"continuationFailures": 1') + + await hooks["tool.execute.after"]!( + { tool: "bash", sessionID: "ses_1", callID: "call_2", args: {} } as never, + { title: "bash", output: "tests passed", metadata: {} } as never, + ) + const afterSuccessfulTool = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(afterSuccessfulTool)).toContain('"continuationFailures": 0') + expect(String(afterSuccessfulTool)).toContain('"pendingContinuationStart": null') +}) + +test("duplicate idle events before any busy never count a failure or send a duplicate", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 0, max_prompt_failures: 1 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + // Paired duplicate idle: the continuation prompt was delivered but no busy + // event has marked it started, so it must neither count an unresolved + // failure nor send a second prompt. + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "idle" } } } as never, + }) + + expect(calls).toHaveLength(1) + const read = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(read)).toContain('"status": "active"') + expect(String(read)).toContain('"continuationFailures": 0') + expect(String(read)).toContain('"pendingContinuationStarted": false') + expect(String(read)).not.toContain('"pendingContinuationStart": null') +}) + +test("paired idle events after a busy count exactly one unresolved failure and pause at the ceiling", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 0, max_prompt_failures: 1 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + expect(calls).toHaveLength(1) + + // The provider picks up the prompt: the busy event marks the attempt started. + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "busy" } } } as never, + }) + const started = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(started)).toContain('"pendingContinuationStarted": true') + + // The following logical idle has no substantive progress: exactly one + // unresolved failure is counted, which hits the ceiling and pauses. + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "idle" } } } as never, + }) + const afterIdle = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(afterIdle)).toContain('"status": "paused"') + expect(String(afterIdle)).toContain('"continuationFailures": 1') + expect(String(afterIdle)).toContain('"autoTurns": 1') + + // The paired session.idle duplicate must not double-count or double-send. + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + const final = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(final)).toContain('"continuationFailures": 1') + expect(calls).toHaveLength(1) +}) + +test("concurrent session.error transport events count at most one failure per pending attempt", async () => { + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async () => {}, + }, + }, + } as never, + { auto_continue: false, max_prompt_failures: 3 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await reserveContinuation("ses_1", 10, 0) + await recordContinuationResult("ses_1", "success", 3) + expect((await getGoal("ses_1"))?.pendingContinuationStart).not.toBeNull() + + const transportEvent = { + event: { + type: "session.error", + properties: { + sessionID: "ses_1", + error: { name: "AI_APICallError", message: "Cannot connect to API: The socket connection was closed unexpectedly." }, + }, + } as never, + } + await Promise.all([hooks.event!(transportEvent), hooks.event!(transportEvent)]) + const afterFirst = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(afterFirst)).toContain('"status": "active"') + expect(String(afterFirst)).toContain('"continuationFailures": 1') + + // With no pending attempt left, duplicate transport events must not + // increment the counter repeatedly. + await hooks.event!({ + event: { + type: "session.error", + properties: { + sessionID: "ses_1", + error: { name: "ProviderHeaderTimeoutError", message: "Provider response headers timed out after 10000ms" }, + }, + } as never, + }) + await hooks.event!({ + event: { + type: "session.error", + properties: { + sessionID: "ses_1", + error: { name: "ProviderHeaderTimeoutError", message: "Provider response headers timed out after 10000ms" }, + }, + } as never, + }) + const final = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(final)).toContain('"continuationFailures": 1') +}) + +test("auto_continue false never schedules a retry after a transport event", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: false, min_continue_interval_seconds: 0, max_prompt_failures: 3 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_no_auto" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await reserveContinuation("ses_no_auto", 10, 0) + await recordContinuationResult("ses_no_auto", "success", 3) + await hooks.event!({ + event: { + type: "session.error", + properties: { sessionID: "ses_no_auto", error: { message: "network connection failed" } }, + } as never, + }) + + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(calls).toHaveLength(0) + expect((await getGoal("ses_no_auto"))?.continuationFailures).toBe(1) +}) + +test("a repeated old assistant message cannot hide a no-response failure", async () => { + const calls: unknown[] = [] + const oldAssistant = { + info: { id: "msg_old", role: "assistant", sessionID: "ses_old_message" }, + parts: [{ type: "text", text: "Earlier progress" }], + } + const hooks = await plugin.server( + { + client: { + session: { + messages: async () => ({ data: [oldAssistant] }), + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, min_continue_interval_seconds: 0, max_prompt_failures: 1 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + const context = { sessionID: "ses_old_message" } as never + + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_old_message" } } as never }) + expect(calls).toHaveLength(1) + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_old_message", status: { type: "busy" } } } as never, + }) + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_old_message", status: { type: "idle" } } } as never, + }) + + const result = await getGoal("ses_old_message") + expect(result?.status).toBe("paused") + expect(result?.continuationFailures).toBe(1) + expect(calls).toHaveLength(1) +}) + +test("non-transport prompt errors preserve failure accounting without automatic retry", async () => { + const logs: unknown[] = [] + let calls = 0 + const hooks = await plugin.server( + { + client: { + app: { log: async (input: unknown) => logs.push(input) }, + session: { + promptAsync: async () => { + calls += 1 + throw new Error("invalid provider configuration") + }, + }, + }, + } as never, + { auto_continue: true, min_continue_interval_seconds: 0, max_prompt_failures: 3 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + const context = { sessionID: "ses_non_transport" } as never + + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_non_transport" } } as never }) + await new Promise((resolve) => setTimeout(resolve, 100)) + + expect(calls).toBe(1) + expect(logs).toHaveLength(1) + expect((await getGoal("ses_non_transport"))?.continuationFailures).toBe(1) +}) + +test("session.error without a pending attempt schedules recovery without a phantom failure", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 0, max_prompt_failures: 3 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await hooks.event!({ + event: { + type: "session.error", + properties: { + sessionID: "ses_1", + error: { name: "AI_APICallError", message: "Cannot connect to API: The socket connection was closed unexpectedly." }, + }, + } as never, + }) + + const afterError = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(afterError)).toContain('"status": "active"') + expect(String(afterError)).toContain('"continuationFailures": 0') + + // The first bounded automatic recovery starts without charging a failure. + await waitForLong(() => calls.length === 1) + const recovered = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(recovered)).toContain('"status": "active"') + expect(String(recovered)).toContain('"continuationFailures": 0') + expect(String(recovered)).toContain('"autoTurns": 1') +}) + +test("restart resolves a persisted started pending attempt at the next idle", async () => { + const firstCalls: unknown[] = [] + const hooks1 = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + firstCalls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 0, max_prompt_failures: 2 }, + ) + const tools1 = hooks1.tool + if (!tools1) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools1.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await hooks1.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + await waitFor(() => firstCalls.length === 1) + await hooks1.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "busy" } } } as never, + }) + await hooks1.dispose?.() + + // A fresh instance reads the same persisted state: the started=true pending + // attempt must be resolvable by the next idle after the restart. + const calls2: unknown[] = [] + const hooks2 = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls2.push(input) + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 0, max_prompt_failures: 2 }, + ) + const tools2 = hooks2.tool + if (!tools2) throw new Error("expected goal tools to be registered") + + await hooks2.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + const read = await requireTool(tools2.get_goal, "get_goal").execute({}, context) + expect(String(read)).toContain('"continuationFailures": 1') + + // The bounded retry then sends the next continuation attempt. + await waitForLong(() => calls2.length === 1) + const retried = await requireTool(tools2.get_goal, "get_goal").execute({}, context) + expect(String(retried)).toContain('"continuationFailures": 1') +}) + +test("persisted started=false pending attempts go stale after restart", async () => { + const hooks1 = await plugin.server( + { + client: { + session: { + promptAsync: async () => {}, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 0, max_prompt_failures: 2 }, + ) + const tools1 = hooks1.tool + if (!tools1) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools1.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await reserveContinuation("ses_1", 10, 0) + await recordContinuationResult("ses_1", "success", 5) + expect((await getGoal("ses_1"))?.pendingContinuationStarted).toBe(false) + + // Simulate an old persisted attempt by writing a stale millisecond timestamp + // directly into the state file instead of waiting 30 seconds. + const file = process.env.OPENCODE_GOAL_STATE_PATH! + const state = JSON.parse(await readFile(file, "utf8")) + state.goals.ses_1.pendingContinuationStart = Date.now() - 60_000 + state.goals.ses_1.pendingContinuationStarted = false + await writeFile(file, JSON.stringify(state)) + await hooks1.dispose?.() + + const calls: unknown[] = [] + const hooks2 = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, max_auto_turns: 5, min_continue_interval_seconds: 0, max_prompt_failures: 2 }, + ) + const tools2 = hooks2.tool + if (!tools2) throw new Error("expected goal tools to be registered") + + await hooks2.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + const stale = await requireTool(tools2.get_goal, "get_goal").execute({}, context) + expect(String(stale)).toContain('"status": "active"') + expect(String(stale)).toContain('"continuationFailures": 1') + + // The stale attempt triggers a bounded retry rather than wedging forever. + await waitForLong(() => calls.length === 1) +}) + +test("a locally delivered unstarted attempt never becomes a false no-response failure", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, min_continue_interval_seconds: 0, max_prompt_failures: 1 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + const context = { sessionID: "ses_local_pending" } as never + + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_local_pending" } } as never }) + expect(calls).toHaveLength(1) + + const file = process.env.OPENCODE_GOAL_STATE_PATH! + const state = JSON.parse(await readFile(file, "utf8")) + state.goals.ses_local_pending.pendingContinuationStart = Date.now() - 60_000 + await writeFile(file, JSON.stringify(state)) + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_local_pending", status: { type: "idle" } } } as never, + }) + + const goal = await getGoal("ses_local_pending") + expect(goal?.status).toBe("active") + expect(goal?.continuationFailures).toBe(0) + expect(calls).toHaveLength(1) +}) + +test("a built-in retry status cancels scheduled transport recovery", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, min_continue_interval_seconds: 0 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + await requireTool(tools.create_goal, "create_goal").execute( + { objective: "keep going" }, + { sessionID: "ses_native_retry" } as never, + ) + + await hooks.event!({ + event: { + type: "session.error", + properties: { sessionID: "ses_native_retry", error: { message: "network connection failed" } }, + } as never, + }) + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_native_retry", status: { type: "retry" } } } as never, + }) + await new Promise((resolve) => setTimeout(resolve, 100)) + + expect(calls).toHaveLength(0) + expect((await getGoal("ses_native_retry"))?.continuationFailures).toBe(0) +}) + +test("assistant progress cancels no-pending transport recovery", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, min_continue_interval_seconds: 0 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + const context = { sessionID: "ses_progress_recovery" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + + await hooks.event!({ + event: { + type: "session.error", + properties: { sessionID: "ses_progress_recovery", error: { message: "network connection failed" } }, + } as never, + }) + await hooks.event!({ + event: { + type: "message.updated", + properties: { + sessionID: "ses_progress_recovery", + message: { + info: { id: "msg_recovered", role: "assistant", sessionID: "ses_progress_recovery" }, + parts: [{ type: "text", text: "The provider recovered without plugin intervention." }], + }, + }, + } as never, + }) + await new Promise((resolve) => setTimeout(resolve, 100)) + + expect(calls).toHaveLength(0) +}) + +test("successful tool progress cancels no-pending transport recovery", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, min_continue_interval_seconds: 0 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + const context = { sessionID: "ses_tool_recovery" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + + await hooks.event!({ + event: { + type: "session.error", + properties: { sessionID: "ses_tool_recovery", error: { message: "network connection failed" } }, + } as never, + }) + await hooks["tool.execute.after"]!( + { tool: "bash", sessionID: "ses_tool_recovery", callID: "call_progress", args: {} } as never, + { title: "bash", output: "tests passed", metadata: {} } as never, + ) + await new Promise((resolve) => setTimeout(resolve, 100)) + + expect(calls).toHaveLength(0) +}) + +test("interrupted connection messages are not classified as transport recovery", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: true, min_continue_interval_seconds: 0 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + await requireTool(tools.create_goal, "create_goal").execute( + { objective: "keep going" }, + { sessionID: "ses_interrupted" } as never, + ) + + await hooks.event!({ + event: { + type: "session.error", + properties: { sessionID: "ses_interrupted", error: { message: "socket connection interrupted by user" } }, + } as never, + }) + await new Promise((resolve) => setTimeout(resolve, 100)) + + expect(calls).toHaveLength(0) + expect((await getGoal("ses_interrupted"))?.continuationFailures).toBe(0) +}) + +test("tool progress honors completed states and never resets on failed or incomplete tools", async () => { + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async () => {}, + }, + }, + } as never, + { auto_continue: false }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await recordContinuationResult("ses_1", "failure", 5) + expect((await getGoal("ses_1"))?.continuationFailures).toBe(1) + + const fireTool = (output: unknown) => + hooks["tool.execute.after"]!( + { tool: "bash", sessionID: "ses_1", callID: `call_${Math.random()}`, args: {} } as never, + output as never, + ) + + await hooks["tool.execute.after"]!( + { tool: "get_goal", sessionID: "ses_1", callID: "call_get_goal", args: {} } as never, + { title: "get_goal", output: '{"goal":{"status":"active"}}', metadata: {} } as never, + ) + expect((await getGoal("ses_1"))?.continuationFailures).toBe(1) + + // Incomplete, failed, cancelled, and aborted states must not reset even + // without an error string. + await fireTool({ title: "bash", output: "task_id: t1\nstate: running", metadata: {} }) + await fireTool({ title: "bash", output: "task_id: t1\nstate: failed", metadata: {} }) + await fireTool({ title: "bash", output: "task_id: t1\nstate: cancelled", metadata: {} }) + await fireTool({ title: "task", output: 'failed', metadata: {} }) + await fireTool({ title: "bash", output: "nope", state: "aborted", metadata: {} }) + await fireTool({ title: "bash", output: "nope", status: "running", metadata: {} }) + await fireTool({ title: "bash", output: "nope", success: false, metadata: {} }) + const unchanged = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(unchanged)).toContain('"continuationFailures": 1') + + // Completed and plain successful outputs reset the failure counter. + await fireTool({ title: "bash", output: "task_id: t1\nstate: completed\n\ndone", metadata: {} }) + const completed = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(completed)).toContain('"continuationFailures": 0') + + await fireTool({ title: "bash", output: "tests passed", metadata: {} }) + const plainSuccess = await requireTool(tools.get_goal, "get_goal").execute({}, context) + expect(String(plainSuccess)).toContain('"continuationFailures": 0') +}) + +test("watchdog rescues at most once per busy episode", async () => { + const calls: unknown[] = [] + const hooks = await plugin.server( + { + client: { + session: { + promptAsync: async (input: unknown) => { + calls.push(input) + }, + }, + }, + } as never, + { auto_continue: false, max_turn_time: 0.02, max_prompt_failures: 5 }, + ) + const tools = hooks.tool + if (!tools) throw new Error("expected goal tools to be registered") + + const context = { sessionID: "ses_1" } as never + await requireTool(tools.create_goal, "create_goal").execute({ objective: "keep going" }, context) + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "busy" } } } as never, + }) + await waitForContinuation(calls) + expect(calls).toHaveLength(1) + + // Another busy event inside the same episode must not re-arm the watchdog. + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "busy" } } } as never, + }) + await new Promise((resolve) => setTimeout(resolve, 80)) + expect(calls).toHaveLength(1) + + // Ending the episode and starting a new one rescues again. + await hooks.event!({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } as never }) + await hooks.event!({ + event: { type: "session.status", properties: { sessionID: "ses_1", status: { type: "busy" } } } as never, + }) + await waitFor(() => calls.length === 2) + expect(calls).toHaveLength(2) +}) diff --git a/test/state.test.ts b/test/state.test.ts index 8e25084..84eb214 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -7,12 +7,14 @@ import { clearGoal, completeGoal, createGoal, + markPendingContinuationStarted, recordAssistantProgress, getGoal, markGoalUnmet, pauseGoalForPlanMode, recordContinuationResult, recordPromptAgent, + recordToolProgress, reserveContinuation, setGoalStatus, updateGoalObjective, @@ -279,3 +281,160 @@ test("does not overwrite corrupt persisted state", async () => { expect(await readFile(process.env.OPENCODE_GOAL_STATE_PATH!, "utf8")).toBe("{not valid json") }) + +test("prompt delivery arms the pending window but never resets the failure count", async () => { + await createGoal("ses_1", "keep going", null) + await reserveContinuation("ses_1", 10, 0) + await recordContinuationResult("ses_1", "failure", 5) + await recordContinuationResult("ses_1", "failure", 5) + + const delivered = await recordContinuationResult("ses_1", "success", 5) + expect(delivered?.continuationFailures).toBe(2) + expect(delivered?.pendingContinuationStart).not.toBeNull() + expect(delivered?.pendingContinuationStarted).toBe(false) + expect(delivered?.awaitingContinuationProgress).toBe(true) + + const failed = await recordContinuationResult("ses_1", "failure", 5) + expect(failed?.continuationFailures).toBe(3) + expect(failed?.pendingContinuationStart).toBeNull() + expect(failed?.pendingContinuationStarted).toBe(false) + expect(failed?.awaitingContinuationProgress).toBe(false) +}) + +test("a session busy event marks the pending attempt as started", async () => { + await createGoal("ses_1", "keep going", null) + await reserveContinuation("ses_1", 10, 0) + await recordContinuationResult("ses_1", "success", 5) + expect((await getGoal("ses_1"))?.pendingContinuationStarted).toBe(false) + + const started = await markPendingContinuationStarted("ses_1") + expect(started?.pendingContinuationStarted).toBe(true) + + // Marking an already-started or absent attempt is idempotent. + const again = await markPendingContinuationStarted("ses_1") + expect(again?.pendingContinuationStarted).toBe(true) + await recordContinuationResult("ses_1", "failure", 5) + expect((await markPendingContinuationStarted("ses_1"))?.pendingContinuationStart).toBeNull() +}) + +test("persists continuation failures and the pending window across restart", async () => { + await createGoal("ses_1", "keep going", null) + await reserveContinuation("ses_1", 10, 0) + await recordContinuationResult("ses_1", "success", 5) + await recordContinuationResult("ses_1", "failure", 5) + await recordContinuationResult("ses_1", "failure", 5) + + // getGoal re-reads the persisted state file, simulating a process restart. + const reloaded = await getGoal("ses_1") + expect(reloaded?.continuationFailures).toBe(2) + expect(reloaded?.pendingContinuationStart).toBeNull() + expect(reloaded?.pendingContinuationStarted).toBe(false) + + await recordContinuationResult("ses_1", "success", 5) + const reloadedPending = await getGoal("ses_1") + expect(reloadedPending?.continuationFailures).toBe(2) + expect(reloadedPending?.pendingContinuationStart).toBeGreaterThanOrEqual(Date.now() - 5_000) + expect(reloadedPending?.pendingContinuationStarted).toBe(false) + + await markPendingContinuationStarted("ses_1") + const reloadedStarted = await getGoal("ses_1") + expect(reloadedStarted?.pendingContinuationStarted).toBe(true) + expect(reloadedStarted?.pendingContinuationStart).not.toBeNull() +}) + +test("decodes persisted state that lacks the retry fields", async () => { + await writeFile( + process.env.OPENCODE_GOAL_STATE_PATH!, + JSON.stringify({ + version: 1, + goals: { + ses_1: { + sessionID: "ses_1", + objective: "continue", + status: "active", + tokenBudget: null, + tokensUsed: 0, + timeUsedSeconds: 0, + createdAt: 1, + updatedAt: 1, + lastAccountedAt: 1, + autoTurns: 0, + lastContinuationAt: null, + }, + }, + }), + ) + + const goal = await getGoal("ses_1") + + expect(goal?.continuationFailures).toBe(0) + expect(goal?.pendingContinuationStart).toBeNull() + expect(goal?.pendingContinuationStarted).toBe(false) +}) + +test("substantive assistant text resets the failure count and pending window", async () => { + await createGoal("ses_1", "keep going", null) + await reserveContinuation("ses_1", 10, 0) + await recordContinuationResult("ses_1", "failure", 5) + await recordContinuationResult("ses_1", "failure", 5) + await recordContinuationResult("ses_1", "success", 5) + expect((await getGoal("ses_1"))?.pendingContinuationStart).not.toBeNull() + + const progressed = await recordAssistantProgress("ses_1", { + messageID: "m1", + text: "Implemented the parser and added passing tests", + outputTokens: 400, + }) + + expect(progressed?.continuationFailures).toBe(0) + expect(progressed?.pendingContinuationStart).toBeNull() + expect(progressed?.status).toBe("active") +}) + +test("successful tool output resets failures and clears the pending window", async () => { + await createGoal("ses_1", "keep going", null) + await recordContinuationResult("ses_1", "failure", 5) + await recordContinuationResult("ses_1", "success", 5) + expect((await getGoal("ses_1"))?.continuationFailures).toBe(1) + + const progressed = await recordToolProgress("ses_1", "tests passed") + + expect(progressed?.continuationFailures).toBe(0) + expect(progressed?.pendingContinuationStart).toBeNull() + expect(progressed?.awaitingContinuationProgress).toBe(false) +}) + +test("re-reading the previous assistant message does not resolve a pending continuation", async () => { + await createGoal("ses_1", "keep going", null) + await recordAssistantProgress("ses_1", { messageID: "m1", text: "Initial progress" }) + await reserveContinuation("ses_1", 10, 0) + await recordContinuationResult("ses_1", "success", 5) + + const repeated = await recordAssistantProgress("ses_1", { + messageID: "m1", + text: "Initial progress", + evaluateContinuation: true, + }) + + expect(repeated?.pendingContinuationStart).not.toBeNull() + expect(repeated?.awaitingContinuationProgress).toBe(true) +}) + +test("lastContinuationAt remains a public seconds timestamp", async () => { + await createGoal("ses_1", "keep going", null) + const reserved = await reserveContinuation("ses_1", 10, 0) + + expect(reserved?.lastContinuationAt).toBe(Math.floor(Date.now() / 1000)) + expect(reserved?.lastContinuationAt).toBeLessThan(1_000_000_000_000) +}) + +test("resuming a paused goal clears the failure count and pending window", async () => { + await createGoal("ses_1", "keep going", null) + await recordContinuationResult("ses_1", "failure", 1) + expect((await getGoal("ses_1"))?.status).toBe("paused") + + const resumed = await setGoalStatus("ses_1", "active") + + expect(resumed?.continuationFailures).toBe(0) + expect(resumed?.pendingContinuationStart).toBeNull() +})