Merge 8466dedda4 into c9e3bb7ca4
This commit is contained in:
Josh Jameson
GitHub
commit
f241bc4c8d
|
|
@ -664,6 +664,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
|||
processGroupId?: number | null;
|
||||
processLossRetryCount?: number;
|
||||
runtimeMode?: "legacy" | "native";
|
||||
scheduledRetryAttempt?: number | null;
|
||||
scheduledRetryReason?: string | null;
|
||||
scheduledRetryAt?: Date | null;
|
||||
includeIssue?: boolean;
|
||||
runErrorCode?: string | null;
|
||||
runError?: string | null;
|
||||
|
|
@ -726,6 +729,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
|||
processGroupId: input?.processGroupId ?? null,
|
||||
processLossRetryCount: input?.processLossRetryCount ?? 0,
|
||||
...(input?.runtimeMode ? { runtimeMode: input.runtimeMode } : {}),
|
||||
scheduledRetryAttempt: input?.scheduledRetryAttempt ?? 0,
|
||||
scheduledRetryReason: input?.scheduledRetryReason ?? null,
|
||||
scheduledRetryAt: input?.scheduledRetryAt ?? null,
|
||||
errorCode: input?.runErrorCode ?? null,
|
||||
error: input?.runError ?? null,
|
||||
nextEventSeq: 2,
|
||||
|
|
@ -2447,8 +2453,364 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
|||
expect(checkoutReleasedIssue?.checkoutRunId).toBeNull();
|
||||
});
|
||||
|
||||
it("requires reconciliation for a lost monitor whose provider outcomes are unknown", async () => {
|
||||
it("schedules a null-environment process loss with diagnostics instead of stranding it", async () => {
|
||||
const { companyId, agentId, runId } = await seedRunFixture({
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [runId] });
|
||||
|
||||
const runs = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId)));
|
||||
const failed = runs.find((row) => row.id === runId);
|
||||
const retry = runs.find((row) => row.retryOfRunId === runId);
|
||||
|
||||
expect(failed?.resultJson).toMatchObject({
|
||||
environmentAllocationDiagnostic: {
|
||||
phase: "environment_selection",
|
||||
outcome: "failed",
|
||||
reasonCode: "no_environment_or_lease_recorded",
|
||||
},
|
||||
});
|
||||
expect(failed?.stderrExcerpt).toContain("[environment-allocation]");
|
||||
expect(retry).toMatchObject({
|
||||
status: "scheduled_retry",
|
||||
scheduledRetryAttempt: 1,
|
||||
scheduledRetryReason: "retry_transient_environment_failure",
|
||||
});
|
||||
expect(retry?.scheduledRetryAt?.getTime()).toBe((failed?.finishedAt?.getTime() ?? 0) + 60_000);
|
||||
expect(retry?.contextSnapshot).toMatchObject({
|
||||
wakeReason: "process_lost_environment_retry",
|
||||
retryReason: "retry_transient_environment_failure",
|
||||
retryOfRunId: runId,
|
||||
issueId: failed?.contextSnapshot?.issueId,
|
||||
});
|
||||
});
|
||||
|
||||
it("schedules attempt 2 of the null-environment ladder at 180s and attempt 3 at 540s", async () => {
|
||||
const attempt2 = await seedRunFixture({
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
scheduledRetryAttempt: 1,
|
||||
scheduledRetryReason: "retry_transient_environment_failure",
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [attempt2.runId] });
|
||||
|
||||
const attempt2Rows = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, attempt2.agentId));
|
||||
const attempt2Failed = attempt2Rows.find((row) => row.id === attempt2.runId);
|
||||
const attempt2Retry = attempt2Rows.find((row) => row.retryOfRunId === attempt2.runId);
|
||||
expect(attempt2Retry).toMatchObject({
|
||||
status: "scheduled_retry",
|
||||
scheduledRetryAttempt: 2,
|
||||
scheduledRetryReason: "retry_transient_environment_failure",
|
||||
});
|
||||
expect(attempt2Retry?.scheduledRetryAt?.getTime()).toBe(
|
||||
(attempt2Failed?.finishedAt?.getTime() ?? 0) + 180_000,
|
||||
);
|
||||
|
||||
const attempt3 = await seedRunFixture({
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
scheduledRetryAttempt: 2,
|
||||
scheduledRetryReason: "retry_transient_environment_failure",
|
||||
contextSnapshot: {
|
||||
wakeReason: "process_lost_environment_retry",
|
||||
retryReason: "retry_transient_environment_failure",
|
||||
retryOfRunId: attempt2.runId,
|
||||
},
|
||||
});
|
||||
|
||||
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [attempt3.runId] });
|
||||
|
||||
const attempt3Rows = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, attempt3.agentId));
|
||||
const attempt3Failed = attempt3Rows.find((row) => row.id === attempt3.runId);
|
||||
const attempt3Retry = attempt3Rows.find((row) => row.retryOfRunId === attempt3.runId);
|
||||
expect(attempt3Retry).toMatchObject({
|
||||
status: "scheduled_retry",
|
||||
scheduledRetryAttempt: 3,
|
||||
scheduledRetryReason: "retry_transient_environment_failure",
|
||||
});
|
||||
expect(attempt3Retry?.scheduledRetryAt?.getTime()).toBe(
|
||||
(attempt3Failed?.finishedAt?.getTime() ?? 0) + 540_000,
|
||||
);
|
||||
});
|
||||
|
||||
it("does not queue another null-environment retry past attempt 3 and emits a single exhaustion event", async () => {
|
||||
const { companyId, agentId, runId } = await seedRunFixture({
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
scheduledRetryAttempt: 3,
|
||||
scheduledRetryReason: "retry_transient_environment_failure",
|
||||
contextSnapshot: {
|
||||
wakeReason: "process_lost_environment_retry",
|
||||
retryReason: "retry_transient_environment_failure",
|
||||
retryOfRunId: "previous-run",
|
||||
},
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [runId] });
|
||||
|
||||
// After attempt 3, the bounded null-environment ladder must NOT queue another
|
||||
// retry on the same reason chain. The release/promote path may still spin up
|
||||
// an issue.continuation_recovery run as a separate auto-recovery attempt, but
|
||||
// it must NOT carry the null-environment retry reason.
|
||||
const retries = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.retryOfRunId, runId),
|
||||
));
|
||||
const nullEnvRetries = retries.filter(
|
||||
(row) => row.scheduledRetryReason === "retry_transient_environment_failure",
|
||||
);
|
||||
expect(nullEnvRetries).toHaveLength(0);
|
||||
|
||||
const exhaustionEvents = await db
|
||||
.select()
|
||||
.from(heartbeatRunEvents)
|
||||
.where(and(
|
||||
eq(heartbeatRunEvents.companyId, companyId),
|
||||
eq(heartbeatRunEvents.runId, runId),
|
||||
eq(heartbeatRunEvents.eventType, "lifecycle"),
|
||||
));
|
||||
const exhaustion = exhaustionEvents.find((event) =>
|
||||
typeof event.message === "string"
|
||||
&& event.message.includes("Bounded retry exhausted")
|
||||
&& (event.payload as Record<string, unknown> | null)?.retryReason === "retry_transient_environment_failure",
|
||||
);
|
||||
expect(exhaustion).toBeDefined();
|
||||
});
|
||||
|
||||
it("does not route a monitor-dispatch loss through the null-environment ladder", async () => {
|
||||
// The fingerprint (no pid, no pgid, no lease) matches monitor-dispatch
|
||||
// losses too. Monitor-dispatch losses are owned by the monitor scheduler
|
||||
// (a future wake is already scheduled), so the reaper must NOT enter the
|
||||
// bounded null-env ladder. Instead the issue is escalated to the board
|
||||
// via a legacy_execution_requires_reconciliation recovery action.
|
||||
const { agentId, runId, issueId } = await seedRunFixture({
|
||||
adapterType: "openclaw_gateway",
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
contextSnapshot: {
|
||||
wakeReason: "issue_monitor_due",
|
||||
nextCheckAt: "2026-03-19T00:00:00.000Z",
|
||||
},
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [runId] });
|
||||
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
const failed = rows.find((row) => row.id === runId);
|
||||
const retry = rows.find((row) => row.retryOfRunId === runId);
|
||||
|
||||
// No environment-allocation diagnostic should be attached — the run was a
|
||||
// monitor dispatch loss, not an environment-selection failure.
|
||||
expect(failed?.resultJson).not.toMatchObject({
|
||||
environmentAllocationDiagnostic: expect.objectContaining({
|
||||
phase: "environment_selection",
|
||||
}),
|
||||
});
|
||||
expect(failed?.stderrExcerpt ?? "").not.toContain("[environment-allocation]");
|
||||
// No retry at all — monitor-dispatch with a future wake is handled by the
|
||||
// monitor scheduler, not by the null-env or legacy retry ladder.
|
||||
expect(retry).toBeUndefined();
|
||||
// The issue is escalated to the board via the reconciliation recovery path.
|
||||
const actions = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actions).toEqual([
|
||||
expect.objectContaining({
|
||||
ownerType: "board",
|
||||
cause: "legacy_execution_requires_reconciliation",
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not route a plan-approval continuation loss through the null-environment ladder", async () => {
|
||||
// A request_confirmation/accepted continuation wake has the same null-env
|
||||
// fingerprint, but is owned by the `interaction_continuation_infra_retry`
|
||||
// path. The bounded null-env ladder must NOT override it.
|
||||
const { companyId, agentId, runId, wakeupRequestId, issueId } = await seedQueuedIssueRunFixture();
|
||||
const interactionId = randomUUID();
|
||||
|
||||
await db.insert(issueThreadInteractions).values({
|
||||
id: interactionId,
|
||||
companyId,
|
||||
issueId,
|
||||
kind: "request_confirmation",
|
||||
status: "accepted",
|
||||
continuationPolicy: "wake_assignee_on_accept",
|
||||
createdByAgentId: agentId,
|
||||
resolvedByUserId: "responsible-user",
|
||||
resolvedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
payload: {
|
||||
version: 1,
|
||||
prompt: "Approve the plan?",
|
||||
target: { type: "issue_document", issueId, key: "plan", revisionId: randomUUID() },
|
||||
},
|
||||
result: { version: 1, outcome: "accepted" },
|
||||
});
|
||||
await db
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
source: "automation",
|
||||
reason: "issue_commented",
|
||||
status: "claimed",
|
||||
payload: {
|
||||
issueId,
|
||||
interactionId,
|
||||
interactionKind: "request_confirmation",
|
||||
interactionStatus: "accepted",
|
||||
mutation: "interaction",
|
||||
},
|
||||
})
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequestId));
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
status: "running",
|
||||
invocationSource: "automation",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
contextSnapshot: {
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason: "issue_commented",
|
||||
mutation: "interaction",
|
||||
interactionId,
|
||||
interactionKind: "request_confirmation",
|
||||
interactionStatus: "accepted",
|
||||
},
|
||||
startedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
updatedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, issueId));
|
||||
|
||||
const heartbeat = heartbeatService(db);
|
||||
const result = await heartbeat.reapOrphanedRuns();
|
||||
expect(result.reaped).toBe(1);
|
||||
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
const retry = rows.find((row) => row.retryOfRunId === runId);
|
||||
// Plan-approval path keeps its own retry reason; the null-env ladder
|
||||
// must not have fired.
|
||||
expect(retry?.scheduledRetryReason).toBe(INTERACTION_CONTINUATION_INFRA_RETRY_REASON);
|
||||
});
|
||||
|
||||
it("does not route a process-loss retry through the null-environment ladder when the legacy retry budget is exhausted", async () => {
|
||||
// The bounded null-env ladder shares the de-facto "have we already retried?"
|
||||
// budget with the legacy `process_lost_retry` path. A run whose
|
||||
// processLossRetryCount has already reached 1 must NOT enter the bounded
|
||||
// null-env ladder even when the fingerprint matches.
|
||||
const { agentId, runId } = await seedRunFixture({
|
||||
adapterType: "codex_local",
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
processLossRetryCount: 1,
|
||||
contextSnapshot: {
|
||||
wakeReason: "process_lost_retry",
|
||||
retryReason: "issue_continuation_needed",
|
||||
retryOfRunId: "original-run",
|
||||
},
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reapOrphanedRuns();
|
||||
expect(result).toEqual({ reaped: 1, runIds: [runId] });
|
||||
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
const failed = rows.find((row) => row.id === runId);
|
||||
// No environment-allocation diagnostic should be attached — this run was
|
||||
// already retried once via the legacy path and is past the budget.
|
||||
expect(failed?.resultJson).not.toMatchObject({
|
||||
environmentAllocationDiagnostic: expect.objectContaining({
|
||||
phase: "environment_selection",
|
||||
}),
|
||||
});
|
||||
// No further retry should be scheduled.
|
||||
const retries = rows.filter((row) => row.retryOfRunId === runId);
|
||||
expect(retries).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("does not route process loss through the null-environment ladder when a child pid was recorded", async () => {
|
||||
const { companyId, agentId, runId } = await seedRunFixture({
|
||||
adapterType: "codex_local",
|
||||
agentStatus: "idle",
|
||||
processPid: 4321,
|
||||
processGroupId: null,
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [runId] });
|
||||
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
const failed = rows.find((row) => row.id === runId);
|
||||
const retry = rows.find((row) => row.retryOfRunId === runId);
|
||||
|
||||
expect(failed?.resultJson).not.toMatchObject({
|
||||
environmentAllocationDiagnostic: expect.objectContaining({
|
||||
phase: "environment_selection",
|
||||
}),
|
||||
});
|
||||
expect(failed?.stderrExcerpt ?? "").not.toContain("[environment-allocation]");
|
||||
// Legacy path enqueues an immediate retry. The retry row is inserted
|
||||
// with status "scheduled_retry" (dueAt = now + 30s) and only promoted by
|
||||
// promoteDueScheduledRetries once scheduledRetryAt has elapsed, so the
|
||||
// observed status here is one of: scheduled_retry, queued, or running.
|
||||
expect(retry).toMatchObject({
|
||||
retryOfRunId: runId,
|
||||
processLossRetryCount: 1,
|
||||
});
|
||||
expect(["scheduled_retry", "queued", "running"]).toContain(retry?.status);
|
||||
// Legacy path uses the immediate process_lost_retry wake reason, not the
|
||||
// bounded environment retry wake reason.
|
||||
expect(retry?.contextSnapshot).toMatchObject({
|
||||
wakeReason: "process_lost_retry",
|
||||
});
|
||||
expect(retry?.contextSnapshot).not.toMatchObject({
|
||||
wakeReason: "process_lost_environment_retry",
|
||||
});
|
||||
// And it must NOT carry the null-environment scheduledRetryReason.
|
||||
expect(retry?.scheduledRetryReason).not.toBe("retry_transient_environment_failure");
|
||||
});
|
||||
|
||||
it("restores one lost monitor dispatch before escalating a second process loss", async () => {
|
||||
const { companyId, agentId, runId, issueId } = await seedRunFixture({
|
||||
adapterType: "openclaw_gateway",
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
|
|
|
|||
|
|
@ -755,13 +755,14 @@ export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [
|
|||
const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_JITTER_RATIO = 0;
|
||||
const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON = "transient_failure";
|
||||
const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_WAKE_REASON = "transient_failure_retry";
|
||||
const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS =
|
||||
BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length;
|
||||
export {
|
||||
INTERACTION_CONTINUATION_INFRA_RETRY_REASON,
|
||||
INTERACTION_CONTINUATION_INFRA_WAKE_REASON,
|
||||
};
|
||||
const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS = BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length;
|
||||
const NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_REASON = "retry_transient_environment_failure";
|
||||
const NULL_ENVIRONMENT_PROCESS_LOSS_WAKE_REASON = "process_lost_environment_retry";
|
||||
const PROCESS_LOST_RETRY_REASON = "process_lost_retry";
|
||||
const PROCESS_LOST_RETRY_WAKE_REASON = "process_lost_retry";
|
||||
const NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_DELAYS_MS = [60_000, 180_000, 540_000] as const;
|
||||
const INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS = 2;
|
||||
const RESOLVED_INTERACTION_CONTINUATION_STATUSES = new Set(["accepted", "answered", "rejected"]);
|
||||
const WORKSPACE_VALIDATION_FAILURE_CODE = "workspace_validation_failed";
|
||||
const WORKSPACE_VALIDATION_RECOVERY_CAUSE = "workspace_validation_failed";
|
||||
const CONFIGURATION_INCOMPLETE_FAILURE_CODE = "configuration_incomplete";
|
||||
|
|
@ -830,6 +831,10 @@ const GIT_SENSITIVE_LOCAL_ADAPTER_TYPES = new Set([
|
|||
]);
|
||||
export { MAX_TURN_CONTINUATION_RETRY_REASON };
|
||||
export const MAX_TURN_CONTINUATION_WAKE_REASON = "max_turns_continuation_retry";
|
||||
export {
|
||||
INTERACTION_CONTINUATION_INFRA_RETRY_REASON,
|
||||
INTERACTION_CONTINUATION_INFRA_WAKE_REASON,
|
||||
};
|
||||
const MAX_TURN_CONTINUATION_DEFAULT_MAX_ATTEMPTS = 2;
|
||||
const MAX_TURN_CONTINUATION_MAX_ATTEMPTS_CAP = 10;
|
||||
const MAX_TURN_CONTINUATION_DEFAULT_DELAY_MS = 1_000;
|
||||
|
|
@ -1932,6 +1937,21 @@ export function computeBoundedTransientHeartbeatRetrySchedule(
|
|||
};
|
||||
}
|
||||
|
||||
// This signature deliberately excludes a process that was ever spawned. A
|
||||
// lost child process remains on the legacy process-loss path; this ladder is
|
||||
// only for dispatches that died before an execution environment existed.
|
||||
export function isNullEnvironmentProcessLoss(input: {
|
||||
usageJson: unknown;
|
||||
processPid: number | null;
|
||||
processGroupId: number | null;
|
||||
hasEnvironmentLease: boolean;
|
||||
}) {
|
||||
return input.usageJson == null &&
|
||||
input.processPid == null &&
|
||||
input.processGroupId == null &&
|
||||
!input.hasEnvironmentLease;
|
||||
}
|
||||
|
||||
async function resolveRunScopedMentionedSkillKeys(input: {
|
||||
db: Db;
|
||||
companyId: string;
|
||||
|
|
@ -14131,8 +14151,20 @@ export function heartbeatService(
|
|||
// transient retries; process loss must not open a second retry budget.
|
||||
if (run.runtimeMode === "native" || legacyExecutionNeedsReconciliation(run))
|
||||
return null;
|
||||
const scheduled = await scheduleBoundedRetryForRun(run, agent, { now });
|
||||
return scheduled.outcome === "scheduled" ? scheduled.run : null;
|
||||
const successorLossRetryCount = (run.processLossRetryCount ?? 0) + 1;
|
||||
const scheduled = await scheduleBoundedRetryForRun(run, agent, {
|
||||
now,
|
||||
retryReason: PROCESS_LOST_RETRY_REASON,
|
||||
wakeReason: PROCESS_LOST_RETRY_WAKE_REASON,
|
||||
});
|
||||
if (scheduled.outcome !== "scheduled" || !scheduled.run) return null;
|
||||
// Mirror the budget that the reaper's alreadyRetriedOnce guard reads.
|
||||
const [bumped] = await db
|
||||
.update(heartbeatRuns)
|
||||
.set({ processLossRetryCount: successorLossRetryCount })
|
||||
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
||||
.returning();
|
||||
return bumped ?? { ...scheduled.run, processLossRetryCount: successorLossRetryCount };
|
||||
}
|
||||
|
||||
function toHotRestartIntentRun(input: {
|
||||
|
|
@ -16057,7 +16089,21 @@ export function heartbeatService(
|
|||
return null;
|
||||
}
|
||||
|
||||
return scheduleBoundedRetryForRun(run, agent, {
|
||||
// The reaper promoted this run to a process_lost CAS failure before any
|
||||
// provider work produced output; the retry is an explicit
|
||||
// infrastructure-loss replay, not an ambiguous bootstrap that the legacy
|
||||
// reconciliation gate is meant to block. Mark the failed run as safe
|
||||
// bootstrap evidence on the in-memory copy we hand to
|
||||
// scheduleBoundedRetryForRun so the shared gate does not refuse the retry.
|
||||
const runForRetry: typeof run = {
|
||||
...run,
|
||||
resultJson: {
|
||||
...(parseObject(run.resultJson) ?? {}),
|
||||
executionRecovery: { kind: "bootstrap", providerWorkStarted: false },
|
||||
},
|
||||
};
|
||||
|
||||
return scheduleBoundedRetryForRun(runForRetry, agent, {
|
||||
retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON,
|
||||
wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON,
|
||||
maxAttempts: INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS,
|
||||
|
|
@ -18706,49 +18752,127 @@ export function heartbeatService(
|
|||
readNonEmptyString(runContext.wakeReason) === "issue_monitor_due" &&
|
||||
monitorNextCheckAt !== undefined &&
|
||||
(!monitorNextCheckAt || monitorNextCheckAt.getTime() <= now.getTime());
|
||||
const shouldRetry =
|
||||
(run.processLossRetryCount ?? 0) < 1 &&
|
||||
((tracksLegacyLocalChild &&
|
||||
(!!run.processPid || !!run.processGroupId)) ||
|
||||
monitorDispatchLostWithoutFutureWake);
|
||||
const environmentLease = await db
|
||||
.select({ id: environmentLeases.id })
|
||||
.from(environmentLeases)
|
||||
.where(and(
|
||||
eq(environmentLeases.companyId, run.companyId),
|
||||
eq(environmentLeases.heartbeatRunId, run.id),
|
||||
))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
// Atomically revoke an expired legacy controller lease before we
|
||||
// classify or terminalize. Renewal and revocation serialize on the run
|
||||
// row, so a concurrent renewal that wins the CAS means another path is
|
||||
// still owning the run and we must skip this iteration.
|
||||
if (!(await revokeExpiredLegacyController(db, run))) continue;
|
||||
// The null-environment retry ladder only fires when no more specific
|
||||
// retry path already owns the run:
|
||||
// - Monitor-dispatch losses fall through to the legacy
|
||||
// `process_lost_retry` path (which uses
|
||||
// monitorDispatchLostWithoutFutureWake to decide whether a future
|
||||
// wake is already scheduled).
|
||||
// - Resolved interaction-continuation wakes fall through to the
|
||||
// plan-approval infrastructure retry.
|
||||
// - Runs that have already been retried once via the legacy path
|
||||
// (`processLossRetryCount >= 1`) are not eligible for the bounded
|
||||
// null-env ladder; the legacy retry count is the de-facto
|
||||
// "have we already retried?" budget shared with the new ladder.
|
||||
const isMonitorDispatchRun = readNonEmptyString(runContext.wakeReason) === "issue_monitor_due";
|
||||
const alreadyRetriedOnce = (run.processLossRetryCount ?? 0) >= 1;
|
||||
const nullEnvironmentProcessLoss =
|
||||
!isMonitorDispatchRun &&
|
||||
!alreadyRetriedOnce &&
|
||||
!isResolvedInteractionContinuationWakeContext(runContext) &&
|
||||
isNullEnvironmentProcessLoss({
|
||||
usageJson: run.usageJson,
|
||||
processPid: run.processPid,
|
||||
processGroupId: run.processGroupId,
|
||||
hasEnvironmentLease: environmentLease !== null,
|
||||
});
|
||||
const shouldRetryLegacyProcessLoss = (run.processLossRetryCount ?? 0) < 1 && (
|
||||
(tracksLegacyLocalChild && (!!run.processPid || !!run.processGroupId)) ||
|
||||
monitorDispatchLostWithoutFutureWake
|
||||
);
|
||||
const shouldRetry = nullEnvironmentProcessLoss || shouldRetryLegacyProcessLoss;
|
||||
const baseMessage = buildProcessLossMessage(run);
|
||||
const conversationContinuationEligible = await runUsedConversationAdapter(db, run);
|
||||
const allocationDiagnostic = nullEnvironmentProcessLoss
|
||||
? {
|
||||
phase: "environment_selection",
|
||||
outcome: "failed",
|
||||
reasonCode: "no_environment_or_lease_recorded",
|
||||
environmentId: null,
|
||||
leaseId: null,
|
||||
scratchDirHealth: "unknown",
|
||||
capturedAt: now.toISOString(),
|
||||
}
|
||||
: null;
|
||||
const allocationDiagnosticLine = allocationDiagnostic
|
||||
? `[environment-allocation] ${allocationDiagnostic.phase}:${allocationDiagnostic.reasonCode}`
|
||||
: null;
|
||||
const unmanagedBackgroundTaskEvidence = null;
|
||||
|
||||
const failurePatch = {
|
||||
error: shouldRetry
|
||||
? `${baseMessage}; ${nullEnvironmentProcessLoss ? "scheduling bounded environment retry" : "retrying once"}`
|
||||
: baseMessage,
|
||||
errorCode: "process_lost",
|
||||
finishedAt: now,
|
||||
resultJson: await (async () => {
|
||||
// Only runs whose historical invocation actually used a conversation
|
||||
// adapter can carry the continuation policy on a process-loss stop.
|
||||
// The agent's CURRENT adapter type must not relabel a lost process
|
||||
// run when an admin switches it mid-flight (see test "does not
|
||||
// relabel a lost process run when its agent changes to a
|
||||
// conversation adapter"); runUsedConversationAdapter checks the
|
||||
// persisted invocation event, not agents.adapterType.
|
||||
const conversationContinuationEligible = await runUsedConversationAdapter(db, run);
|
||||
const result = mergeRunStopMetadataForAgent(
|
||||
{ adapterType, adapterConfig },
|
||||
"failed",
|
||||
{
|
||||
conversationContinuationEligible,
|
||||
resultJson: parseObject(run.resultJson),
|
||||
errorCode: "process_lost",
|
||||
errorMessage: shouldRetry
|
||||
? `${baseMessage}; ${nullEnvironmentProcessLoss ? "scheduling bounded environment retry" : "retrying once"}`
|
||||
: baseMessage,
|
||||
},
|
||||
);
|
||||
const withAllocationDiagnostic = allocationDiagnostic
|
||||
? { ...result, environmentAllocationDiagnostic: allocationDiagnostic }
|
||||
: result;
|
||||
return unmanagedBackgroundTaskEvidence
|
||||
? {
|
||||
...withAllocationDiagnostic,
|
||||
stopReason: UNMANAGED_BACKGROUND_TASK_STOP_REASON,
|
||||
unmanagedBackgroundTask: unmanagedBackgroundTaskEvidence,
|
||||
}
|
||||
: withAllocationDiagnostic;
|
||||
})(),
|
||||
...(allocationDiagnosticLine
|
||||
? { stderrExcerpt: appendWithByteCap(run.stderrExcerpt ?? "", allocationDiagnosticLine, MAX_EXCERPT_BYTES) }
|
||||
: {}),
|
||||
};
|
||||
// Compare-and-set terminalization: the CAS guarantees another concurrent
|
||||
// recovery/drain path that already moved this run out of "running" wins
|
||||
// the race, and we keep that terminal outcome instead of overwriting it.
|
||||
// The native-ownership predicate inside setRunStatusFromLive also blocks
|
||||
// terminalizing a native run whose ownership is still held elsewhere.
|
||||
const failureWrite = await setRunStatusFromLive(
|
||||
run.id,
|
||||
"failed",
|
||||
["running"],
|
||||
{
|
||||
error: shouldRetry ? `${baseMessage}; retrying once` : baseMessage,
|
||||
errorCode: "process_lost",
|
||||
finishedAt: now,
|
||||
resultJson: (() => {
|
||||
const result = mergeRunStopMetadataForAgent(
|
||||
{ adapterType, adapterConfig },
|
||||
"failed",
|
||||
{
|
||||
conversationContinuationEligible,
|
||||
resultJson: parseObject(run.resultJson),
|
||||
errorCode: "process_lost",
|
||||
errorMessage: shouldRetry
|
||||
? `${baseMessage}; retrying once`
|
||||
: baseMessage,
|
||||
},
|
||||
);
|
||||
return result;
|
||||
})(),
|
||||
},
|
||||
failurePatch,
|
||||
);
|
||||
if (!failureWrite.updated || !failureWrite.run) continue;
|
||||
let finalizedRun: typeof heartbeatRuns.$inferSelect | null =
|
||||
failureWrite.run;
|
||||
if (!(failureWrite.updated && failureWrite.run)) continue;
|
||||
let finalizedRun: typeof failureWrite.run = failureWrite.run;
|
||||
await setWakeupStatus(run.wakeupRequestId, "failed", {
|
||||
finishedAt: now,
|
||||
error: shouldRetry ? `${baseMessage}; retrying once` : baseMessage,
|
||||
error: shouldRetry
|
||||
? `${baseMessage}; ${nullEnvironmentProcessLoss ? "scheduling bounded environment retry" : "retrying once"}`
|
||||
: baseMessage,
|
||||
});
|
||||
if (!finalizedRun) finalizedRun = await getRun(run.id);
|
||||
if (!finalizedRun) continue;
|
||||
finalizedRun =
|
||||
(await classifyAndPersistRunLiveness(
|
||||
finalizedRun,
|
||||
|
|
@ -18764,7 +18888,20 @@ export function heartbeatService(
|
|||
|
||||
let retriedRun: typeof heartbeatRuns.$inferSelect | null = null;
|
||||
const retryAgent = await getAgent(run.agentId);
|
||||
if (shouldRetry) {
|
||||
if (nullEnvironmentProcessLoss) {
|
||||
if (retryAgent) {
|
||||
const attempt = (finalizedRun.scheduledRetryAttempt ?? 0) + 1;
|
||||
const delayMs = NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_DELAYS_MS[attempt - 1];
|
||||
const scheduled = await scheduleBoundedRetryForRun(finalizedRun, retryAgent, {
|
||||
now,
|
||||
retryReason: NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_REASON,
|
||||
wakeReason: NULL_ENVIRONMENT_PROCESS_LOSS_WAKE_REASON,
|
||||
maxAttempts: NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_DELAYS_MS.length,
|
||||
...(delayMs != null ? { delayMs } : {}),
|
||||
});
|
||||
retriedRun = scheduled.outcome === "scheduled" ? scheduled.run : null;
|
||||
}
|
||||
} else if (shouldRetryLegacyProcessLoss) {
|
||||
if (retryAgent) {
|
||||
retriedRun = await enqueueProcessLossRetry(
|
||||
finalizedRun,
|
||||
|
|
@ -18795,6 +18932,7 @@ export function heartbeatService(
|
|||
payload: {
|
||||
...(run.processPid ? { processPid: run.processPid } : {}),
|
||||
...(run.processGroupId ? { processGroupId: run.processGroupId } : {}),
|
||||
...(allocationDiagnostic ? { environmentAllocationDiagnostic: allocationDiagnostic } : {}),
|
||||
...(retriedRun ? { retryRunId: retriedRun.id } : {}),
|
||||
},
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in New Issue