mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-26 18:31:34 +01:00
fix(server): add a stranded in-review feedback recovery backstop
Periodic reconciliation used to skip every `in_review` issue without a typed execution participant, so review feedback whose only remaining path was a pre-launch-failed comment run stayed silently stranded. Adds a narrow backstop for exactly that shape: agent-assigned `in_review`, no typed participant, interaction, approval, monitor, user owner, active run, or queued wake, and a failed latest run that carried user feedback context. It requeues one bounded normal-model replay for the original assignee, or opens one explicit recovery action naming the owner and next action once replay is exhausted or the assignee is no longer invokable. Dedupe against the immediate terminalization lane is structural: both lanes now derive the fingerprint, replay wake idempotency key, recovery-action kind/cause, and activity keys from one shared module, so a second parallel delivery lane cannot be opened even if the two lanes race. Legitimate waits stay untouched: pending Board confirmations, approvals, monitors and blockers, human-owned reviews, typed review participants, pause holds, paused/budget-gated assignees, system-originated wakes, and terminal issues. Exhausted delivery is still recorded for a held assignee because the recovery action carries a `manual_repair_required` wake policy and wakes nobody. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
afc4e6a81b
commit
8c5db890e0
7 files changed
+1603
-36
No files matched your search
@@ -108,6 +108,10 @@ import {
|
||||
redactDetectedSuccessfulRunProgressSummaryForBoard,
|
||||
redactSuccessfulRunHandoffEvidence,
|
||||
} from "../services/heartbeat.ts";
|
||||
import {
|
||||
buildFeedbackDeliveryFingerprint,
|
||||
buildFeedbackDeliveryRetryIdempotencyKey,
|
||||
} from "../services/recovery/feedback-delivery.ts";
|
||||
import {
|
||||
readHotRestartIntent,
|
||||
resolveLegacyHotRestartIntentPath,
|
||||
@@ -976,6 +980,195 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
return { companyId, agentId, runId, wakeupRequestId, issueId, stageId };
|
||||
}
|
||||
|
||||
/**
|
||||
* PAP-17302: the exact shape immediate terminalization missed on PAP-17271 —
|
||||
* agent-assigned `in_review` whose only review path was a Board comment wake,
|
||||
* whose confirmation the comment expired, and whose comment run failed
|
||||
* pre-launch leaving no participant, interaction, run, wake, or recovery path.
|
||||
*/
|
||||
async function seedStrandedFeedbackReviewFixture(input?: {
|
||||
retryGeneration?: number;
|
||||
rootWakeActorType?: "user" | "agent" | "system";
|
||||
agentStatus?: "idle" | "paused";
|
||||
assignToUser?: boolean;
|
||||
monitorNextCheckAt?: Date | null;
|
||||
interactionStatus?: "pending" | "expired" | null;
|
||||
runStatus?: "failed" | "succeeded";
|
||||
typedReviewParticipant?: boolean;
|
||||
}) {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
const rootWakeupRequestId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const commentId = randomUUID();
|
||||
const interactionId = randomUUID();
|
||||
const now = new Date("2026-03-19T00:00:00.000Z");
|
||||
const finishedAt = new Date("2026-03-19T00:05:00.000Z");
|
||||
const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`;
|
||||
const retryGeneration = input?.retryGeneration ?? 0;
|
||||
const replayWakeupRequestId = randomUUID();
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix,
|
||||
defaultResponsibleUserId: "responsible-user",
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "CodexCoderPro",
|
||||
role: "engineer",
|
||||
status: input?.agentStatus ?? "idle",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
id: rootWakeupRequestId,
|
||||
companyId,
|
||||
agentId,
|
||||
source: "on_demand",
|
||||
triggerDetail: "callback",
|
||||
reason: "issue_commented",
|
||||
payload: { issueId, commentId },
|
||||
status: "failed",
|
||||
runId: retryGeneration === 0 ? runId : null,
|
||||
requestedByActorType: input?.rootWakeActorType ?? "user",
|
||||
requestedByActorId: "responsible-user",
|
||||
requestedAt: now,
|
||||
claimedAt: now,
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
});
|
||||
|
||||
if (retryGeneration > 0) {
|
||||
// The generation-1 replay wake the immediate lane already queued and which
|
||||
// itself failed; it keeps the original user wake id in its context.
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
id: replayWakeupRequestId,
|
||||
companyId,
|
||||
agentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "feedback_delivery_retry",
|
||||
payload: { issueId, rootWakeupRequestId, commentIds: [commentId] },
|
||||
status: "failed",
|
||||
runId,
|
||||
requestedByActorType: "system",
|
||||
requestedAt: finishedAt,
|
||||
claimedAt: finishedAt,
|
||||
finishedAt: new Date("2026-03-19T00:10:00.000Z"),
|
||||
updatedAt: new Date("2026-03-19T00:10:00.000Z"),
|
||||
});
|
||||
}
|
||||
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
invocationSource: "on_demand",
|
||||
triggerDetail: "callback",
|
||||
status: input?.runStatus ?? "failed",
|
||||
wakeupRequestId: retryGeneration === 0 ? rootWakeupRequestId : replayWakeupRequestId,
|
||||
contextSnapshot: {
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason: retryGeneration === 0 ? "issue_commented" : "feedback_delivery_retry",
|
||||
wakeCommentIds: [commentId],
|
||||
commentId,
|
||||
wakeCommentId: commentId,
|
||||
...(retryGeneration > 0
|
||||
? {
|
||||
feedbackDeliveryRootWakeupRequestId: rootWakeupRequestId,
|
||||
feedbackDeliverySourceRunId: randomUUID(),
|
||||
feedbackDeliveryRetryGeneration: retryGeneration,
|
||||
retryReason: "feedback_delivery_retry",
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
startedAt: now,
|
||||
finishedAt,
|
||||
updatedAt: finishedAt,
|
||||
errorCode: input?.runStatus === "succeeded" ? null : "process_lost",
|
||||
error: input?.runStatus === "succeeded" ? null : "Process lost -- server may have restarted",
|
||||
// Pre-launch loss: no process identity was ever persisted.
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
});
|
||||
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Plan feedback never reached the assignee",
|
||||
status: "in_review",
|
||||
priority: "medium",
|
||||
assigneeAgentId: input?.assignToUser ? null : agentId,
|
||||
assigneeUserId: input?.assignToUser ? "user-1" : null,
|
||||
executionRunId: null,
|
||||
monitorNextCheckAt: input?.monitorNextCheckAt ?? null,
|
||||
responsibleUserId: "responsible-user",
|
||||
issueNumber: 1,
|
||||
identifier: `${issuePrefix}-1`,
|
||||
startedAt: now,
|
||||
...(input?.typedReviewParticipant
|
||||
? {
|
||||
executionState: {
|
||||
status: "pending",
|
||||
currentStageId: randomUUID(),
|
||||
currentStageIndex: 0,
|
||||
currentStageType: "review",
|
||||
currentParticipant: { type: "agent", agentId, userId: null },
|
||||
returnAssignee: { type: "agent", agentId, userId: null },
|
||||
reviewRequest: null,
|
||||
completedStageIds: [],
|
||||
lastDecisionId: null,
|
||||
lastDecisionOutcome: null,
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
|
||||
await db.insert(issueComments).values({
|
||||
id: commentId,
|
||||
companyId,
|
||||
issueId,
|
||||
authorType: "user",
|
||||
authorUserId: "responsible-user",
|
||||
body: "Please revise the plan before I approve it.",
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
});
|
||||
|
||||
if (input?.interactionStatus) {
|
||||
await db.insert(issueThreadInteractions).values({
|
||||
id: interactionId,
|
||||
companyId,
|
||||
issueId,
|
||||
kind: "request_confirmation",
|
||||
status: input.interactionStatus,
|
||||
continuationPolicy: "wake_assignee",
|
||||
payload: {
|
||||
version: 1,
|
||||
prompt: "Approve the plan?",
|
||||
target: { type: "issue_document", issueId, key: "plan" },
|
||||
},
|
||||
createdByAgentId: agentId,
|
||||
sourceRunId: runId,
|
||||
...(input.interactionStatus === "expired"
|
||||
? { resolvedAt: now, result: { version: 1, outcome: "superseded_by_comment" } }
|
||||
: {}),
|
||||
});
|
||||
}
|
||||
|
||||
return { companyId, agentId, runId, rootWakeupRequestId, issueId, commentId, interactionId };
|
||||
}
|
||||
|
||||
async function seedAssignedTodoNoRunFixture(input?: {
|
||||
agentStatus?: "paused" | "idle" | "running";
|
||||
}) {
|
||||
@@ -5741,6 +5934,287 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("requeues one normal-model recovery for stranded in-review feedback and never loops", async () => {
|
||||
const { companyId, agentId, issueId, runId, rootWakeupRequestId, commentId } =
|
||||
await seedStrandedFeedbackReviewFixture({ interactionStatus: "expired" });
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.strandedFeedbackRequeued).toBe(1);
|
||||
expect(result.strandedFeedbackRecoveryActions).toBe(0);
|
||||
expect(result.escalated).toBe(0);
|
||||
expect(result.issueIds).toEqual([issueId]);
|
||||
|
||||
const replayRuns = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.agentId, agentId), sql`${heartbeatRuns.id} <> ${runId}`));
|
||||
expect(replayRuns).toHaveLength(1);
|
||||
expect(replayRuns[0]).toMatchObject({ retryOfRunId: runId });
|
||||
expect(replayRuns[0]?.contextSnapshot).toMatchObject({
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
retryReason: FEEDBACK_DELIVERY_RETRY_REASON,
|
||||
source: "issue.stranded_feedback_delivery_backstop",
|
||||
feedbackDeliveryRootWakeupRequestId: rootWakeupRequestId,
|
||||
feedbackDeliveryRetryGeneration: 1,
|
||||
feedbackDeliveryBackstop: true,
|
||||
wakeCommentIds: [commentId],
|
||||
commentId,
|
||||
});
|
||||
// Recovery must be a normal-model replay, not a cheap status-only nudge.
|
||||
expect((replayRuns[0]?.contextSnapshot as Record<string, unknown>).modelProfile).toBeUndefined();
|
||||
|
||||
const replayWake = await db
|
||||
.select()
|
||||
.from(agentWakeupRequests)
|
||||
.where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.reason, FEEDBACK_DELIVERY_RETRY_WAKE_REASON),
|
||||
));
|
||||
expect(replayWake).toHaveLength(1);
|
||||
|
||||
const activity = await db
|
||||
.select()
|
||||
.from(activityLog)
|
||||
.where(and(eq(activityLog.entityId, issueId), eq(activityLog.action, "issue.feedback_retry_queued")));
|
||||
expect(activity).toHaveLength(1);
|
||||
expect(activity[0]?.details).toMatchObject({
|
||||
detectedBy: "issue.stranded_feedback_delivery_backstop",
|
||||
rootWakeupRequestId,
|
||||
sourceCommentIds: [commentId],
|
||||
retryGeneration: 1,
|
||||
});
|
||||
|
||||
// Bounded: a second sweep while the replay is still live must not open a
|
||||
// parallel lane, and no recovery action is created yet.
|
||||
const second = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(second.strandedFeedbackRequeued).toBe(0);
|
||||
expect(second.strandedFeedbackRecoveryActions).toBe(0);
|
||||
const runsAfter = await db
|
||||
.select({ id: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
expect(runsAfter).toHaveLength(2);
|
||||
const actionsAfter = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actionsAfter).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("opens one explicit recovery action when stranded feedback replay is exhausted", async () => {
|
||||
const { companyId, agentId, issueId, runId, rootWakeupRequestId, commentId } =
|
||||
await seedStrandedFeedbackReviewFixture({ retryGeneration: 1, interactionStatus: "expired" });
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.strandedFeedbackRequeued).toBe(0);
|
||||
expect(result.strandedFeedbackRecoveryActions).toBe(1);
|
||||
expect(result.issueIds).toEqual([issueId]);
|
||||
|
||||
// Exhaustion must not queue another replay run.
|
||||
const runs = await db
|
||||
.select({ id: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, agentId));
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0]?.id).toBe(runId);
|
||||
|
||||
const actions = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actions).toHaveLength(1);
|
||||
expect(actions[0]).toMatchObject({
|
||||
companyId,
|
||||
kind: "feedback_delivery",
|
||||
cause: "feedback_delivery_exhausted",
|
||||
status: "active",
|
||||
ownerType: "agent",
|
||||
ownerAgentId: agentId,
|
||||
previousOwnerAgentId: agentId,
|
||||
returnOwnerAgentId: agentId,
|
||||
fingerprint: buildFeedbackDeliveryFingerprint({
|
||||
companyId,
|
||||
issueId,
|
||||
agentId,
|
||||
rootWakeupRequestId,
|
||||
}),
|
||||
});
|
||||
expect(actions[0]?.nextAction).toContain("CodexCoderPro");
|
||||
expect(actions[0]?.nextAction).toContain("record a valid disposition");
|
||||
expect(actions[0]?.evidence).toMatchObject({
|
||||
detectedBy: "issue.stranded_feedback_delivery_backstop",
|
||||
backstopReason: "retry_exhausted",
|
||||
rootWakeupRequestId,
|
||||
outstandingCommentIds: [commentId],
|
||||
retryGeneration: 1,
|
||||
});
|
||||
expect(actions[0]?.wakePolicy).toMatchObject({ type: "manual_repair_required" });
|
||||
|
||||
// The issue stays in_review with the recovery action as its explicit path.
|
||||
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0]);
|
||||
expect(issue?.status).toBe("in_review");
|
||||
|
||||
const activity = await db
|
||||
.select()
|
||||
.from(activityLog)
|
||||
.where(and(eq(activityLog.entityId, issueId), eq(activityLog.action, "issue.feedback_recovery_exhausted")));
|
||||
expect(activity).toHaveLength(1);
|
||||
|
||||
// No loop: repeated sweeps neither duplicate the action nor add a replay.
|
||||
const second = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(second.strandedFeedbackRecoveryActions).toBe(0);
|
||||
expect(second.strandedFeedbackRequeued).toBe(0);
|
||||
const actionsAfter = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actionsAfter).toHaveLength(1);
|
||||
expect(actionsAfter[0]?.id).toBe(actions[0]?.id);
|
||||
const activityAfter = await db
|
||||
.select({ id: activityLog.id })
|
||||
.from(activityLog)
|
||||
.where(and(eq(activityLog.entityId, issueId), eq(activityLog.action, "issue.feedback_recovery_exhausted")));
|
||||
expect(activityAfter).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("does not open a second feedback lane when immediate terminalization already queued the replay", async () => {
|
||||
const { companyId, agentId, issueId, rootWakeupRequestId } =
|
||||
await seedStrandedFeedbackReviewFixture({ interactionStatus: "expired" });
|
||||
// Immediate lane already claimed generation 1 with the shared key.
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
companyId,
|
||||
agentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
payload: { issueId },
|
||||
status: "skipped",
|
||||
requestedByActorType: "system",
|
||||
idempotencyKey: buildFeedbackDeliveryRetryIdempotencyKey({
|
||||
fingerprint: buildFeedbackDeliveryFingerprint({
|
||||
companyId,
|
||||
issueId,
|
||||
agentId,
|
||||
rootWakeupRequestId,
|
||||
}),
|
||||
generation: 1,
|
||||
}),
|
||||
finishedAt: new Date("2026-03-19T00:06:00.000Z"),
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.strandedFeedbackRequeued).toBe(0);
|
||||
expect(result.strandedFeedbackRecoveryActions).toBe(0);
|
||||
expect(result.issueIds).not.toContain(issueId);
|
||||
});
|
||||
|
||||
it("does not open a second feedback lane when a recovery action already owns the issue", async () => {
|
||||
const { companyId, agentId, issueId, rootWakeupRequestId } =
|
||||
await seedStrandedFeedbackReviewFixture({ retryGeneration: 1, interactionStatus: "expired" });
|
||||
const existingActionId = randomUUID();
|
||||
await db.insert(issueRecoveryActions).values({
|
||||
id: existingActionId,
|
||||
companyId,
|
||||
sourceIssueId: issueId,
|
||||
kind: "feedback_delivery",
|
||||
status: "active",
|
||||
ownerType: "agent",
|
||||
ownerAgentId: agentId,
|
||||
cause: "feedback_delivery_exhausted",
|
||||
fingerprint: buildFeedbackDeliveryFingerprint({
|
||||
companyId,
|
||||
issueId,
|
||||
agentId,
|
||||
rootWakeupRequestId,
|
||||
}),
|
||||
evidence: {},
|
||||
nextAction: "Immediate lane already owns this feedback batch.",
|
||||
attemptCount: 1,
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.strandedFeedbackRequeued).toBe(0);
|
||||
expect(result.strandedFeedbackRecoveryActions).toBe(0);
|
||||
|
||||
const actions = await db
|
||||
.select()
|
||||
.from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actions).toHaveLength(1);
|
||||
expect(actions[0]?.id).toBe(existingActionId);
|
||||
expect(actions[0]?.attemptCount).toBe(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["a pending Board confirmation", { interactionStatus: "pending" as const }],
|
||||
["a human-owned review", { assignToUser: true }],
|
||||
["a monitor wait", {
|
||||
interactionStatus: "expired" as const,
|
||||
monitorNextCheckAt: new Date("2036-03-19T00:00:00.000Z"),
|
||||
}],
|
||||
["a paused assignee", { interactionStatus: "expired" as const, agentStatus: "paused" as const }],
|
||||
["a system-originated wake", { interactionStatus: "expired" as const, rootWakeActorType: "agent" as const }],
|
||||
["a successful latest run", { interactionStatus: "expired" as const, runStatus: "succeeded" as const }],
|
||||
])("leaves stranded-feedback lookalikes untouched: %s", async (_label, overrides) => {
|
||||
const { issueId } = await seedStrandedFeedbackReviewFixture(overrides);
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.strandedFeedbackRequeued).toBe(0);
|
||||
expect(result.issueIds).not.toContain(issueId);
|
||||
|
||||
const actions = await db
|
||||
.select({ id: issueRecoveryActions.id, kind: issueRecoveryActions.kind })
|
||||
.from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actions.filter((action) => action.kind === "feedback_delivery")).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("leaves an in-review issue with a typed participant to the existing participant lane", async () => {
|
||||
const { issueId } = await seedStrandedFeedbackReviewFixture({
|
||||
interactionStatus: "expired",
|
||||
typedReviewParticipant: true,
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.strandedFeedbackRequeued).toBe(0);
|
||||
expect(result.strandedFeedbackRecoveryActions).toBe(0);
|
||||
// The pre-existing execution-review participant lane still owns the issue.
|
||||
expect(result.reviewParticipantRequeued).toBe(1);
|
||||
|
||||
const actions = await db
|
||||
.select({ kind: issueRecoveryActions.kind })
|
||||
.from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actions.filter((action) => action.kind === "feedback_delivery")).toHaveLength(0);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["done", "done"],
|
||||
["cancelled", "cancelled"],
|
||||
])("leaves %s issues with stranded feedback context untouched", async (_label, status) => {
|
||||
const { issueId } = await seedStrandedFeedbackReviewFixture({ interactionStatus: "expired" });
|
||||
await db.update(issues).set({ status }).where(eq(issues.id, issueId));
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.strandedFeedbackRequeued).toBe(0);
|
||||
expect(result.strandedFeedbackRecoveryActions).toBe(0);
|
||||
|
||||
const actions = await db
|
||||
.select({ id: issueRecoveryActions.id })
|
||||
.from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actions).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("escalates accepted interaction continuation recovery after three review-park cancellations", async () => {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
|
||||
@@ -1130,6 +1130,8 @@ export async function startServer(): Promise<StartedServer> {
|
||||
reconciled.dispatchRequeued > 0 ||
|
||||
reconciled.continuationRequeued > 0 ||
|
||||
reconciled.successfulRunHandoffEscalated > 0 ||
|
||||
reconciled.strandedFeedbackRequeued > 0 ||
|
||||
reconciled.strandedFeedbackRecoveryActions > 0 ||
|
||||
reconciled.escalated > 0
|
||||
) {
|
||||
logger.warn(
|
||||
@@ -1344,6 +1346,8 @@ export async function startServer(): Promise<StartedServer> {
|
||||
reconciled.dispatchRequeued > 0 ||
|
||||
reconciled.continuationRequeued > 0 ||
|
||||
reconciled.successfulRunHandoffEscalated > 0 ||
|
||||
reconciled.strandedFeedbackRequeued > 0 ||
|
||||
reconciled.strandedFeedbackRecoveryActions > 0 ||
|
||||
reconciled.escalated > 0
|
||||
) {
|
||||
logger.warn(
|
||||
|
||||
@@ -224,6 +224,16 @@ import {
|
||||
isSuccessfulRunHandoffValidPathSkip,
|
||||
SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY,
|
||||
readContinuationAttempt,
|
||||
FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
|
||||
FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND,
|
||||
FEEDBACK_DELIVERY_RECOVERY_CAUSE,
|
||||
FEEDBACK_DELIVERY_RETRY_REASON,
|
||||
FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
buildFeedbackDeliveryFingerprint,
|
||||
buildFeedbackDeliveryRetryIdempotencyKey,
|
||||
carriesExplicitFeedbackResume,
|
||||
isEligibleFeedbackDeliveryWake,
|
||||
readFeedbackDeliveryRetryGeneration,
|
||||
} from "./recovery/index.js";
|
||||
import { isAutomaticRecoverySuppressedByPauseHold } from "./recovery/pause-hold-guard.js";
|
||||
import {
|
||||
@@ -486,10 +496,7 @@ export const WORKSPACE_BUSY_RETRY_WAKE_REASON = "workspace_busy_retry";
|
||||
export const WORKSPACE_BUSY_ERROR_CODE = "workspace_busy";
|
||||
export const WORKSPACE_BUSY_RETRY_BASE_DELAY_MS = 60 * 1000;
|
||||
export const WORKSPACE_BUSY_RETRY_JITTER_MS = 60 * 1000;
|
||||
export const FEEDBACK_DELIVERY_RETRY_REASON = "feedback_delivery_retry";
|
||||
export const FEEDBACK_DELIVERY_RETRY_WAKE_REASON = "feedback_delivery_retry";
|
||||
const FEEDBACK_DELIVERY_RECOVERY_CAUSE = "feedback_delivery_exhausted";
|
||||
const FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES = 1;
|
||||
export { FEEDBACK_DELIVERY_RETRY_REASON, FEEDBACK_DELIVERY_RETRY_WAKE_REASON };
|
||||
// A running run stops counting as a shared-workspace holder once it has been
|
||||
// silent this long. This is recovery's own "suspicious silence" bar for active
|
||||
// runs (scanSilentActiveRuns escalates such runs), so a zombie holder cannot
|
||||
@@ -10155,17 +10162,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
|
||||
const wakeReason = readNonEmptyString(context.wakeReason);
|
||||
const commentIds = mergeWakeCommentIds(context, deriveCommentId(context, null));
|
||||
const carriesExplicitResume = context.resumeIntent === true || context.followUpRequested === true;
|
||||
const eligibleWake =
|
||||
wakeReason === "issue_commented" ||
|
||||
wakeReason === "issue_reopened_via_comment" ||
|
||||
wakeReason === FEEDBACK_DELIVERY_RETRY_WAKE_REASON ||
|
||||
(carriesExplicitResume && (
|
||||
wakeReason === "issue_assigned" ||
|
||||
wakeReason === "issue_status_changed" ||
|
||||
wakeReason === "issue_commented" ||
|
||||
wakeReason === "issue_reopened_via_comment"
|
||||
));
|
||||
const carriesExplicitResume = carriesExplicitFeedbackResume(context);
|
||||
const eligibleWake = isEligibleFeedbackDeliveryWake({ wakeReason, carriesExplicitResume });
|
||||
if (!eligibleWake || (commentIds.length === 0 && !carriesExplicitResume)) return null;
|
||||
|
||||
const rootWake = await db
|
||||
@@ -10181,28 +10179,23 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
// generation-1 wakes retain their original user wake id in context.
|
||||
if (rootWake?.requestedByActorType !== "user") return null;
|
||||
|
||||
const generation = Math.max(
|
||||
0,
|
||||
Math.floor(
|
||||
typeof context.feedbackDeliveryRetryGeneration === "number"
|
||||
? context.feedbackDeliveryRetryGeneration
|
||||
: run.scheduledRetryReason === FEEDBACK_DELIVERY_RETRY_REASON
|
||||
? run.scheduledRetryAttempt ?? 0
|
||||
: 0,
|
||||
),
|
||||
);
|
||||
return {
|
||||
issueId,
|
||||
rootWakeupRequestId,
|
||||
commentIds,
|
||||
generation,
|
||||
fingerprint: [
|
||||
"feedback_delivery",
|
||||
run.companyId,
|
||||
generation: readFeedbackDeliveryRetryGeneration({
|
||||
contextSnapshot: context,
|
||||
scheduledRetryReason: run.scheduledRetryReason,
|
||||
scheduledRetryAttempt: run.scheduledRetryAttempt,
|
||||
}),
|
||||
// Shared with the periodic backstop so neither lane can drift into a
|
||||
// second parallel delivery lane for the same feedback batch.
|
||||
fingerprint: buildFeedbackDeliveryFingerprint({
|
||||
companyId: run.companyId,
|
||||
issueId,
|
||||
run.agentId,
|
||||
agentId: run.agentId,
|
||||
rootWakeupRequestId,
|
||||
].join(":"),
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -10323,7 +10316,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
|
||||
const existing = await recoveryActions.getActiveForIssue(input.run.companyId, issue.id);
|
||||
if (
|
||||
existing?.kind === "feedback_delivery"
|
||||
existing?.kind === FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND
|
||||
&& existing.cause === FEEDBACK_DELIVERY_RECOVERY_CAUSE
|
||||
&& existing.fingerprint === input.delivery.fingerprint
|
||||
) {
|
||||
@@ -10333,7 +10326,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const action = await recoveryActions.upsertSourceScoped({
|
||||
companyId: input.run.companyId,
|
||||
sourceIssueId: issue.id,
|
||||
kind: "feedback_delivery",
|
||||
kind: FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND,
|
||||
ownerType: "agent",
|
||||
ownerAgentId: input.agent.id,
|
||||
previousOwnerAgentId: input.agent.id,
|
||||
@@ -10415,7 +10408,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const taskKey = deriveTaskKeyWithHeartbeatFallback(retryContext, null);
|
||||
const sessionBefore = await resolveSessionBeforeForWakeup(input.agent, taskKey);
|
||||
const responsibleUserId = await resolveResponsibleUserIdForRunContext(input.run, retryContext);
|
||||
const idempotencyKey = `${input.delivery.fingerprint}:generation:1`;
|
||||
const idempotencyKey = buildFeedbackDeliveryRetryIdempotencyKey({
|
||||
fingerprint: input.delivery.fingerprint,
|
||||
generation: 1,
|
||||
});
|
||||
|
||||
const result = await db.transaction(async (tx) => {
|
||||
const companyIsActive = await tx
|
||||
|
||||
@@ -0,0 +1,333 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
|
||||
FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
|
||||
buildFeedbackDeliveryFingerprint,
|
||||
buildFeedbackDeliveryRetryIdempotencyKey,
|
||||
decideStrandedFeedbackDeliveryBackstop,
|
||||
readFeedbackDeliveryRunContext,
|
||||
type StrandedFeedbackDeliveryBackstopProbe,
|
||||
} from "./feedback-delivery.js";
|
||||
|
||||
const COMPANY_ID = "company-1";
|
||||
const ISSUE_ID = "issue-1";
|
||||
const AGENT_ID = "agent-1";
|
||||
const ROOT_WAKE_ID = "wake-root-1";
|
||||
const COMMENT_ID = "comment-1";
|
||||
|
||||
const openProbe: StrandedFeedbackDeliveryBackstopProbe = {
|
||||
hasActiveExecutionPath: false,
|
||||
hasPendingWakeInteraction: false,
|
||||
hasPendingApproval: false,
|
||||
hasPersistedDurableWaitPath: false,
|
||||
hasQueuedIssueWake: false,
|
||||
hasActiveRecoveryAction: false,
|
||||
hasExistingRetryWake: false,
|
||||
assigneeInvokable: true,
|
||||
assigneeHold: null,
|
||||
budgetBlocked: false,
|
||||
pauseHeld: false,
|
||||
};
|
||||
|
||||
function failedFeedbackRun(overrides: Record<string, unknown> = {}) {
|
||||
const {
|
||||
contextSnapshot: contextOverrides,
|
||||
...runOverrides
|
||||
} = overrides as { contextSnapshot?: Record<string, unknown> } & Record<string, unknown>;
|
||||
return {
|
||||
id: "run-1",
|
||||
agentId: AGENT_ID,
|
||||
status: "failed",
|
||||
errorCode: "process_lost",
|
||||
wakeupRequestId: ROOT_WAKE_ID,
|
||||
contextSnapshot: {
|
||||
issueId: ISSUE_ID,
|
||||
taskId: ISSUE_ID,
|
||||
wakeReason: "issue_commented",
|
||||
wakeCommentIds: [COMMENT_ID],
|
||||
...(contextOverrides ?? {}),
|
||||
},
|
||||
scheduledRetryReason: null,
|
||||
scheduledRetryAttempt: null,
|
||||
...runOverrides,
|
||||
};
|
||||
}
|
||||
|
||||
function decide(input: {
|
||||
probe?: Partial<StrandedFeedbackDeliveryBackstopProbe>;
|
||||
issue?: Partial<{
|
||||
id: string;
|
||||
status: string;
|
||||
assigneeAgentId: string | null;
|
||||
assigneeUserId: string | null;
|
||||
}>;
|
||||
latestRun?: ReturnType<typeof failedFeedbackRun> | null;
|
||||
hasTypedReviewParticipant?: boolean;
|
||||
rootWakeRequestedByActorType?: string | null;
|
||||
} = {}) {
|
||||
return decideStrandedFeedbackDeliveryBackstop({
|
||||
companyId: COMPANY_ID,
|
||||
issue: {
|
||||
id: ISSUE_ID,
|
||||
status: "in_review",
|
||||
assigneeAgentId: AGENT_ID,
|
||||
assigneeUserId: null,
|
||||
...(input.issue ?? {}),
|
||||
},
|
||||
hasTypedReviewParticipant: input.hasTypedReviewParticipant ?? false,
|
||||
latestRun: input.latestRun === undefined ? failedFeedbackRun() : input.latestRun,
|
||||
rootWakeRequestedByActorType: input.rootWakeRequestedByActorType ?? "user",
|
||||
probe: { ...openProbe, ...(input.probe ?? {}) },
|
||||
});
|
||||
}
|
||||
|
||||
const FINGERPRINT = buildFeedbackDeliveryFingerprint({
|
||||
companyId: COMPANY_ID,
|
||||
issueId: ISSUE_ID,
|
||||
agentId: AGENT_ID,
|
||||
rootWakeupRequestId: ROOT_WAKE_ID,
|
||||
});
|
||||
|
||||
describe("stranded feedback delivery backstop", () => {
|
||||
it("requeues exactly one normal-model recovery for the original assignee", () => {
|
||||
const decision = decide();
|
||||
|
||||
expect(decision).toMatchObject({
|
||||
kind: "requeue",
|
||||
generation: 1,
|
||||
idempotencyKey: buildFeedbackDeliveryRetryIdempotencyKey({
|
||||
fingerprint: FINGERPRINT,
|
||||
generation: 1,
|
||||
}),
|
||||
payload: {
|
||||
issueId: ISSUE_ID,
|
||||
retryOfRunId: "run-1",
|
||||
feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID,
|
||||
feedbackDeliveryRetryGeneration: 1,
|
||||
feedbackDeliveryBackstop: true,
|
||||
wakeCommentIds: [COMMENT_ID],
|
||||
commentId: COMMENT_ID,
|
||||
},
|
||||
contextSnapshot: {
|
||||
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
source: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
|
||||
},
|
||||
});
|
||||
if (decision.kind !== "requeue") throw new Error("expected a requeue decision");
|
||||
// Normal-model recovery must not carry the cheap/status-only guard hints.
|
||||
expect(decision.contextSnapshot.modelProfile).toBeUndefined();
|
||||
expect(decision.contextSnapshot.recoveryIntent).toBeUndefined();
|
||||
expect(decision.delivery.agentId).toBe(AGENT_ID);
|
||||
});
|
||||
|
||||
it("shares the immediate lane's fingerprint and replay key so there is never a second lane", () => {
|
||||
const immediate = readFeedbackDeliveryRunContext({
|
||||
companyId: COMPANY_ID,
|
||||
run: failedFeedbackRun(),
|
||||
});
|
||||
const decision = decide();
|
||||
|
||||
expect(immediate?.fingerprint).toBe(FINGERPRINT);
|
||||
if (decision.kind !== "requeue") throw new Error("expected a requeue decision");
|
||||
expect(decision.idempotencyKey).toBe(
|
||||
buildFeedbackDeliveryRetryIdempotencyKey({ fingerprint: immediate!.fingerprint, generation: 1 }),
|
||||
);
|
||||
});
|
||||
|
||||
it("skips when the immediate lane already queued the same replay generation", () => {
|
||||
expect(decide({ probe: { hasExistingRetryWake: true } })).toEqual({
|
||||
kind: "skip",
|
||||
reason: "feedback delivery replay wake already exists",
|
||||
});
|
||||
});
|
||||
|
||||
it("skips when any recovery action already owns the issue", () => {
|
||||
expect(decide({ probe: { hasActiveRecoveryAction: true } })).toEqual({
|
||||
kind: "skip",
|
||||
reason: "issue already has an active recovery action",
|
||||
});
|
||||
});
|
||||
|
||||
it("opens one explicit recovery action once automatic replay is exhausted", () => {
|
||||
const exhausted = failedFeedbackRun({
|
||||
contextSnapshot: {
|
||||
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID,
|
||||
feedbackDeliveryRetryGeneration: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
|
||||
wakeCommentIds: [COMMENT_ID],
|
||||
},
|
||||
wakeupRequestId: "wake-retry-1",
|
||||
});
|
||||
|
||||
const decision = decide({ latestRun: exhausted });
|
||||
expect(decision).toMatchObject({ kind: "recovery_action", reason: "retry_exhausted" });
|
||||
if (decision.kind !== "recovery_action") throw new Error("expected a recovery action");
|
||||
// The exhausted generation keeps the original root wake fingerprint so it
|
||||
// lands on the immediate lane's action instead of creating a new one.
|
||||
expect(decision.delivery.fingerprint).toBe(FINGERPRINT);
|
||||
expect(decision.delivery.generation).toBe(FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES);
|
||||
});
|
||||
|
||||
it("reads the exhausted generation from a scheduled retry run that lost its context hint", () => {
|
||||
const decision = decide({
|
||||
latestRun: failedFeedbackRun({
|
||||
contextSnapshot: {
|
||||
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID,
|
||||
wakeCommentIds: [COMMENT_ID],
|
||||
},
|
||||
scheduledRetryReason: "feedback_delivery_retry",
|
||||
scheduledRetryAttempt: 1,
|
||||
}),
|
||||
});
|
||||
|
||||
expect(decision).toMatchObject({ kind: "recovery_action", reason: "retry_exhausted" });
|
||||
});
|
||||
|
||||
it("cannot loop: a replayed generation never requeues another replay", () => {
|
||||
let run = failedFeedbackRun();
|
||||
const generations: number[] = [];
|
||||
for (let round = 0; round < 5; round += 1) {
|
||||
const decision = decide({ latestRun: run });
|
||||
if (decision.kind !== "requeue") {
|
||||
expect(decision).toMatchObject({ kind: "recovery_action" });
|
||||
break;
|
||||
}
|
||||
generations.push(decision.generation);
|
||||
// Simulate the replay run also failing without handling the feedback.
|
||||
run = failedFeedbackRun({
|
||||
contextSnapshot: decision.contextSnapshot as Record<string, unknown>,
|
||||
wakeupRequestId: `wake-retry-${decision.generation}`,
|
||||
});
|
||||
}
|
||||
|
||||
expect(generations).toEqual([1]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["a paused assignee", { assigneeInvokable: false, assigneeHold: "paused" as const }],
|
||||
["a budget-blocked assignee", { budgetBlocked: true }],
|
||||
])("still records exhausted feedback for %s, because the action never wakes anyone", (_label, probe) => {
|
||||
const decision = decide({
|
||||
probe,
|
||||
latestRun: failedFeedbackRun({
|
||||
contextSnapshot: {
|
||||
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID,
|
||||
feedbackDeliveryRetryGeneration: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
|
||||
wakeCommentIds: [COMMENT_ID],
|
||||
},
|
||||
wakeupRequestId: "wake-retry-1",
|
||||
}),
|
||||
});
|
||||
|
||||
expect(decision).toMatchObject({ kind: "recovery_action", reason: "retry_exhausted" });
|
||||
});
|
||||
|
||||
it("routes to an explicit recovery action when the original assignee is not invokable", () => {
|
||||
expect(decide({ probe: { assigneeInvokable: false, assigneeHold: null } })).toMatchObject({
|
||||
kind: "recovery_action",
|
||||
reason: "assignee_not_invokable",
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
["a pending Board interaction", { hasPendingWakeInteraction: true }, "issue has a pending waking interaction"],
|
||||
["a pending approval", { hasPendingApproval: true }, "issue has a pending approval"],
|
||||
["a monitor or blocker wait", { hasPersistedDurableWaitPath: true }, "issue has a persisted durable wait path"],
|
||||
["an active run", { hasActiveExecutionPath: true }, "issue already has an active execution path"],
|
||||
["a queued wake", { hasQueuedIssueWake: true }, "issue already has a queued wake"],
|
||||
["a pause hold", { pauseHeld: true }, "automatic recovery is suppressed by a pause hold"],
|
||||
["a budget block", { budgetBlocked: true }, "assignee invocation is budget blocked"],
|
||||
[
|
||||
"a paused assignee",
|
||||
{ assigneeInvokable: false, assigneeHold: "paused" as const },
|
||||
"assignee is held (paused)",
|
||||
],
|
||||
[
|
||||
"an assignee awaiting approval",
|
||||
{ assigneeInvokable: false, assigneeHold: "pending_approval" as const },
|
||||
"assignee is held (pending_approval)",
|
||||
],
|
||||
])("leaves a legitimate wait untouched: %s", (_label, probe, reason) => {
|
||||
expect(decide({ probe })).toEqual({ kind: "skip", reason });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["done", "done"],
|
||||
["cancelled", "cancelled"],
|
||||
["in_progress", "in_progress"],
|
||||
["blocked", "blocked"],
|
||||
])("leaves %s issues untouched", (_label, status) => {
|
||||
expect(decide({ issue: { status } })).toEqual({ kind: "skip", reason: "issue is not in review" });
|
||||
});
|
||||
|
||||
it("leaves human-owned reviews untouched", () => {
|
||||
expect(decide({ issue: { assigneeUserId: "user-1" } })).toEqual({
|
||||
kind: "skip",
|
||||
reason: "review is owned by a user",
|
||||
});
|
||||
});
|
||||
|
||||
it("leaves reviews with a typed execution participant to the participant lane", () => {
|
||||
expect(decide({ hasTypedReviewParticipant: true })).toEqual({
|
||||
kind: "skip",
|
||||
reason: "review issue has a typed execution participant",
|
||||
});
|
||||
});
|
||||
|
||||
it("ignores a latest run that carried no feedback context", () => {
|
||||
expect(
|
||||
decide({
|
||||
latestRun: failedFeedbackRun({
|
||||
contextSnapshot: { issueId: ISSUE_ID, wakeReason: "issue_assigned" },
|
||||
}),
|
||||
}),
|
||||
).toEqual({ kind: "skip", reason: "latest failed run carried no feedback delivery context" });
|
||||
});
|
||||
|
||||
it("ignores feedback whose root wake did not come from a user", () => {
|
||||
expect(decide({ rootWakeRequestedByActorType: "agent" })).toEqual({
|
||||
kind: "skip",
|
||||
reason: "feedback wake did not originate from a user",
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
["succeeded", "succeeded"],
|
||||
["running", "running"],
|
||||
["cancelled", "cancelled"],
|
||||
["interrupted", "interrupted"],
|
||||
])("ignores a latest run in status %s", (_label, status) => {
|
||||
expect(decide({ latestRun: failedFeedbackRun({ status }) })).toEqual({
|
||||
kind: "skip",
|
||||
reason: "latest run did not fail",
|
||||
});
|
||||
});
|
||||
|
||||
it("ignores a failed feedback run owned by a different agent", () => {
|
||||
expect(decide({ latestRun: failedFeedbackRun({ agentId: "agent-2" }) })).toEqual({
|
||||
kind: "skip",
|
||||
reason: "latest failed run belongs to a different agent",
|
||||
});
|
||||
});
|
||||
|
||||
it("ignores a review issue with no runs at all", () => {
|
||||
expect(decide({ latestRun: null })).toEqual({ kind: "skip", reason: "review issue has no latest run" });
|
||||
});
|
||||
|
||||
it("accepts an explicit resume wake with no comment ids", () => {
|
||||
expect(
|
||||
decide({
|
||||
latestRun: failedFeedbackRun({
|
||||
contextSnapshot: {
|
||||
issueId: ISSUE_ID,
|
||||
wakeReason: "issue_status_changed",
|
||||
resumeIntent: true,
|
||||
},
|
||||
}),
|
||||
}),
|
||||
).toMatchObject({ kind: "requeue", generation: 1 });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,348 @@
|
||||
import { withRecoveryModelProfileHint } from "./model-profile-hint.js";
|
||||
|
||||
export const FEEDBACK_DELIVERY_RETRY_REASON = "feedback_delivery_retry";
|
||||
export const FEEDBACK_DELIVERY_RETRY_WAKE_REASON = "feedback_delivery_retry";
|
||||
export const FEEDBACK_DELIVERY_RECOVERY_CAUSE = "feedback_delivery_exhausted";
|
||||
export const FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND = "feedback_delivery" as const;
|
||||
export const FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES = 1;
|
||||
export const FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY = "wakeCommentIds";
|
||||
|
||||
export const STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE = "issue.stranded_feedback_delivery_backstop";
|
||||
export const STRANDED_FEEDBACK_DELIVERY_BACKSTOP_INSTRUCTION =
|
||||
"Human feedback on this in-review issue was never durably handled: the run created for it failed and no reviewer, " +
|
||||
"interaction, approval, monitor, active run, or queued wake remained. Handle the outstanding comment(s) now and " +
|
||||
"record a valid disposition. This is the only automatic backstop wake for this feedback batch.";
|
||||
export const STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION =
|
||||
"Handle the outstanding human feedback comment(s) on this in-review issue and record a valid disposition, " +
|
||||
"or state the exact blocker and named unblock owner. Automatic feedback replay is exhausted.";
|
||||
|
||||
// Feedback delivery is a human-input contract, so only wakes whose root request
|
||||
// came from a user create a delivery obligation.
|
||||
export const FEEDBACK_DELIVERY_ROOT_WAKE_ACTOR_TYPE = "user";
|
||||
|
||||
const FEEDBACK_DELIVERY_WAKE_REASONS = [
|
||||
"issue_commented",
|
||||
"issue_reopened_via_comment",
|
||||
FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
] as const;
|
||||
const FEEDBACK_DELIVERY_RESUME_WAKE_REASONS = [
|
||||
"issue_assigned",
|
||||
"issue_status_changed",
|
||||
"issue_commented",
|
||||
"issue_reopened_via_comment",
|
||||
] as const;
|
||||
|
||||
function readNonEmptyString(value: unknown) {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
|
||||
}
|
||||
|
||||
export function buildFeedbackDeliveryFingerprint(input: {
|
||||
companyId: string;
|
||||
issueId: string;
|
||||
agentId: string;
|
||||
rootWakeupRequestId: string;
|
||||
}) {
|
||||
return [
|
||||
"feedback_delivery",
|
||||
input.companyId,
|
||||
input.issueId,
|
||||
input.agentId,
|
||||
input.rootWakeupRequestId,
|
||||
].join(":");
|
||||
}
|
||||
|
||||
/**
|
||||
* Both delivery lanes (immediate terminalization in heartbeat and the periodic
|
||||
* backstop in recovery) derive the replay wake key from this function, so a
|
||||
* generation can only ever be queued once no matter which lane notices first.
|
||||
*/
|
||||
export function buildFeedbackDeliveryRetryIdempotencyKey(input: {
|
||||
fingerprint: string;
|
||||
generation: number;
|
||||
}) {
|
||||
return `${input.fingerprint}:generation:${input.generation}`;
|
||||
}
|
||||
|
||||
export function readFeedbackDeliveryWakeCommentIds(
|
||||
contextSnapshot: Record<string, unknown> | null | undefined,
|
||||
): string[] {
|
||||
const merged: string[] = [];
|
||||
const append = (value: unknown) => {
|
||||
const normalized = readNonEmptyString(value);
|
||||
if (!normalized || merged.includes(normalized)) return;
|
||||
merged.push(normalized);
|
||||
};
|
||||
|
||||
const batched = contextSnapshot?.[FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY];
|
||||
if (Array.isArray(batched)) {
|
||||
for (const entry of batched) append(entry);
|
||||
}
|
||||
append(contextSnapshot?.wakeCommentId);
|
||||
append(contextSnapshot?.commentId);
|
||||
return merged;
|
||||
}
|
||||
|
||||
export function carriesExplicitFeedbackResume(
|
||||
contextSnapshot: Record<string, unknown> | null | undefined,
|
||||
) {
|
||||
return contextSnapshot?.resumeIntent === true || contextSnapshot?.followUpRequested === true;
|
||||
}
|
||||
|
||||
export function isEligibleFeedbackDeliveryWake(input: {
|
||||
wakeReason: string | null;
|
||||
carriesExplicitResume: boolean;
|
||||
}) {
|
||||
if (!input.wakeReason) return false;
|
||||
if ((FEEDBACK_DELIVERY_WAKE_REASONS as readonly string[]).includes(input.wakeReason)) return true;
|
||||
return (
|
||||
input.carriesExplicitResume &&
|
||||
(FEEDBACK_DELIVERY_RESUME_WAKE_REASONS as readonly string[]).includes(input.wakeReason)
|
||||
);
|
||||
}
|
||||
|
||||
export function readFeedbackDeliveryRetryGeneration(input: {
|
||||
contextSnapshot: Record<string, unknown> | null | undefined;
|
||||
scheduledRetryReason?: string | null;
|
||||
scheduledRetryAttempt?: number | null;
|
||||
}) {
|
||||
const raw = input.contextSnapshot?.feedbackDeliveryRetryGeneration;
|
||||
const fromContext = typeof raw === "number" ? raw : null;
|
||||
const fromRun = input.scheduledRetryReason === FEEDBACK_DELIVERY_RETRY_REASON
|
||||
? input.scheduledRetryAttempt ?? 0
|
||||
: 0;
|
||||
return Math.max(0, Math.floor(fromContext ?? fromRun));
|
||||
}
|
||||
|
||||
export type FeedbackDeliveryRunContext = {
|
||||
issueId: string;
|
||||
agentId: string;
|
||||
rootWakeupRequestId: string;
|
||||
commentIds: string[];
|
||||
generation: number;
|
||||
fingerprint: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Pure half of feedback-delivery context extraction: everything both lanes can
|
||||
* decide from the run row alone. Callers still confirm the root wake was
|
||||
* user-requested, which needs a lookup.
|
||||
*/
|
||||
export function readFeedbackDeliveryRunContext(input: {
|
||||
companyId: string;
|
||||
run: {
|
||||
id: string;
|
||||
agentId: string;
|
||||
wakeupRequestId: string | null;
|
||||
contextSnapshot: unknown;
|
||||
scheduledRetryReason?: string | null;
|
||||
scheduledRetryAttempt?: number | null;
|
||||
};
|
||||
}): FeedbackDeliveryRunContext | null {
|
||||
const context = (typeof input.run.contextSnapshot === "object" && input.run.contextSnapshot !== null
|
||||
? input.run.contextSnapshot
|
||||
: {}) as Record<string, unknown>;
|
||||
const issueId = readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId);
|
||||
const rootWakeupRequestId =
|
||||
readNonEmptyString(context.feedbackDeliveryRootWakeupRequestId) ?? input.run.wakeupRequestId;
|
||||
if (!issueId || !rootWakeupRequestId) return null;
|
||||
|
||||
const commentIds = readFeedbackDeliveryWakeCommentIds(context);
|
||||
const carriesExplicitResume = carriesExplicitFeedbackResume(context);
|
||||
if (
|
||||
!isEligibleFeedbackDeliveryWake({
|
||||
wakeReason: readNonEmptyString(context.wakeReason),
|
||||
carriesExplicitResume,
|
||||
})
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
if (commentIds.length === 0 && !carriesExplicitResume) return null;
|
||||
|
||||
return {
|
||||
issueId,
|
||||
agentId: input.run.agentId,
|
||||
rootWakeupRequestId,
|
||||
commentIds,
|
||||
generation: readFeedbackDeliveryRetryGeneration({
|
||||
contextSnapshot: context,
|
||||
scheduledRetryReason: input.run.scheduledRetryReason ?? null,
|
||||
scheduledRetryAttempt: input.run.scheduledRetryAttempt ?? null,
|
||||
}),
|
||||
fingerprint: buildFeedbackDeliveryFingerprint({
|
||||
companyId: input.companyId,
|
||||
issueId,
|
||||
agentId: input.run.agentId,
|
||||
rootWakeupRequestId,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
export type StrandedFeedbackDeliveryBackstopDecision =
|
||||
| {
|
||||
kind: "requeue";
|
||||
delivery: FeedbackDeliveryRunContext;
|
||||
generation: number;
|
||||
idempotencyKey: string;
|
||||
payload: Record<string, unknown>;
|
||||
contextSnapshot: Record<string, unknown>;
|
||||
}
|
||||
| {
|
||||
kind: "recovery_action";
|
||||
delivery: FeedbackDeliveryRunContext;
|
||||
reason: "retry_exhausted" | "assignee_not_invokable";
|
||||
}
|
||||
| { kind: "skip"; reason: string };
|
||||
|
||||
export type StrandedFeedbackDeliveryBackstopProbe = {
|
||||
/** Any live or queued-for-execution run on the issue, regardless of agent. */
|
||||
hasActiveExecutionPath: boolean;
|
||||
hasPendingWakeInteraction: boolean;
|
||||
hasPendingApproval: boolean;
|
||||
/** Monitor check or an unresolved `blocks` dependency. */
|
||||
hasPersistedDurableWaitPath: boolean;
|
||||
hasQueuedIssueWake: boolean;
|
||||
/** Any active/escalated recovery action already owns this issue. */
|
||||
hasActiveRecoveryAction: boolean;
|
||||
/** A wake already exists for the replay idempotency key this lane would use. */
|
||||
hasExistingRetryWake: boolean;
|
||||
assigneeInvokable: boolean;
|
||||
/**
|
||||
* Set when the assignee is deliberately held (operator pause, or still
|
||||
* awaiting approval). That is a legitimate wait, not stranded feedback, so it
|
||||
* is left alone rather than routed to a takeover owner.
|
||||
*/
|
||||
assigneeHold: "paused" | "pending_approval" | null;
|
||||
budgetBlocked: boolean;
|
||||
pauseHeld: boolean;
|
||||
};
|
||||
|
||||
/**
|
||||
* Narrow reconciliation backstop for review feedback that immediate
|
||||
* terminalization missed: agent-assigned `in_review` with no typed review
|
||||
* participant, no interaction/approval/monitor/user owner, no active run or
|
||||
* queued wake, and a failed latest run that carried user feedback context.
|
||||
*/
|
||||
export function decideStrandedFeedbackDeliveryBackstop(input: {
|
||||
companyId: string;
|
||||
issue: {
|
||||
id: string;
|
||||
status: string;
|
||||
assigneeAgentId: string | null;
|
||||
assigneeUserId: string | null;
|
||||
};
|
||||
hasTypedReviewParticipant: boolean;
|
||||
latestRun: {
|
||||
id: string;
|
||||
agentId: string;
|
||||
status: string;
|
||||
errorCode: string | null;
|
||||
wakeupRequestId: string | null;
|
||||
contextSnapshot: unknown;
|
||||
scheduledRetryReason?: string | null;
|
||||
scheduledRetryAttempt?: number | null;
|
||||
} | null;
|
||||
rootWakeRequestedByActorType: string | null;
|
||||
probe: StrandedFeedbackDeliveryBackstopProbe;
|
||||
maxAutomaticRetries?: number;
|
||||
}): StrandedFeedbackDeliveryBackstopDecision {
|
||||
if (input.issue.status !== "in_review") {
|
||||
return { kind: "skip", reason: "issue is not in review" };
|
||||
}
|
||||
if (input.issue.assigneeUserId) {
|
||||
return { kind: "skip", reason: "review is owned by a user" };
|
||||
}
|
||||
if (!input.issue.assigneeAgentId) {
|
||||
return { kind: "skip", reason: "review issue has no agent assignee" };
|
||||
}
|
||||
if (input.hasTypedReviewParticipant) {
|
||||
return { kind: "skip", reason: "review issue has a typed execution participant" };
|
||||
}
|
||||
if (!input.latestRun) {
|
||||
return { kind: "skip", reason: "review issue has no latest run" };
|
||||
}
|
||||
if (input.latestRun.status !== "failed" && input.latestRun.status !== "timed_out") {
|
||||
return { kind: "skip", reason: "latest run did not fail" };
|
||||
}
|
||||
if (input.latestRun.agentId !== input.issue.assigneeAgentId) {
|
||||
return { kind: "skip", reason: "latest failed run belongs to a different agent" };
|
||||
}
|
||||
|
||||
const delivery = readFeedbackDeliveryRunContext({
|
||||
companyId: input.companyId,
|
||||
run: input.latestRun,
|
||||
});
|
||||
if (!delivery || delivery.issueId !== input.issue.id) {
|
||||
return { kind: "skip", reason: "latest failed run carried no feedback delivery context" };
|
||||
}
|
||||
if (input.rootWakeRequestedByActorType !== FEEDBACK_DELIVERY_ROOT_WAKE_ACTOR_TYPE) {
|
||||
return { kind: "skip", reason: "feedback wake did not originate from a user" };
|
||||
}
|
||||
|
||||
const probe = input.probe;
|
||||
if (probe.hasActiveExecutionPath) return { kind: "skip", reason: "issue already has an active execution path" };
|
||||
if (probe.hasPendingWakeInteraction) return { kind: "skip", reason: "issue has a pending waking interaction" };
|
||||
if (probe.hasPendingApproval) return { kind: "skip", reason: "issue has a pending approval" };
|
||||
if (probe.hasPersistedDurableWaitPath) return { kind: "skip", reason: "issue has a persisted durable wait path" };
|
||||
if (probe.hasQueuedIssueWake) return { kind: "skip", reason: "issue already has a queued wake" };
|
||||
if (probe.pauseHeld) return { kind: "skip", reason: "automatic recovery is suppressed by a pause hold" };
|
||||
// The immediate lane owns this issue; never open a parallel second lane.
|
||||
if (probe.hasActiveRecoveryAction) return { kind: "skip", reason: "issue already has an active recovery action" };
|
||||
|
||||
// Exhaustion is checked before the hold/budget gates: the recovery action is a
|
||||
// visibility record with a `manual_repair_required` wake policy, so recording
|
||||
// it never wakes a held or budget-gated assignee, and exhausted feedback must
|
||||
// not stay silently `in_review`.
|
||||
const maxAutomaticRetries = input.maxAutomaticRetries ?? FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES;
|
||||
if (delivery.generation >= maxAutomaticRetries) {
|
||||
return { kind: "recovery_action", delivery, reason: "retry_exhausted" };
|
||||
}
|
||||
|
||||
// Replay would wake the assignee, so a deliberate hold or budget gate is a
|
||||
// legitimate wait that keeps its own lane as the single owner.
|
||||
if (probe.assigneeHold) {
|
||||
return { kind: "skip", reason: `assignee is held (${probe.assigneeHold})` };
|
||||
}
|
||||
if (probe.budgetBlocked) return { kind: "skip", reason: "assignee invocation is budget blocked" };
|
||||
|
||||
if (!probe.assigneeInvokable) {
|
||||
return { kind: "recovery_action", delivery, reason: "assignee_not_invokable" };
|
||||
}
|
||||
|
||||
const generation = delivery.generation + 1;
|
||||
const idempotencyKey = buildFeedbackDeliveryRetryIdempotencyKey({
|
||||
fingerprint: delivery.fingerprint,
|
||||
generation,
|
||||
});
|
||||
if (probe.hasExistingRetryWake) {
|
||||
return { kind: "skip", reason: "feedback delivery replay wake already exists" };
|
||||
}
|
||||
|
||||
const latestCommentId = delivery.commentIds.at(-1);
|
||||
const payload = withRecoveryModelProfileHint({
|
||||
issueId: input.issue.id,
|
||||
taskId: input.issue.id,
|
||||
sourceIssueId: input.issue.id,
|
||||
retryOfRunId: input.latestRun.id,
|
||||
retryReason: FEEDBACK_DELIVERY_RETRY_REASON,
|
||||
feedbackDeliveryRootWakeupRequestId: delivery.rootWakeupRequestId,
|
||||
feedbackDeliverySourceRunId: input.latestRun.id,
|
||||
feedbackDeliveryRetryGeneration: generation,
|
||||
feedbackDeliveryBackstop: true,
|
||||
[FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY]: delivery.commentIds,
|
||||
...(latestCommentId ? { commentId: latestCommentId, wakeCommentId: latestCommentId } : {}),
|
||||
}, "normal_model");
|
||||
|
||||
return {
|
||||
kind: "requeue",
|
||||
delivery,
|
||||
generation,
|
||||
idempotencyKey,
|
||||
payload,
|
||||
contextSnapshot: withRecoveryModelProfileHint({
|
||||
...payload,
|
||||
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
source: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
|
||||
instruction: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_INSTRUCTION,
|
||||
}, "normal_model"),
|
||||
};
|
||||
}
|
||||
@@ -28,6 +28,29 @@ export type {
|
||||
IssueLivenessSeverity,
|
||||
IssueLivenessState,
|
||||
} from "./issue-graph-liveness.js";
|
||||
export {
|
||||
FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
|
||||
FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND,
|
||||
FEEDBACK_DELIVERY_RECOVERY_CAUSE,
|
||||
FEEDBACK_DELIVERY_RETRY_REASON,
|
||||
FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
STRANDED_FEEDBACK_DELIVERY_BACKSTOP_INSTRUCTION,
|
||||
STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION,
|
||||
STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
|
||||
buildFeedbackDeliveryFingerprint,
|
||||
buildFeedbackDeliveryRetryIdempotencyKey,
|
||||
carriesExplicitFeedbackResume,
|
||||
decideStrandedFeedbackDeliveryBackstop,
|
||||
isEligibleFeedbackDeliveryWake,
|
||||
readFeedbackDeliveryRetryGeneration,
|
||||
readFeedbackDeliveryRunContext,
|
||||
readFeedbackDeliveryWakeCommentIds,
|
||||
} from "./feedback-delivery.js";
|
||||
export type {
|
||||
FeedbackDeliveryRunContext,
|
||||
StrandedFeedbackDeliveryBackstopDecision,
|
||||
StrandedFeedbackDeliveryBackstopProbe,
|
||||
} from "./feedback-delivery.js";
|
||||
export {
|
||||
recoveryService,
|
||||
} from "./service.js";
|
||||
|
||||
@@ -51,7 +51,7 @@ import {
|
||||
buildIssueBlockersResolvedWakeIdempotencyKey,
|
||||
findExistingIssueBlockersResolvedWakeForAnyKey,
|
||||
} from "../issue-dependency-wakeups.js";
|
||||
import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js";
|
||||
import { evaluateAgentInvokabilityFromDb, type AgentInvokability } from "../agent-invokability.js";
|
||||
import { getRunLogStore } from "../run-log-store.js";
|
||||
import {
|
||||
DEFAULT_MAX_SUCCESSFUL_RUN_HANDOFF_ATTEMPTS,
|
||||
@@ -82,6 +82,18 @@ import {
|
||||
withRecoveryModelProfileHint,
|
||||
} from "./model-profile-hint.js";
|
||||
import { isAutomaticRecoverySuppressedByPauseHold } from "./pause-hold-guard.js";
|
||||
import {
|
||||
FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
|
||||
FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND,
|
||||
FEEDBACK_DELIVERY_RECOVERY_CAUSE,
|
||||
FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION,
|
||||
STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
|
||||
buildFeedbackDeliveryRetryIdempotencyKey,
|
||||
decideStrandedFeedbackDeliveryBackstop,
|
||||
readFeedbackDeliveryRunContext,
|
||||
type StrandedFeedbackDeliveryBackstopDecision,
|
||||
} from "./feedback-delivery.js";
|
||||
|
||||
const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;
|
||||
const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["interrupted", "failed", "cancelled", "timed_out"] as const;
|
||||
@@ -3638,6 +3650,366 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
readNonEmptyString(monitor.externalRef) === latestRun.id;
|
||||
}
|
||||
|
||||
async function getLatestFeedbackDeliveryRun(companyId: string, issueId: string) {
|
||||
return db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
error: heartbeatRuns.error,
|
||||
wakeupRequestId: heartbeatRuns.wakeupRequestId,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
livenessState: heartbeatRuns.livenessState,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
startedAt: heartbeatRuns.startedAt,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
scheduledRetryReason: heartbeatRuns.scheduledRetryReason,
|
||||
scheduledRetryAttempt: heartbeatRuns.scheduledRetryAttempt,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function evaluateIssueAssigneeInvokability(
|
||||
issue: typeof issues.$inferSelect,
|
||||
): Promise<AgentInvokability> {
|
||||
if (!issue.assigneeAgentId) {
|
||||
return { invokable: false, reason: "missing", message: "Issue has no agent assignee", details: {}, invalidOrgChain: false };
|
||||
}
|
||||
const agent = await getAgent(issue.assigneeAgentId);
|
||||
if (!agent || agent.companyId !== issue.companyId) {
|
||||
return { invokable: false, reason: "missing", message: "Agent no longer exists", details: {}, invalidOrgChain: false };
|
||||
}
|
||||
return evaluateAgentInvokabilityFromDb(db, agent);
|
||||
}
|
||||
|
||||
async function hasPendingIssueApproval(companyId: string, issueId: string) {
|
||||
return db
|
||||
.select({ id: issueApprovals.approvalId })
|
||||
.from(issueApprovals)
|
||||
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
|
||||
.where(
|
||||
and(
|
||||
eq(issueApprovals.companyId, companyId),
|
||||
eq(issueApprovals.issueId, issueId),
|
||||
inArray(approvals.status, ["pending", "revision_requested"]),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => Boolean(rows[0]));
|
||||
}
|
||||
|
||||
async function readRootFeedbackWakeActorType(input: {
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
rootWakeupRequestId: string;
|
||||
}) {
|
||||
return db
|
||||
.select({ requestedByActorType: agentWakeupRequests.requestedByActorType })
|
||||
.from(agentWakeupRequests)
|
||||
.where(
|
||||
and(
|
||||
eq(agentWakeupRequests.id, input.rootWakeupRequestId),
|
||||
eq(agentWakeupRequests.companyId, input.companyId),
|
||||
eq(agentWakeupRequests.agentId, input.agentId),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0]?.requestedByActorType ?? null);
|
||||
}
|
||||
|
||||
async function hasFeedbackDeliveryWakeForIdempotencyKey(companyId: string, idempotencyKey: string) {
|
||||
return db
|
||||
.select({ id: agentWakeupRequests.id })
|
||||
.from(agentWakeupRequests)
|
||||
.where(
|
||||
and(
|
||||
eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.idempotencyKey, idempotencyKey),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => Boolean(rows[0]));
|
||||
}
|
||||
|
||||
/**
|
||||
* The immediate lane logs `issue.feedback_retry_queued` keyed on the root wake
|
||||
* id. Re-using the same dedupe check keeps the audit trail single-entry per
|
||||
* feedback batch no matter which lane queued the replay.
|
||||
*/
|
||||
async function hasFeedbackDeliveryActivity(input: {
|
||||
companyId: string;
|
||||
issueId: string;
|
||||
action: string;
|
||||
rootWakeupRequestId: string;
|
||||
}) {
|
||||
return db
|
||||
.select({ id: activityLog.id })
|
||||
.from(activityLog)
|
||||
.where(
|
||||
and(
|
||||
eq(activityLog.companyId, input.companyId),
|
||||
eq(activityLog.entityType, "issue"),
|
||||
eq(activityLog.entityId, input.issueId),
|
||||
eq(activityLog.action, input.action),
|
||||
sql`${activityLog.details} ->> 'rootWakeupRequestId' = ${input.rootWakeupRequestId}`,
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => Boolean(rows[0]));
|
||||
}
|
||||
|
||||
async function ensureStrandedFeedbackDeliveryRecoveryAction(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
latestRun: Awaited<ReturnType<typeof getLatestFeedbackDeliveryRun>>;
|
||||
decision: Extract<StrandedFeedbackDeliveryBackstopDecision, { kind: "recovery_action" }>;
|
||||
}) {
|
||||
const delivery = input.decision.delivery;
|
||||
const ownerAgentId = input.decision.reason === "assignee_not_invokable"
|
||||
? await resolveStrandedIssueRecoveryOwnerAgentId(input.issue)
|
||||
: await resolveInvokableRecoveryAgentId(input.issue, delivery.agentId)
|
||||
?? await resolveStrandedIssueRecoveryOwnerAgentId(input.issue);
|
||||
const ownerName = ownerAgentId
|
||||
? (await getAgent(ownerAgentId))?.name ?? null
|
||||
: null;
|
||||
const routingFallbackReason = ownerAgentId === delivery.agentId
|
||||
? null
|
||||
: "The original assignee is not invokable; stranded feedback recovery fell through to the manager ladder.";
|
||||
|
||||
const action = await recoveryActionsSvc.upsertSourceScoped({
|
||||
companyId: input.issue.companyId,
|
||||
sourceIssueId: input.issue.id,
|
||||
kind: FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND,
|
||||
ownerType: ownerAgentId ? "agent" : "board",
|
||||
ownerAgentId,
|
||||
previousOwnerAgentId: delivery.agentId,
|
||||
returnOwnerAgentId: delivery.agentId,
|
||||
cause: FEEDBACK_DELIVERY_RECOVERY_CAUSE,
|
||||
fingerprint: delivery.fingerprint,
|
||||
evidence: {
|
||||
sourceIssueId: input.issue.id,
|
||||
sourceIdentifier: input.issue.identifier,
|
||||
previousStatus: "in_review",
|
||||
detectedBy: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
|
||||
backstopReason: input.decision.reason,
|
||||
sourceRunId: input.latestRun?.id ?? null,
|
||||
latestRunStatus: input.latestRun?.status ?? null,
|
||||
lastFailureCode: input.latestRun?.errorCode ?? null,
|
||||
rootWakeupRequestId: delivery.rootWakeupRequestId,
|
||||
outstandingCommentIds: delivery.commentIds,
|
||||
retryGeneration: delivery.generation,
|
||||
maxAutomaticRetries: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
|
||||
failureSummary: summarizeRunFailureForIssueComment(input.latestRun ?? null)?.trim() ?? null,
|
||||
routingFallbackReason,
|
||||
},
|
||||
nextAction: [
|
||||
ownerName ? `Owner: ${ownerName}.` : "Owner: the Board (no invokable recovery agent).",
|
||||
STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION,
|
||||
].join(" "),
|
||||
wakePolicy: {
|
||||
type: "manual_repair_required",
|
||||
reason: FEEDBACK_DELIVERY_RECOVERY_CAUSE,
|
||||
ownerAgentId,
|
||||
},
|
||||
maxAttempts: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
|
||||
lastAttemptAt: new Date(),
|
||||
});
|
||||
|
||||
if (
|
||||
!(await hasFeedbackDeliveryActivity({
|
||||
companyId: input.issue.companyId,
|
||||
issueId: input.issue.id,
|
||||
action: "issue.feedback_recovery_exhausted",
|
||||
rootWakeupRequestId: delivery.rootWakeupRequestId,
|
||||
}))
|
||||
) {
|
||||
await logActivity(db, {
|
||||
companyId: input.issue.companyId,
|
||||
actorType: "system",
|
||||
actorId: "recovery",
|
||||
agentId: delivery.agentId,
|
||||
action: "issue.feedback_recovery_exhausted",
|
||||
entityType: "issue",
|
||||
entityId: input.issue.id,
|
||||
details: {
|
||||
recoveryActionId: action.id,
|
||||
detectedBy: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
|
||||
backstopReason: input.decision.reason,
|
||||
sourceRunId: input.latestRun?.id ?? null,
|
||||
rootWakeupRequestId: delivery.rootWakeupRequestId,
|
||||
sourceCommentIds: delivery.commentIds,
|
||||
retryGeneration: delivery.generation,
|
||||
ownerAgentId,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return action;
|
||||
}
|
||||
|
||||
/**
|
||||
* PAP-17302: periodic backstop for the narrow review-feedback shape immediate
|
||||
* terminalization can miss — agent-assigned `in_review` with no typed review
|
||||
* participant, no interaction/approval/monitor/user owner, no active run or
|
||||
* queued wake, and a failed latest run that carried user feedback context.
|
||||
*
|
||||
* Dedupes against the immediate retry/recovery lane by sharing its wake
|
||||
* idempotency key, recovery-action fingerprint, and activity keys, so this can
|
||||
* never open a parallel second delivery lane.
|
||||
*/
|
||||
async function reconcileStrandedFeedbackDelivery(input: {
|
||||
issue: typeof issues.$inferSelect;
|
||||
hasTypedReviewParticipant: boolean;
|
||||
}): Promise<{ outcome: "requeued" | "recovery_action" | "skipped"; reason?: string }> {
|
||||
const latestRun = await getLatestFeedbackDeliveryRun(input.issue.companyId, input.issue.id);
|
||||
const candidate = latestRun
|
||||
? readFeedbackDeliveryRunContext({ companyId: input.issue.companyId, run: latestRun })
|
||||
: null;
|
||||
// Cheap pure shape gate first: most `in_review` issues without a typed
|
||||
// participant are legitimate waits, and they must not pay for the probes.
|
||||
if (!latestRun || !candidate || candidate.issueId !== input.issue.id) {
|
||||
return { outcome: "skipped", reason: "not_stranded_feedback_shape" };
|
||||
}
|
||||
|
||||
const retryIdempotencyKey = buildFeedbackDeliveryRetryIdempotencyKey({
|
||||
fingerprint: candidate.fingerprint,
|
||||
generation: candidate.generation + 1,
|
||||
});
|
||||
const [
|
||||
rootWakeRequestedByActorType,
|
||||
activeExecutionPath,
|
||||
pendingWakeInteraction,
|
||||
pendingApproval,
|
||||
durableWaitPath,
|
||||
queuedIssueWake,
|
||||
activeRecoveryAction,
|
||||
existingRetryWake,
|
||||
assigneeInvokability,
|
||||
budgetBlocked,
|
||||
] = await Promise.all([
|
||||
readRootFeedbackWakeActorType({
|
||||
companyId: input.issue.companyId,
|
||||
agentId: candidate.agentId,
|
||||
rootWakeupRequestId: candidate.rootWakeupRequestId,
|
||||
}),
|
||||
hasActiveExecutionPath(input.issue.companyId, input.issue.id),
|
||||
hasPendingWakeInteraction(input.issue.companyId, input.issue.id),
|
||||
hasPendingIssueApproval(input.issue.companyId, input.issue.id),
|
||||
hasPersistedDurableWaitPath(input.issue),
|
||||
hasQueuedIssueWake(input.issue.companyId, input.issue.id),
|
||||
recoveryActionsSvc.getActiveForIssue(input.issue.companyId, input.issue.id),
|
||||
hasFeedbackDeliveryWakeForIdempotencyKey(input.issue.companyId, retryIdempotencyKey),
|
||||
evaluateIssueAssigneeInvokability(input.issue),
|
||||
input.issue.assigneeAgentId
|
||||
? isInvocationBudgetBlocked(input.issue, input.issue.assigneeAgentId)
|
||||
: Promise.resolve(false),
|
||||
]);
|
||||
|
||||
const decision = decideStrandedFeedbackDeliveryBackstop({
|
||||
companyId: input.issue.companyId,
|
||||
issue: {
|
||||
id: input.issue.id,
|
||||
status: input.issue.status,
|
||||
assigneeAgentId: input.issue.assigneeAgentId,
|
||||
assigneeUserId: input.issue.assigneeUserId,
|
||||
},
|
||||
hasTypedReviewParticipant: input.hasTypedReviewParticipant,
|
||||
latestRun,
|
||||
rootWakeRequestedByActorType,
|
||||
probe: {
|
||||
hasActiveExecutionPath: activeExecutionPath,
|
||||
hasPendingWakeInteraction: pendingWakeInteraction,
|
||||
hasPendingApproval: pendingApproval,
|
||||
hasPersistedDurableWaitPath: durableWaitPath,
|
||||
hasQueuedIssueWake: queuedIssueWake,
|
||||
hasActiveRecoveryAction: Boolean(activeRecoveryAction),
|
||||
hasExistingRetryWake: existingRetryWake,
|
||||
assigneeInvokable: assigneeInvokability.invokable,
|
||||
assigneeHold: !assigneeInvokability.invokable &&
|
||||
(assigneeInvokability.reason === "paused" || assigneeInvokability.reason === "pending_approval")
|
||||
? assigneeInvokability.reason
|
||||
: null,
|
||||
budgetBlocked,
|
||||
// The caller gates the whole scan on pause holds before reaching here.
|
||||
pauseHeld: false,
|
||||
},
|
||||
});
|
||||
|
||||
if (decision.kind === "skip") {
|
||||
return { outcome: "skipped", reason: decision.reason };
|
||||
}
|
||||
|
||||
if (decision.kind === "recovery_action") {
|
||||
await ensureStrandedFeedbackDeliveryRecoveryAction({
|
||||
issue: input.issue,
|
||||
latestRun,
|
||||
decision,
|
||||
});
|
||||
return { outcome: "recovery_action", reason: decision.reason };
|
||||
}
|
||||
|
||||
const queued = await deps.enqueueWakeup(decision.delivery.agentId, {
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
|
||||
idempotencyKey: decision.idempotencyKey,
|
||||
payload: decision.payload,
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
contextSnapshot: decision.contextSnapshot,
|
||||
});
|
||||
if (!queued) {
|
||||
return { outcome: "skipped", reason: "replay_wake_not_queued" };
|
||||
}
|
||||
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({ retryOfRunId: latestRun.id, updatedAt: new Date() })
|
||||
.where(eq(heartbeatRuns.id, queued.id));
|
||||
|
||||
if (
|
||||
!(await hasFeedbackDeliveryActivity({
|
||||
companyId: input.issue.companyId,
|
||||
issueId: input.issue.id,
|
||||
action: "issue.feedback_retry_queued",
|
||||
rootWakeupRequestId: decision.delivery.rootWakeupRequestId,
|
||||
}))
|
||||
) {
|
||||
await logActivity(db, {
|
||||
companyId: input.issue.companyId,
|
||||
actorType: "system",
|
||||
actorId: "recovery",
|
||||
agentId: decision.delivery.agentId,
|
||||
runId: queued.id,
|
||||
action: "issue.feedback_retry_queued",
|
||||
entityType: "issue",
|
||||
entityId: input.issue.id,
|
||||
details: {
|
||||
detectedBy: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
|
||||
sourceRunId: latestRun.id,
|
||||
sourceWakeupRequestId: latestRun.wakeupRequestId,
|
||||
rootWakeupRequestId: decision.delivery.rootWakeupRequestId,
|
||||
sourceCommentIds: decision.delivery.commentIds,
|
||||
retryRunId: queued.id,
|
||||
retryWakeupRequestId: queued.wakeupRequestId,
|
||||
retryGeneration: decision.generation,
|
||||
disposition: "queued",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return { outcome: "requeued" };
|
||||
}
|
||||
|
||||
async function reconcileStrandedAssignedIssues(opts?: { issueCreatedAtGte?: Date | null }) {
|
||||
const candidates = await db
|
||||
.select()
|
||||
@@ -3663,6 +4035,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
orphanBlockersAssigned: 0,
|
||||
successfulRunHandoffEscalated: 0,
|
||||
reviewParticipantRequeued: 0,
|
||||
strandedFeedbackRequeued: 0,
|
||||
strandedFeedbackRecoveryActions: 0,
|
||||
escalated: 0,
|
||||
waitingOnReviewResolved: 0,
|
||||
providerQuotaMonitored: 0,
|
||||
@@ -3893,7 +4267,22 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
|
||||
if (issue.status === "in_review") {
|
||||
if (!participantAgentId || !pendingExecutionState) {
|
||||
result.skipped += 1;
|
||||
// PAP-17302: an `in_review` issue with no typed participant used to be
|
||||
// skipped unconditionally, which stranded review feedback whose only
|
||||
// path was a failed comment run. Everything else stays skipped.
|
||||
const feedbackBackstop = await reconcileStrandedFeedbackDelivery({
|
||||
issue,
|
||||
hasTypedReviewParticipant: Boolean(participantAgentId && pendingExecutionState),
|
||||
});
|
||||
if (feedbackBackstop.outcome === "requeued") {
|
||||
result.strandedFeedbackRequeued += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else if (feedbackBackstop.outcome === "recovery_action") {
|
||||
result.strandedFeedbackRecoveryActions += 1;
|
||||
result.issueIds.push(issue.id);
|
||||
} else {
|
||||
result.skipped += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
const participantLatestRun = participantLatestRunForRecovery;
|
||||
|
||||
Reference in new issue
Block a user