diff --git a/src/agents/subagents/announce/subagent-announce-output.ts b/src/agents/subagents/announce/subagent-announce-output.ts index 382df235ad7f..ded441a5043a 100644 --- a/src/agents/subagents/announce/subagent-announce-output.ts +++ b/src/agents/subagents/announce/subagent-announce-output.ts @@ -10,10 +10,7 @@ import type { SessionTranscriptRuntimeTarget } from "../../../config/sessions/se import { resolveFreshSessionTotalTokens } from "../../../config/sessions/types.js"; import { isFastTestRuntimeEnv } from "../../../infra/env.js"; import { formatDurationCompact } from "../../../infra/format-time/format-duration.js"; -import { - buildAgentRunTerminalOutcomeFromWaitResult, - classifyAgentRunTerminalOutcome, -} from "../../agent-run-terminal-outcome.js"; +import { buildAgentRunTerminalOutcomeFromWaitResult } from "../../agent-run-terminal-outcome.js"; import { wrapPromptDataBlock } from "../../sanitize-for-prompt.js"; import { extractStoredAssistantText, sanitizeTextContent } from "../../tools/chat-history-text.js"; import { @@ -21,6 +18,7 @@ import { selectDeliverableSessionsReply, } from "../../tools/sessions-send-tokens.js"; import { compareSubagentRunGeneration } from "../registry/subagent-run-generation.js"; +import { classifySubagentTerminalOutcome } from "../subagent-terminal-outcome.js"; import { captureSubagentCompletionReplyUsing, readLatestSubagentOutputWithRetryUsing, @@ -356,7 +354,7 @@ export function applySubagentWaitOutcome(params: { // primary normalizers, so apply the canonical classification here instead // of re-enumerating reason groups. if (terminalOutcome) { - switch (classifyAgentRunTerminalOutcome(terminalOutcome)) { + switch (classifySubagentTerminalOutcome(terminalOutcome)) { case "timeout": outcome = { status: "timeout" }; break; diff --git a/src/agents/subagents/registry/subagent-registry-listener.ts b/src/agents/subagents/registry/subagent-registry-listener.ts index 84880a0e2523..c12f5fe0cf2b 100644 --- a/src/agents/subagents/registry/subagent-registry-listener.ts +++ b/src/agents/subagents/registry/subagent-registry-listener.ts @@ -1,13 +1,8 @@ import type { AgentEventPayload } from "../../../infra/agent-events.js"; import { runWithGatewayIndependentRootWorkAdmission } from "../../../process/gateway-work-admission.js"; -import { - formatAbandonedLivenessError, - formatBlockedLivenessError, - isAbandonedLivenessState, - isBlockedLivenessState, -} from "../../../shared/agent-liveness.js"; +import { buildAgentRunTerminalOutcomeFromLifecycleEvent } from "../../agent-run-terminal-outcome.js"; import { normalizeAgentRunTerminalReplySnapshot } from "../../agent-run-terminal-reply.js"; -import { isAbortedAgentStopReason } from "../../run-termination.js"; +import { classifySubagentTerminalOutcome } from "../subagent-terminal-outcome.js"; import { SUBAGENT_ENDED_REASON_COMPLETE, SUBAGENT_ENDED_REASON_ERROR, @@ -82,11 +77,6 @@ export function createSubagentRegistryListener(config: { } const endedAt = typeof evt.data?.endedAt === "number" ? evt.data.endedAt : Date.now(); const startedAt = typeof evt.data?.startedAt === "number" ? evt.data.startedAt : undefined; - const error = typeof evt.data?.error === "string" ? evt.data.error : undefined; - const livenessState = - typeof evt.data?.livenessState === "string" ? evt.data.livenessState : undefined; - const stopReason = - typeof evt.data?.stopReason === "string" ? evt.data.stopReason : undefined; const terminalReply = normalizeAgentRunTerminalReplySnapshot(evt.data?.terminalReply); // sessions_yield ends the turn by aborting the run signal, so a yielded // terminal can also look aborted. An explicit yield is authoritative — pause, @@ -106,64 +96,29 @@ export function createSubagentRegistryListener(config: { } return; } - if (isAbortedAgentStopReason(stopReason)) { - pendingLifecycle.clear(evt.runId); - await completeSubagentRunWithRecovery( - { - runId: evt.runId, - endedAt, - outcome: { - status: "error", - error: "subagent run terminated", - }, - reason: SUBAGENT_ENDED_REASON_KILLED, - sendFarewell: true, - accountId: entry.requesterOrigin?.accountId, - triggerCleanup: true, - startedAt, - terminalReply, - }, - "lifecycle-killed-event", - ); - return; - } - if (phase === "error") { - pendingLifecycle.scheduleError({ + const terminalOutcome = buildAgentRunTerminalOutcomeFromLifecycleEvent({ + phase, + data: evt.data, + startedAt, + endedAt, + }); + const classification = classifySubagentTerminalOutcome(terminalOutcome); + if ( + classification === "cancellation" && + evt.data?.aborted === true && + evt.data.stopReason === undefined && + evt.data.status === undefined && + evt.data.timeoutPhase === undefined + ) { + pendingLifecycle.scheduleCancellation({ runId: evt.runId, endedAt, startedAt, terminalReply, - error, }); return; } - const blocked = isBlockedLivenessState(livenessState); - const abandoned = isAbandonedLivenessState(livenessState); - if (blocked || abandoned) { - pendingLifecycle.clear(evt.runId); - const blockedParams = { - runId: evt.runId, - endedAt, - outcome: { - status: "error" as const, - error: blocked - ? formatBlockedLivenessError(error) - : formatAbandonedLivenessError(error), - }, - reason: SUBAGENT_ENDED_REASON_ERROR, - sendFarewell: true, - accountId: entry.requesterOrigin?.accountId, - triggerCleanup: true, - startedAt, - terminalReply, - }; - await completeSubagentRunWithRecovery( - blockedParams, - blocked ? "lifecycle-blocked-event" : "lifecycle-abandoned-event", - ); - return; - } - if (evt.data?.aborted) { + if (classification === "timeout") { pendingLifecycle.scheduleTimeout({ runId: evt.runId, endedAt, @@ -172,6 +127,38 @@ export function createSubagentRegistryListener(config: { }); return; } + if (phase === "error" && classification === "failure") { + pendingLifecycle.scheduleError({ + runId: evt.runId, + endedAt, + startedAt, + terminalReply, + error: terminalOutcome.error, + }); + return; + } + if (classification !== "success") { + const cancelled = classification === "cancellation"; + pendingLifecycle.clear(evt.runId); + await completeSubagentRunWithRecovery( + { + runId: evt.runId, + endedAt, + outcome: { + status: "error" as const, + error: cancelled ? "subagent run terminated" : terminalOutcome.error, + }, + reason: cancelled ? SUBAGENT_ENDED_REASON_KILLED : SUBAGENT_ENDED_REASON_ERROR, + sendFarewell: true, + accountId: entry.requesterOrigin?.accountId, + triggerCleanup: true, + startedAt, + terminalReply, + }, + cancelled ? "lifecycle-killed-event" : `lifecycle-${terminalOutcome.reason}-event`, + ); + return; + } pendingLifecycle.clear(evt.runId); const completionParams = { runId: evt.runId, diff --git a/src/agents/subagents/registry/subagent-registry-pending-lifecycle.ts b/src/agents/subagents/registry/subagent-registry-pending-lifecycle.ts index a75c6e08298f..6d9658f964be 100644 --- a/src/agents/subagents/registry/subagent-registry-pending-lifecycle.ts +++ b/src/agents/subagents/registry/subagent-registry-pending-lifecycle.ts @@ -1,23 +1,29 @@ +import { AGENT_RUN_TERMINAL_RETRY_GRACE_MS } from "../../agent-run-terminal-outcome.js"; import { SUBAGENT_ENDED_REASON_COMPLETE, SUBAGENT_ENDED_REASON_ERROR, + SUBAGENT_ENDED_REASON_KILLED, } from "./subagent-lifecycle-events.js"; import type { SubagentCompletionRequest, SubagentRunRecord } from "./subagent-registry.types.js"; -const LIFECYCLE_RETRY_GRACE_MS = 15_000; const PENDING_LIFECYCLE_TERMINAL_TTL_MS = 5 * 60_000; type PendingLifecycleKind = "error" | "timeout"; -type PendingLifecycleTerminal = { - kind: PendingLifecycleKind; - timer: NodeJS.Timeout; +type PendingLifecycleParams = { + runId: string; endedAt: number; startedAt?: number; + cancellation?: true; error?: string; terminalReply?: SubagentCompletionRequest["terminalReply"]; }; +type PendingLifecycleTerminal = PendingLifecycleParams & { + kind: PendingLifecycleKind; + timer: NodeJS.Timeout; +}; + export function createPendingLifecycleScheduler(params: { runs: Map; completeInBackground: (completion: SubagentCompletionRequest, source: string) => void; @@ -38,16 +44,7 @@ export function createPendingLifecycleScheduler(params: { pendingByRunId.clear(); } - function schedule( - kind: PendingLifecycleKind, - scheduleParams: { - runId: string; - endedAt: number; - startedAt?: number; - error?: string; - terminalReply?: SubagentCompletionRequest["terminalReply"]; - }, - ) { + function schedule(kind: PendingLifecycleKind, scheduleParams: PendingLifecycleParams) { clearKind(scheduleParams.runId); const timer = setTimeout(() => { const pending = pendingByRunId.get(scheduleParams.runId); @@ -72,17 +69,26 @@ export function createPendingLifecycleScheduler(params: { runId: scheduleParams.runId, endedAt: pending.endedAt, outcome: - kind === "error" ? { status: "error", error: pending.error } : { status: "timeout" }, - reason: kind === "error" ? SUBAGENT_ENDED_REASON_ERROR : SUBAGENT_ENDED_REASON_COMPLETE, + kind === "timeout" + ? { status: "timeout" } + : { + status: "error", + error: pending.cancellation ? "subagent run terminated" : pending.error, + }, + reason: pending.cancellation + ? SUBAGENT_ENDED_REASON_KILLED + : kind === "error" + ? SUBAGENT_ENDED_REASON_ERROR + : SUBAGENT_ENDED_REASON_COMPLETE, sendFarewell: true, accountId: entry.requesterOrigin?.accountId, triggerCleanup: true, startedAt: pending.startedAt, terminalReply: pending.terminalReply, }, - `lifecycle-${kind}-grace`, + pending.cancellation ? "lifecycle-cancellation-grace" : `lifecycle-${kind}-grace`, ); - }, LIFECYCLE_RETRY_GRACE_MS); + }, AGENT_RUN_TERMINAL_RETRY_GRACE_MS); timer.unref?.(); pendingByRunId.set(scheduleParams.runId, { ...scheduleParams, kind, timer }); } @@ -92,6 +98,8 @@ export function createPendingLifecycleScheduler(params: { clearError: (runId: string) => clearKind(runId, "error"), clearTimeout: (runId: string) => clearKind(runId, "timeout"), clearAll, + scheduleCancellation: (scheduleParams: Parameters[1]) => + schedule("error", { ...scheduleParams, cancellation: true }), scheduleError: (scheduleParams: Parameters[1]) => schedule("error", scheduleParams), scheduleTimeout: (scheduleParams: Parameters[1]) => diff --git a/src/agents/subagents/registry/subagent-registry-run-wait.ts b/src/agents/subagents/registry/subagent-registry-run-wait.ts index 75ae4c14e19c..793c527510b9 100644 --- a/src/agents/subagents/registry/subagent-registry-run-wait.ts +++ b/src/agents/subagents/registry/subagent-registry-run-wait.ts @@ -6,15 +6,13 @@ import { callGateway } from "../../../gateway/call.js"; import { isFastTestRuntimeEnv } from "../../../infra/env.js"; import { createSubsystemLogger } from "../../../logging/subsystem.js"; import type { DetachedTaskFindResult } from "../../../tasks/detached-task-runtime-contract.js"; -import { - buildAgentRunTerminalOutcomeFromWaitResult, - classifyAgentRunTerminalOutcome, -} from "../../agent-run-terminal-outcome.js"; +import { buildAgentRunTerminalOutcomeFromWaitResult } from "../../agent-run-terminal-outcome.js"; import { isRecoverableAgentWaitError, waitForAgentRun } from "../../run-wait.js"; import { type SubagentRunOutcome, withSubagentOutcomeTiming, } from "../announce/subagent-announce-output.js"; +import { classifySubagentTerminalOutcome } from "../subagent-terminal-outcome.js"; import { clearDeliveryState, ensureCompletionState } from "./subagent-delivery-state.js"; import { SUBAGENT_ENDED_REASON_COMPLETE, @@ -293,7 +291,7 @@ export class SubagentWaitManager { const waitBlocked = waitTerminalOutcome?.reason === "blocked"; const waitAborted = waitTerminalOutcome !== undefined && - classifyAgentRunTerminalOutcome(waitTerminalOutcome) === "cancellation"; + classifySubagentTerminalOutcome(waitTerminalOutcome) === "cancellation"; const waitStatus = waitTerminalOutcome?.status ?? wait.status; if (wait.yielded === true && waitStatus !== "timeout" && !waitBlocked) { this.options.clearPendingLifecycleError(runId); diff --git a/src/agents/subagents/registry/subagent-registry.lifecycle-retry-grace.e2e.test.ts b/src/agents/subagents/registry/subagent-registry.lifecycle-retry-grace.e2e.test.ts index b5a76c0bcbcb..ea4025ba6b8e 100644 --- a/src/agents/subagents/registry/subagent-registry.lifecycle-retry-grace.e2e.test.ts +++ b/src/agents/subagents/registry/subagent-registry.lifecycle-retry-grace.e2e.test.ts @@ -19,7 +19,11 @@ type LifecycleData = { endedAt?: number; aborted?: boolean; error?: string; + stopReason?: string; terminalReply?: AgentRunTerminalReplySnapshot; + status?: string; + timeoutPhase?: string; + providerStarted?: boolean; }; type LifecycleEvent = { stream?: string; @@ -840,28 +844,65 @@ describe("subagent registry lifecycle error grace", () => { expect(run.completion?.capturedAt).toBeTypeOf("number"); }); - it("completes with timeout status when aborted end event fires after grace window", async () => { - registerCompletionRun("run-timeout", "timeout", "timeout test"); - setAssistantOutput("agent:main:subagent:timeout", "Partial output before timeout"); + it("records a bare aborted end event as cancellation after retry grace", async () => { + registerCompletionRun("run-aborted", "aborted", "aborted test"); + setAssistantOutput("agent:main:subagent:aborted", "Partial output before cancellation"); - // Emit an end event with aborted=true which triggers the timeout grace path - emitLifecycleEvent("run-timeout", { + emitLifecycleEvent("run-aborted", { phase: "end", aborted: true, endedAt: 3_000, - } as LifecycleData & { aborted: boolean }); + }); await flushAsync(); + expect(getAgentCalls()).toHaveLength(0); + expect( + mod + .listSubagentRunsForRequester(MAIN_REQUESTER_SESSION_KEY) + .find((candidate) => candidate.runId === "run-aborted")?.execution.status, + ).toBe("running"); - // Advance past the lifecycle timeout retry grace window - await vi.advanceTimersByTimeAsync(30_000); + await vi.advanceTimersByTimeAsync(15_000); await flushAsync(); - await waitForAgentCallCount(1); - const run = mod .listSubagentRunsForRequester(MAIN_REQUESTER_SESSION_KEY) - .find((candidate) => candidate.runId === "run-timeout"); + .find((candidate) => candidate.runId === "run-aborted"); + expect(run).toMatchObject({ + endedReason: "subagent-killed", + execution: { outcome: { status: "error", error: "subagent run terminated" } }, + }); + expect(getAgentCalls()).toHaveLength(0); + }); + + it("announces a provider hard timeout from its canonical lifecycle metadata", async () => { + registerCompletionRun("run-provider-timeout", "provider-timeout", "provider timeout test"); + setAssistantOutput( + "agent:main:subagent:provider-timeout", + "Partial output before provider timeout", + ); + + emitLifecycleEvent("run-provider-timeout", { + phase: "end", + aborted: true, + stopReason: "restart", + status: "timeout", + timeoutPhase: "provider", + providerStarted: true, + endedAt: 3_000, + error: "provider timed out", + }); + await flushAsync(); + expect(getAgentCalls()).toHaveLength(0); + + await vi.advanceTimersByTimeAsync(30_000); + await flushAsync(); + await waitForAgentCallCount(1); + + expect(readFirstAnnounceOutcome()?.status).toBe("timeout"); + const run = mod + .listSubagentRunsForRequester(MAIN_REQUESTER_SESSION_KEY) + .find((candidate) => candidate.runId === "run-provider-timeout"); expect(run?.execution.outcome?.status).toBe("timeout"); }); @@ -869,12 +910,15 @@ describe("subagent registry lifecycle error grace", () => { registerCompletionRun("run-timeout-cancel", "timeout-cancel", "timeout cancel test"); setAssistantOutput("agent:main:subagent:timeout-cancel", "Final answer after recovery"); - // Emit an aborted end event (starts timeout grace) + // Emit a structured timeout terminal (starts timeout grace). emitLifecycleEvent("run-timeout-cancel", { phase: "end", aborted: true, + status: "timeout", + timeoutPhase: "provider", + providerStarted: true, endedAt: 4_000, - } as LifecycleData & { aborted: boolean }); + }); await flushAsync(); expect(getAgentCalls()).toHaveLength(0); diff --git a/src/agents/subagents/registry/subagent-registry.persistence.resume.test.ts b/src/agents/subagents/registry/subagent-registry.persistence.resume.test.ts index 162921eb0572..e86083242329 100644 --- a/src/agents/subagents/registry/subagent-registry.persistence.resume.test.ts +++ b/src/agents/subagents/registry/subagent-registry.persistence.resume.test.ts @@ -133,14 +133,19 @@ describe("subagent registry persistence resume", () => { }); }); - it("retries pending child delivery before a recovered requester-turn wake", async () => { + it.each([ + { label: "successful", status: "ok" as const }, + { label: "timed-out", status: "timeout" as const }, + ])("retries pending $label child delivery after restart", async ({ label, status }) => { tempStateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-subagent-")); const stateDir = tempStateDir; await withEnvAsync({ OPENCLAW_STATE_DIR: stateDir }, async () => { + const runId = `run-pending-${label}-delivery`; + const childSessionKey = `agent:main:subagent:pending-${label}-delivery`; const run: SubagentRunRecord = { - runId: "run-pending-delivery", + runId, requesterTurnRunId: "run-requester", - childSessionKey: "agent:main:subagent:pending-delivery", + childSessionKey, requesterSessionKey: "agent:main:main", requesterDisplayKey: "main", task: "deliver before waking requester", @@ -151,7 +156,7 @@ describe("subagent registry persistence resume", () => { status: "terminal", startedAt: 110, endedAt: 200, - outcome: { status: "ok" }, + outcome: { status }, }, expectsCompletionMessage: true, completion: { required: true, resultText: "done", capturedAt: 200 }, @@ -160,12 +165,12 @@ describe("subagent registry persistence resume", () => { payload: { requesterSessionKey: "agent:main:main", requesterDisplayKey: "main", - childSessionKey: "agent:main:subagent:pending-delivery", - childRunId: "run-pending-delivery", + childSessionKey, + childRunId: runId, task: "deliver before waking requester", startedAt: 110, endedAt: 200, - outcome: { status: "ok" }, + outcome: { status }, expectsCompletionMessage: true, }, }, @@ -176,8 +181,8 @@ describe("subagent registry persistence resume", () => { stateDir, agentId: "main", sessionKey: run.childSessionKey, - sessionId: "sess-pending-delivery", - defaultSessionId: "sess-pending-delivery", + sessionId: `sess-pending-${label}-delivery`, + defaultSessionId: `sess-pending-${label}-delivery`, }); mod.initSubagentRegistry(); @@ -188,8 +193,9 @@ describe("subagent registry persistence resume", () => { interval: 10, }); expect(announceSpy).toHaveBeenCalledWith( - expect.objectContaining({ childRunId: "run-pending-delivery" }), + expect.objectContaining({ childRunId: runId, outcome: { status } }), ); + expect(mod.getSubagentRunByRunId(runId)?.execution.outcome).toEqual({ status }); }); }); diff --git a/src/agents/subagents/subagent-terminal-outcome.test.ts b/src/agents/subagents/subagent-terminal-outcome.test.ts new file mode 100644 index 000000000000..2866d6739061 --- /dev/null +++ b/src/agents/subagents/subagent-terminal-outcome.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vitest"; +import { buildAgentRunTerminalOutcome } from "../agent-run-terminal-outcome.js"; +import { classifySubagentTerminalOutcome } from "./subagent-terminal-outcome.js"; + +describe("classifySubagentTerminalOutcome", () => { + it("preserves provider timeout attribution over a retained restart marker", () => { + const outcome = buildAgentRunTerminalOutcome({ + status: "timeout", + stopReason: "restart", + timeoutPhase: "provider", + providerStarted: true, + }); + + expect(classifySubagentTerminalOutcome(outcome)).toBe("timeout"); + }); + + it("applies restart cancellation over incomplete liveness projections", () => { + const outcome = buildAgentRunTerminalOutcome({ + status: "error", + stopReason: "restart", + livenessState: "blocked", + }); + + expect(classifySubagentTerminalOutcome(outcome)).toBe("cancellation"); + }); +}); diff --git a/src/agents/subagents/subagent-terminal-outcome.ts b/src/agents/subagents/subagent-terminal-outcome.ts new file mode 100644 index 000000000000..c8069f10fd7a --- /dev/null +++ b/src/agents/subagents/subagent-terminal-outcome.ts @@ -0,0 +1,13 @@ +import { + classifyAgentRunTerminalOutcome, + type AgentRunTerminalOutcome, +} from "../agent-run-terminal-outcome.js"; +import { isAbortedAgentStopReason } from "../run-termination.js"; + +/** Subagents apply explicit cancellation ownership after canonical timeout attribution. */ +export function classifySubagentTerminalOutcome(outcome: AgentRunTerminalOutcome) { + const classification = classifyAgentRunTerminalOutcome(outcome); + return classification === "timeout" || !isAbortedAgentStopReason(outcome.stopReason) + ? classification + : "cancellation"; +}