Split run progress from liveness disposition

This commit is contained in:
Dotta committed 2026-04-28 15:23:52 -05:00
1 parent 90e6a394c7
commit 55518170ad
5 files changed
+137 -24

No files matched your search

@@ -2110,8 +2110,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
expect(wakeups).toHaveLength(1);
});
it("records productive continuation instead of recovery when the latest automatic continuation succeeded", async () => {
const { agentId, issueId, runId } = await seedStrandedIssueFixture({
it("escalates a successful automatic continuation when no live post-run path remains", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
retryReason: "issue_continuation_needed",
@@ -2121,16 +2121,17 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
const result = await heartbeat.reconcileStrandedAssignedIssues();
expect(result.continuationRequeued).toBe(0);
expect(result.productiveContinuationObserved).toBe(1);
expect(result.productiveContinuationObserved).toBe(0);
expect(result.successfulContinuationObserved).toBe(0);
expect(result.escalated).toBe(0);
expect(result.issueIds).toEqual([]);
expect(result.escalated).toBe(1);
expect(result.issueIds).toEqual([issueId]);
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
expect(issue?.status).toBe("in_progress");
expect(issue?.status).toBe("blocked");
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
expect(comments).toHaveLength(0);
expect(comments).toHaveLength(1);
expect(comments[0]?.body).toContain("successful automatic continuation");
const runs = await db
.select()
@@ -2140,9 +2141,15 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId));
expect(wakeups).toHaveLength(1);
const recoveryIssues = await db
.select()
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery")));
expect(recoveryIssues).toHaveLength(1);
});
it.fails("does not treat a productive terminal run as healthy when in-progress work has no live path", async () => {
it("does not treat a productive terminal run as healthy when in-progress work has no live path", async () => {
const { companyId, agentId, issueId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
@@ -106,7 +106,7 @@ describe("run liveness continuations", () => {
expect(decision.nextAttempt).toBe(2);
});
it.fails("treats an advanced terminal run as progress evidence, not a live continuation path", () => {
it("treats an advanced terminal run as progress evidence, not a live continuation path", () => {
const decision = decideRunLivenessContinuation({
run: run(),
issue: issue(),
+3 -1
View File
@@ -2770,7 +2770,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
async function handleRunLivenessContinuation(run: typeof heartbeatRuns.$inferSelect) {
const livenessState = run.livenessState as RunLivenessState | null;
if (livenessState !== "plan_only" && livenessState !== "empty_response") return;
if (!livenessState) return;
const hasProgressFollowup = livenessState === "advanced" && Boolean(run.nextAction?.trim());
if (livenessState !== "plan_only" && livenessState !== "empty_response" && !hasProgressFollowup) return;
const context = parseObject(run.contextSnapshot);
const issueId = readNonEmptyString(context.issueId);
@@ -8,6 +8,7 @@ export const RUN_LIVENESS_CONTINUATION_REASON = RECOVERY_REASON_KINDS.runLivenes
export const DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS = 2;
const ACTIONABLE_LIVENESS_STATES = new Set<RunLivenessState>(["plan_only", "empty_response"]);
const PROGRESS_WITH_FOLLOWUP_LIVENESS_STATES = new Set<RunLivenessState>(["advanced"]);
const CONTINUATION_ACTIVE_ISSUE_STATUSES = new Set(["todo", "in_progress"]);
// A prior adapter error should not permanently suppress bounded liveness
// continuations; the max-attempt/idempotency guards prevent unbounded retries.
@@ -104,7 +105,15 @@ export function decideRunLivenessContinuation(input: {
} = input;
const maxAttempts = input.maxAttempts ?? DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS;
if (!livenessState || !ACTIONABLE_LIVENESS_STATES.has(livenessState)) {
if (!livenessState) {
return { kind: "skip", reason: "liveness state is not actionable for continuation" };
}
const hasExplicitRunnableFollowup = Boolean(nextAction?.trim());
const livenessCanContinue =
ACTIONABLE_LIVENESS_STATES.has(livenessState) ||
PROGRESS_WITH_FOLLOWUP_LIVENESS_STATES.has(livenessState) && hasExplicitRunnableFollowup;
if (!livenessCanContinue) {
return { kind: "skip", reason: "liveness state is not actionable for continuation" };
}
if (!issue) return { kind: "skip", reason: "issue not found" };
+108 -13
View File
@@ -72,9 +72,11 @@ type RecoveryWakeup = (
type LatestIssueRun = Pick<
typeof heartbeatRuns.$inferSelect,
"id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" | "livenessState"
"id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" | "livenessState" | "nextAction"
> | null;
type PostRunIssueDisposition = "terminal" | "explicitly_waiting" | "explicitly_live" | "invalid";
type WatchdogDecisionActor =
| { type: "board"; userId?: string | null; runId?: string | null }
| { type: "agent"; agentId?: string | null; runId?: string | null }
@@ -188,16 +190,14 @@ function isUnsuccessfulTerminalIssueRun(latestRun: LatestIssueRun) {
);
}
function isSuccessfulInProgressContinuationRun(latestRun: LatestIssueRun) {
return latestRun?.status === "succeeded";
}
function isProductiveContinuationRun(latestRun: LatestIssueRun) {
function isProductiveProgressRun(latestRun: LatestIssueRun) {
return latestRun?.status === "succeeded" &&
(latestRun.livenessState === "advanced" ||
(
latestRun.livenessState === "advanced" ||
latestRun.livenessState === "completed" ||
latestRun.livenessState === "blocked" ||
latestRun.livenessState === "needs_followup");
latestRun.livenessState === "needs_followup"
);
}
function parseLivenessIncidentKey(incidentKey: string | null | undefined) {
@@ -312,6 +312,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
errorCode: heartbeatRuns.errorCode,
contextSnapshot: heartbeatRuns.contextSnapshot,
livenessState: heartbeatRuns.livenessState,
nextAction: heartbeatRuns.nextAction,
})
.from(heartbeatRuns)
.where(
@@ -371,6 +372,76 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
.then((rows) => Boolean(rows[0]));
}
async function hasExplicitWaitingPath(issue: Pick<typeof issues.$inferSelect, "companyId" | "id" | "status" | "executionState">) {
if (issue.status === "done" || issue.status === "cancelled") return true;
if (issue.status === "blocked" || issue.status === "in_review") return true;
if (issue.executionState) return true;
const [interaction, approval, recoveryIssue] = await Promise.all([
db
.select({ id: issueThreadInteractions.id })
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.companyId, issue.companyId),
eq(issueThreadInteractions.issueId, issue.id),
eq(issueThreadInteractions.status, "pending"),
),
)
.limit(1)
.then((rows) => rows[0] ?? null),
db
.select({ id: issueApprovals.issueId })
.from(issueApprovals)
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
.where(
and(
eq(issueApprovals.companyId, issue.companyId),
eq(issueApprovals.issueId, issue.id),
inArray(approvals.status, ["pending", "revision_requested"]),
),
)
.limit(1)
.then((rows) => rows[0] ?? null),
db
.select({ id: issues.id })
.from(issues)
.where(
and(
eq(issues.companyId, issue.companyId),
eq(issues.originKind, STRANDED_ISSUE_RECOVERY_ORIGIN_KIND),
eq(issues.originId, issue.id),
notInArray(issues.status, ["done", "cancelled"]),
),
)
.limit(1)
.then((rows) => rows[0] ?? null),
]);
return Boolean(interaction || approval || recoveryIssue);
}
async function classifyPostRunIssueDisposition(
issue: typeof issues.$inferSelect,
latestRun: LatestIssueRun,
): Promise<PostRunIssueDisposition> {
if (issue.status === "done" || issue.status === "cancelled") return "terminal";
if (await hasActiveExecutionPath(issue.companyId, issue.id)) return "explicitly_live";
if (await hasExplicitWaitingPath(issue)) return "explicitly_waiting";
if (latestRun && EXECUTION_PATH_HEARTBEAT_RUN_STATUSES.includes(
latestRun.status as (typeof EXECUTION_PATH_HEARTBEAT_RUN_STATUSES)[number],
)) {
return "explicitly_live";
}
return "invalid";
}
function didAutomaticContinuationFinishWithoutLivePath(latestRun: LatestIssueRun) {
if (latestRun?.status !== "succeeded") return false;
const latestContext = parseObject(latestRun.contextSnapshot);
return readNonEmptyString(latestContext.retryReason) === "issue_continuation_needed";
}
async function enqueueStrandedIssueRecovery(input: {
issueId: string;
agentId: string;
@@ -1705,15 +1776,39 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
result.skipped += 1;
continue;
}
if (isSuccessfulInProgressContinuationRun(latestRun)) {
if (isProductiveContinuationRun(latestRun)) {
result.productiveContinuationObserved += 1;
} else {
result.successfulContinuationObserved += 1;
const postRunDisposition = await classifyPostRunIssueDisposition(issue, latestRun);
if (postRunDisposition !== "invalid") {
if (postRunDisposition === "explicitly_live" && latestRun?.status === "succeeded") {
if (isProductiveProgressRun(latestRun)) {
result.productiveContinuationObserved += 1;
} else {
result.successfulContinuationObserved += 1;
}
}
result.skipped += 1;
continue;
}
if (didAutomaticContinuationFinishWithoutLivePath(latestRun)) {
const updated = await escalateStrandedAssignedIssue({
issue,
previousStatus: "in_progress",
latestRun,
comment:
"Paperclip observed a successful automatic continuation for this assigned `in_progress` issue, " +
"but the run has finished and no live execution, queued wake, waiting interaction, approval, or recovery owner now holds the next action. " +
"Moving it to `blocked` so it is visible for intervention.",
});
if (updated) {
result.escalated += 1;
result.issueIds.push(issue.id);
} else {
result.skipped += 1;
}
continue;
}
if (didAutomaticRecoveryFail(latestRun, "issue_continuation_needed")) {
const failureSummary = summarizeRunFailureForIssueComment(latestRun);
const updated = await escalateStrandedAssignedIssue({