fix(server): recover lost human feedback delivery

Replay feedback lost before process launch, reconcile stranded review feedback, and project durable delivery state onto source comments.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-16 20:05:21 +00:00
1 parent 623f7aa555
commit a6c28fa414
16 files changed
+3762 -33

No files matched your search

+18
View File
@@ -376,6 +376,7 @@ export const ISSUE_SURFACE_VISIBILITIES = ["default", "plugin_operation"] as con
export type IssueSurfaceVisibility = (typeof ISSUE_SURFACE_VISIBILITIES)[number];
export const ISSUE_RECOVERY_ACTION_KINDS = [
"feedback_delivery",
"missing_disposition",
"stranded_assigned_issue",
"workspace_validation",
@@ -401,6 +402,23 @@ export const ISSUE_RECOVERY_ACTION_OWNER_TYPES = [
] as const;
export type IssueRecoveryActionOwnerType = (typeof ISSUE_RECOVERY_ACTION_OWNER_TYPES)[number];
/**
* Delivery state of one human feedback comment, as projected onto that comment
* for operator surfaces. This is deliberately *not* a task status: the task
* keeps its own workflow status while a delivery attempt is in flight.
*
* - `retrying` — automatic replay is still in flight; no operator action.
* - `exhausted_needs_attention` — automatic replay gave up; a human must act.
* - `recovered_after_attention` — delivery succeeded after it had needed
* attention, so the resolution stays visible in thread history.
*/
export const ISSUE_FEEDBACK_DELIVERY_STATES = [
"retrying",
"exhausted_needs_attention",
"recovered_after_attention",
] as const;
export type IssueFeedbackDeliveryState = (typeof ISSUE_FEEDBACK_DELIVERY_STATES)[number];
export const ISSUE_RECOVERY_ACTION_OUTCOMES = [
"restored",
"handed_back",
+4
View File
@@ -299,6 +299,7 @@ export {
ONBOARDING_FIRST_TASK_ORIGIN_KIND,
ISSUE_WATCHDOG_DISCOVERY_KINDS,
ISSUE_SURFACE_VISIBILITIES,
ISSUE_FEEDBACK_DELIVERY_STATES,
ISSUE_RECOVERY_ACTION_KINDS,
ISSUE_RECOVERY_ACTION_STATUSES,
ISSUE_RECOVERY_ACTION_OWNER_TYPES,
@@ -492,6 +493,7 @@ export {
type IssueOriginKind,
type IssueWatchdogDiscoveryKind,
type IssueSurfaceVisibility,
type IssueFeedbackDeliveryState,
type IssueRecoveryActionKind,
type IssueRecoveryActionStatus,
type IssueRecoveryActionOwnerType,
@@ -1008,6 +1010,8 @@ export type {
IssueExecutionStagePrincipal,
IssueExecutionDecision,
IssueComment,
IssueCommentFeedbackDelivery,
IssueFeedbackDeliveryRetryResponse,
IssueCommentDerivedAuthorSource,
IssueCommentMetadata,
IssueCommentMetadataSection,
+2
View File
@@ -649,6 +649,8 @@ export type {
IssueReviewRequest,
IssueExecutionDecision,
IssueComment,
IssueCommentFeedbackDelivery,
IssueFeedbackDeliveryRetryResponse,
IssueCommentDerivedAuthorSource,
IssueCommentMetadata,
IssueCommentMetadataSection,
+55
View File
@@ -14,6 +14,7 @@ import type {
IssueReferenceSourceKind,
IssueExecutionStageType,
IssueExecutionStateStatus,
IssueFeedbackDeliveryState,
IssueHarnessKind,
IssueOriginKind,
IssuePriority,
@@ -941,6 +942,54 @@ export type IssueCommentDerivedAuthorSource =
| "run_id"
| "run_log_comment_post";
/**
* Delivery state projected onto a single human feedback comment.
*
* Every field here is safe to render to any operator who can already read the
* comment: raw adapter errors, stack traces, request bodies, credentials, and
* filesystem paths are mapped to a short `safeFailureReason` code server-side
* and never passed through verbatim.
*/
export interface IssueCommentFeedbackDelivery {
state: IssueFeedbackDeliveryState;
/** The human comment this delivery obligation belongs to. */
sourceCommentId: string;
/** Completed automatic replay generations, or `null` when unknown. */
attempt: number | null;
/** Automatic replay budget, or `null` when unknown. */
maxAttempts: number | null;
nextAttemptAt: string | null;
lastAttemptAt: string | null;
deliveredAt: string | null;
/**
* Run the operator may open for context. Omitted (`null`) when no run exists
* or the viewer is not authorized to read it — never rendered as a disabled
* mystery action.
*/
targetRunId: string | null;
/** True when the viewer may ask for one more explicit delivery attempt. */
canRetry: boolean;
/**
* True when a replay is already queued for this obligation. Keeps the
* "Retrying…" affordance durable across reloads instead of living only in
* button-local component state.
*/
retryInFlight: boolean;
/** Short operator-safe reason code, never a raw adapter error. */
safeFailureReason: string | null;
/**
* How many human comments share this one delivery obligation (a batched wake
* can cover several). Used for the thread rollup copy.
*/
batchedCommentCount: number;
}
export interface IssueFeedbackDeliveryRetryResponse {
outcome: "queued" | "already_queued" | "not_exhausted" | "no_invokable_assignee";
message: string;
feedbackDelivery: IssueCommentFeedbackDelivery | null;
}
export interface IssueComment {
id: string;
companyId: string;
@@ -964,6 +1013,12 @@ export interface IssueComment {
deletedByRunId?: string | null;
sourceTrust?: SourceTrustMetadata | null;
followUpRequested?: boolean;
/**
* Present only while this human comment's delivery is retrying, exhausted, or
* freshly recovered. Absent for the overwhelming majority of comments, which
* were delivered on the first attempt.
*/
feedbackDelivery?: IssueCommentFeedbackDelivery | null;
createdAt: Date;
updatedAt: Date;
}
@@ -100,12 +100,18 @@ vi.mock("../adapters/index.ts", async () => {
});
import {
FEEDBACK_DELIVERY_RETRY_REASON,
FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
INTERACTION_CONTINUATION_INFRA_RETRY_REASON,
INTERACTION_CONTINUATION_INFRA_WAKE_REASON,
heartbeatService,
redactDetectedSuccessfulRunProgressSummaryForBoard,
redactSuccessfulRunHandoffEvidence,
} from "../services/heartbeat.ts";
import {
buildFeedbackDeliveryFingerprint,
buildFeedbackDeliveryRetryIdempotencyKey,
} from "../services/recovery/feedback-delivery.ts";
import {
readHotRestartIntent,
resolveLegacyHotRestartIntentPath,
@@ -569,6 +575,69 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
return { companyId, agentId, runId, wakeupRequestId, issueId };
}
async function prepareFeedbackWake(
fixture: Awaited<ReturnType<typeof seedRunFixture>>,
input?: {
commentId?: string;
retryGeneration?: number;
rootWakeupRequestId?: string;
issueStatus?: "in_progress" | "in_review" | "done";
},
) {
const commentId = input?.commentId ?? randomUUID();
const rootWakeupRequestId = input?.rootWakeupRequestId ?? fixture.wakeupRequestId;
const retryGeneration = input?.retryGeneration ?? 0;
const createdAt = new Date("2026-03-19T00:00:01.000Z");
await db.insert(issueComments).values({
id: commentId,
companyId: fixture.companyId,
issueId: fixture.issueId,
authorType: "user",
authorUserId: "responsible-user",
body: "Please incorporate this review feedback.",
createdAt,
updatedAt: createdAt,
});
await db
.update(agentWakeupRequests)
.set({
source: "automation",
triggerDetail: "user",
reason: retryGeneration > 0 ? FEEDBACK_DELIVERY_RETRY_WAKE_REASON : "issue_commented",
payload: { issueId: fixture.issueId, commentId },
requestedByActorType: "user",
requestedByActorId: "responsible-user",
})
.where(eq(agentWakeupRequests.id, rootWakeupRequestId));
await db
.update(heartbeatRuns)
.set({
contextSnapshot: {
issueId: fixture.issueId,
taskId: fixture.issueId,
wakeReason: retryGeneration > 0 ? FEEDBACK_DELIVERY_RETRY_WAKE_REASON : "issue_commented",
commentId,
wakeCommentId: commentId,
wakeCommentIds: [commentId],
...(retryGeneration > 0
? {
retryReason: FEEDBACK_DELIVERY_RETRY_REASON,
feedbackDeliveryRootWakeupRequestId: rootWakeupRequestId,
feedbackDeliveryRetryGeneration: retryGeneration,
}
: {}),
},
})
.where(eq(heartbeatRuns.id, fixture.runId));
await db
.update(issues)
.set({ status: input?.issueStatus ?? "in_progress" })
.where(eq(issues.id, fixture.issueId));
return { commentId, rootWakeupRequestId };
}
async function seedEnvironmentLeaseFixture(input: {
companyId: string;
runId: string;
@@ -911,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";
}) {
@@ -1368,6 +1626,337 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
expect(checkoutReleasedIssue?.checkoutRunId).toBeNull();
});
it("replays a user feedback wake once when restart loses it before process identity exists", async () => {
let releaseAdapter: (() => void) | null = null;
const adapterStarted = new Promise<void>((resolve) => {
mockAdapterExecute.mockImplementationOnce(async () => {
resolve();
await new Promise<void>((release) => {
releaseAdapter = release;
});
return {
exitCode: 0,
signal: null,
timedOut: false,
errorMessage: null,
summary: "Handled the recovered feedback.",
provider: "test",
model: "test-model",
};
});
});
const fixture = await seedRunFixture({
agentStatus: "idle",
processPid: null,
processGroupId: null,
});
const { commentId } = await prepareFeedbackWake(fixture);
const heartbeat = heartbeatService(db);
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] });
await Promise.race([
adapterStarted,
new Promise<never>((_, reject) => setTimeout(() => reject(new Error("feedback retry did not start")), 3_000)),
]);
const runs = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.agentId, fixture.agentId));
expect(runs).toHaveLength(2);
const retryRun = runs.find((run) => run.id !== fixture.runId);
expect(retryRun).toMatchObject({
retryOfRunId: fixture.runId,
scheduledRetryAttempt: 1,
scheduledRetryReason: FEEDBACK_DELIVERY_RETRY_REASON,
});
expect(retryRun?.contextSnapshot).toMatchObject({
issueId: fixture.issueId,
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
retryReason: FEEDBACK_DELIVERY_RETRY_REASON,
feedbackDeliveryRootWakeupRequestId: fixture.wakeupRequestId,
feedbackDeliverySourceRunId: fixture.runId,
feedbackDeliveryRetryGeneration: 1,
wakeCommentIds: [commentId],
});
const retryWake = await db
.select()
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.runId, retryRun?.id ?? ""))
.then((rows) => rows[0] ?? null);
expect(retryWake).toMatchObject({
reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
requestedByActorType: "system",
});
const auditRows = await db
.select()
.from(activityLog)
.where(and(
eq(activityLog.runId, fixture.runId),
eq(activityLog.action, "issue.feedback_retry_queued"),
));
expect(auditRows).toHaveLength(1);
expect(auditRows[0]?.details).toMatchObject({
rootWakeupRequestId: fixture.wakeupRequestId,
sourceCommentIds: [commentId],
retryRunId: retryRun?.id,
retryGeneration: 1,
});
if (!retryRun || !releaseAdapter) throw new Error("feedback retry did not reach the adapter");
await db.insert(issueComments).values({
companyId: fixture.companyId,
issueId: fixture.issueId,
authorType: "agent",
authorAgentId: fixture.agentId,
createdByRunId: retryRun.id,
body: "Applied the requested feedback.",
});
await db.update(issues).set({ status: "done" }).where(eq(issues.id, fixture.issueId));
releaseAdapter();
expect((await waitForRunToSettle(heartbeat, retryRun.id, 5_000))?.status).toBe("succeeded");
await heartbeat.waitForRunExecutionDrain(retryRun.id);
const finalRuns = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.agentId, fixture.agentId));
expect(finalRuns).toHaveLength(2);
});
it("creates one source-scoped recovery action when the feedback replay is exhausted", async () => {
const fixture = await seedRunFixture({
agentStatus: "idle",
processPid: null,
processGroupId: null,
});
const { commentId } = await prepareFeedbackWake(fixture, { retryGeneration: 1 });
const heartbeat = heartbeatService(db);
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] });
const runs = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.agentId, fixture.agentId));
expect(runs).toHaveLength(1);
const recoveryRows = await db
.select()
.from(issueRecoveryActions)
.where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId));
expect(recoveryRows).toHaveLength(1);
expect(recoveryRows[0]).toMatchObject({
kind: "feedback_delivery",
status: "active",
ownerAgentId: fixture.agentId,
cause: "feedback_delivery_exhausted",
maxAttempts: 1,
});
expect(recoveryRows[0]?.evidence).toMatchObject({
rootWakeupRequestId: fixture.wakeupRequestId,
outstandingCommentIds: [commentId],
retryGeneration: 1,
});
const recoveryAudits = await db
.select()
.from(activityLog)
.where(and(
eq(activityLog.runId, fixture.runId),
eq(activityLog.action, "issue.feedback_recovery_exhausted"),
));
expect(recoveryAudits).toHaveLength(1);
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 0, runIds: [] });
expect(await db
.select({ id: issueRecoveryActions.id })
.from(issueRecoveryActions)
.where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))).toHaveLength(1);
});
it("preserves terminal issue gates instead of replaying stale feedback", async () => {
const fixture = await seedRunFixture({
agentStatus: "idle",
processPid: null,
processGroupId: null,
});
await prepareFeedbackWake(fixture, { issueStatus: "done" });
const heartbeat = heartbeatService(db);
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] });
expect(await db
.select({ id: heartbeatRuns.id })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.agentId, fixture.agentId))).toHaveLength(1);
expect(await db
.select({ id: issueRecoveryActions.id })
.from(issueRecoveryActions)
.where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))).toHaveLength(0);
const issue = await db
.select({ status: issues.status, executionRunId: issues.executionRunId })
.from(issues)
.where(eq(issues.id, fixture.issueId))
.then((rows) => rows[0] ?? null);
expect(issue).toEqual({ status: "done", executionRunId: null });
});
it("preserves pause gates and records recovery instead of invoking the agent", async () => {
const fixture = await seedRunFixture({
agentStatus: "paused",
processPid: null,
processGroupId: null,
});
await prepareFeedbackWake(fixture);
const heartbeat = heartbeatService(db);
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] });
expect(mockAdapterExecute).not.toHaveBeenCalled();
expect(await db
.select({ id: heartbeatRuns.id })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.agentId, fixture.agentId))).toHaveLength(1);
const recovery = await db
.select()
.from(issueRecoveryActions)
.where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))
.then((rows) => rows[0] ?? null);
expect(recovery).toMatchObject({
kind: "feedback_delivery",
cause: "feedback_delivery_exhausted",
evidence: {
gateErrorCode: "agent_not_invokable",
},
});
});
it("coalesces a recovered feedback wake ahead of a newer queued comment without another handler", async () => {
let releaseAdapter: (() => void) | null = null;
const adapterStarted = new Promise<void>((resolve) => {
mockAdapterExecute.mockImplementationOnce(async () => {
resolve();
await new Promise<void>((release) => {
releaseAdapter = release;
});
return {
exitCode: 0,
signal: null,
timedOut: false,
errorMessage: null,
summary: "Handled the coalesced feedback.",
provider: "test",
model: "test-model",
};
});
});
const fixture = await seedRunFixture({
agentStatus: "idle",
processPid: null,
processGroupId: null,
});
const first = await prepareFeedbackWake(fixture);
const secondCommentId = randomUUID();
const secondWakeId = randomUUID();
const secondRunId = randomUUID();
const secondCreatedAt = new Date("2026-03-19T00:00:02.000Z");
await db.insert(issueComments).values({
id: secondCommentId,
companyId: fixture.companyId,
issueId: fixture.issueId,
authorType: "user",
authorUserId: "responsible-user",
body: "One more adjustment after the first review.",
createdAt: secondCreatedAt,
updatedAt: secondCreatedAt,
});
await db.insert(agentWakeupRequests).values({
id: secondWakeId,
companyId: fixture.companyId,
agentId: fixture.agentId,
source: "automation",
triggerDetail: "user",
reason: "issue_commented",
payload: { issueId: fixture.issueId, commentId: secondCommentId },
status: "queued",
runId: secondRunId,
requestedByActorType: "user",
requestedByActorId: "responsible-user",
requestedAt: secondCreatedAt,
updatedAt: secondCreatedAt,
});
await db.insert(heartbeatRuns).values({
id: secondRunId,
companyId: fixture.companyId,
agentId: fixture.agentId,
invocationSource: "automation",
triggerDetail: "user",
status: "queued",
wakeupRequestId: secondWakeId,
contextSnapshot: {
issueId: fixture.issueId,
taskId: fixture.issueId,
wakeReason: "issue_commented",
commentId: secondCommentId,
wakeCommentId: secondCommentId,
wakeCommentIds: [secondCommentId],
},
createdAt: secondCreatedAt,
updatedAt: secondCreatedAt,
});
await db
.update(issues)
.set({ executionRunId: secondRunId, updatedAt: secondCreatedAt })
.where(eq(issues.id, fixture.issueId));
const heartbeat = heartbeatService(db);
expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] });
await Promise.race([
adapterStarted,
new Promise<never>((_, reject) => setTimeout(() => reject(new Error("coalesced retry did not start")), 3_000)),
]);
const runs = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.agentId, fixture.agentId));
expect(runs).toHaveLength(2);
const coalescedRun = runs.find((run) => run.id === secondRunId);
expect(coalescedRun?.contextSnapshot).toMatchObject({
feedbackDeliveryRootWakeupRequestId: fixture.wakeupRequestId,
feedbackDeliveryRetryGeneration: 1,
wakeCommentIds: [first.commentId, secondCommentId],
commentId: secondCommentId,
wakeCommentId: secondCommentId,
});
const auditRows = await db
.select()
.from(activityLog)
.where(and(
eq(activityLog.runId, fixture.runId),
eq(activityLog.action, "issue.feedback_retry_queued"),
));
expect(auditRows).toHaveLength(1);
expect(auditRows[0]?.details).toMatchObject({
retryRunId: secondRunId,
disposition: "coalesced",
sourceCommentIds: [first.commentId],
});
if (!releaseAdapter) throw new Error("coalesced feedback run did not reach the adapter");
await db.insert(issueComments).values({
companyId: fixture.companyId,
issueId: fixture.issueId,
authorType: "agent",
authorAgentId: fixture.agentId,
createdByRunId: secondRunId,
body: "Applied both feedback comments in order.",
});
await db.update(issues).set({ status: "done" }).where(eq(issues.id, fixture.issueId));
releaseAdapter();
expect((await waitForRunToSettle(heartbeat, secondRunId, 5_000))?.status).toBe("succeeded");
await heartbeat.waitForRunExecutionDrain(secondRunId);
});
it("restores one lost monitor dispatch before escalating a second process loss", async () => {
const { companyId, agentId, runId, issueId } = await seedRunFixture({
adapterType: "openclaw_gateway",
@@ -5345,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();
+4
View File
@@ -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(
+164 -1
View File
@@ -88,6 +88,7 @@ import {
type IssueWakeDiagnosticWakeFailureClass,
type IssueWakeDiagnosticWakeRequest,
type IssueWakeDiagnosticsResponse,
type IssueFeedbackDeliveryRetryResponse,
type IssueRelationIssueSummary,
type IssueReviewPolicy,
type IssueThreadInteractionCanonicalResolverPolicy,
@@ -104,6 +105,14 @@ import {
type IssueWriteDenialContext,
} from "@paperclipai/shared";
import { trackAgentTaskCompleted } from "@paperclipai/shared/telemetry";
import {
buildFeedbackDeliveryManualRetryIdempotencyKey,
FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND,
FEEDBACK_DELIVERY_RECOVERY_CAUSE,
FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY,
} from "../services/recovery/feedback-delivery.js";
import { getTelemetryClient } from "../telemetry.js";
import { isUniqueViolation } from "../db-errors.js";
import type { StorageService } from "../storage/types.js";
@@ -116,6 +125,7 @@ import {
companyService,
companySearchService,
executionWorkspaceService,
feedbackDeliveryCommentStateService,
goalService,
heartbeatService,
issueApprovalService,
@@ -2756,6 +2766,7 @@ export function issueRoutes(
const goalsSvc = goalService(db);
const issueApprovalsSvc = issueApprovalService(db);
const recoveryActionsSvc = issueRecoveryActionService(db);
const feedbackDeliveryStatesSvc = feedbackDeliveryCommentStateService(db);
const executionWorkspacesSvc = executionWorkspaceServiceDirect(db);
const workProductsSvc = workProductService(db);
const documentsSvc = documentService(db);
@@ -10763,7 +10774,159 @@ export function issueRoutes(
order,
limit,
});
res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, comments));
// Operator surfaces only: an agent reading its own thread has no delivery
// banner to act on, and the run reference stays gated behind the same
// board/user authority that guards run detail.
const isOperatorViewer = req.actor.type !== "agent";
const feedbackDeliveryByComment = await feedbackDeliveryStatesSvc.listForIssue({
companyId: issue.companyId,
issueId: issue.id,
viewerCanRetry: isOperatorViewer,
viewerCanReadRuns: isOperatorViewer,
});
const withFeedbackDelivery = feedbackDeliveryByComment.size === 0
? comments
: comments.map((comment) => {
const feedbackDelivery = feedbackDeliveryByComment.get(comment.id) ?? null;
return feedbackDelivery ? { ...comment, feedbackDelivery } : comment;
});
res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, withFeedbackDelivery));
});
/**
* Explicit operator "Retry delivery" for a human comment whose automatic
* replay budget is exhausted. Board-only: this is the operator escape hatch
* from the exhausted-needs-attention state, not an agent-callable API.
*
* The recovery action stays open. It is discharged by the normal
* revalidation path once the replay actually produces a disposition, which is
* what turns the banner into the recovered notice — a retry request alone must
* not clear the attention item.
*/
router.post("/issues/:id/feedback-delivery/retry", async (req, res) => {
assertBoard(req);
const id = req.params.id as string;
const issue = await getAccessibleResource(req, res, svc.getById(id), "Issue not found");
if (!issue) return;
const actor = getActorInfo(req);
const readStates = () =>
feedbackDeliveryStatesSvc.listForIssue({
companyId: issue.companyId,
issueId: issue.id,
viewerCanRetry: true,
viewerCanReadRuns: true,
});
const activeAction = await recoveryActionsSvc.getActiveForIssue(issue.companyId, issue.id);
const isFeedbackDeliveryAction =
activeAction?.kind === FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND
&& activeAction.cause === FEEDBACK_DELIVERY_RECOVERY_CAUSE;
if (!activeAction || !isFeedbackDeliveryAction) {
const states = await readStates();
res.json({
outcome: "not_exhausted",
message: "This task has no exhausted feedback delivery to retry.",
feedbackDelivery: states.values().next().value ?? null,
} satisfies IssueFeedbackDeliveryRetryResponse);
return;
}
const evidence = (typeof activeAction.evidence === "object" && activeAction.evidence !== null
? activeAction.evidence
: {}) as Record<string, unknown>;
const rootWakeupRequestId = typeof evidence.rootWakeupRequestId === "string"
? evidence.rootWakeupRequestId
: null;
const outstandingCommentIds = Array.isArray(evidence.outstandingCommentIds)
? evidence.outstandingCommentIds.filter((entry): entry is string => typeof entry === "string")
: [];
// The original assignee owns the feedback: hand the replay back to it rather
// than to whichever manager the recovery ladder parked the action on.
const targetAgentId = activeAction.returnOwnerAgentId
?? activeAction.previousOwnerAgentId
?? issue.assigneeAgentId;
if (!targetAgentId || !rootWakeupRequestId) {
res.json({
outcome: "no_invokable_assignee",
message: "This delivery has no agent to retry against. Reassign the task first.",
feedbackDelivery: (await readStates()).get(outstandingCommentIds[0] ?? "") ?? null,
} satisfies IssueFeedbackDeliveryRetryResponse);
return;
}
const statesBefore = await readStates();
const alreadyInFlight = outstandingCommentIds.some(
(commentId) => statesBefore.get(commentId)?.retryInFlight === true,
);
if (alreadyInFlight) {
res.json({
outcome: "already_queued",
message: "A delivery retry is already queued.",
feedbackDelivery: statesBefore.get(outstandingCommentIds[0] ?? "") ?? null,
} satisfies IssueFeedbackDeliveryRetryResponse);
return;
}
const latestCommentId = outstandingCommentIds.at(-1) ?? null;
const deliveryContext = {
issueId: issue.id,
taskId: issue.id,
feedbackDeliveryRootWakeupRequestId: rootWakeupRequestId,
feedbackDeliveryRetryGeneration: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
feedbackDeliveryOperatorRequested: true,
[FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY]: outstandingCommentIds,
...(latestCommentId ? { commentId: latestCommentId, wakeCommentId: latestCommentId } : {}),
};
await heartbeat.wakeup(targetAgentId, {
source: "on_demand",
triggerDetail: "manual",
reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
payload: deliveryContext,
idempotencyKey: buildFeedbackDeliveryManualRetryIdempotencyKey({
fingerprint: activeAction.fingerprint,
attempt: (activeAction.attemptCount ?? 0) + 1,
}),
requestedByActorType: actor.actorType === "user" ? "user" : "system",
requestedByActorId: actor.actorId,
contextSnapshot: {
...deliveryContext,
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
source: "issue.operator_feedback_delivery_retry",
instruction:
"An operator asked Paperclip to deliver human feedback on this task again. Handle the outstanding "
+ "comment(s) now and record a valid disposition.",
},
});
await logActivity(db, {
companyId: issue.companyId,
actorType: actor.actorType,
actorId: actor.actorId,
agentId: targetAgentId,
action: "issue.feedback_delivery_retry_requested",
entityType: "issue",
entityId: issue.id,
details: {
recoveryActionId: activeAction.id,
rootWakeupRequestId,
sourceCommentIds: outstandingCommentIds,
attempt: (activeAction.attemptCount ?? 0) + 1,
},
});
const statesAfter = await readStates();
const projected = latestCommentId ? statesAfter.get(latestCommentId) ?? null : null;
res.json({
outcome: "queued",
message: "Retrying feedback delivery.",
// Deferred behind a live run still reads as in-flight to the operator, so
// fall back to a synthetic in-flight projection rather than reporting a
// state that would re-enable the button it just disabled.
feedbackDelivery: projected
? { ...projected, retryInFlight: true, canRetry: false }
: null,
} satisfies IssueFeedbackDeliveryRetryResponse);
});
router.get("/issues/:id/interactions", async (req, res) => {
@@ -0,0 +1,228 @@
import { and, eq, gte, inArray, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { agentWakeupRequests, heartbeatRuns, issueRecoveryActions } from "@paperclipai/db";
import type { IssueCommentFeedbackDelivery } from "@paperclipai/shared";
import {
buildFeedbackDeliveryFingerprint,
FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND,
FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
readFeedbackDeliveryWakeCommentIds,
} from "./recovery/feedback-delivery.js";
import {
FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS,
projectFeedbackDeliveryByComment,
type FeedbackDeliveryPendingReplay,
} from "./recovery/feedback-delivery-projection.js";
/** Run statuses that still represent a delivery attempt the operator is waiting on. */
const PENDING_REPLAY_RUN_STATUSES = ["queued", "scheduled_retry", "running"] as const;
/**
* Wake statuses that still represent an undelivered replay. A replay deferred
* behind a live run on the same issue is still something the operator is
* waiting on, so it reads as `retrying` just like a queued one.
*/
const PENDING_REPLAY_WAKE_STATUSES = ["queued", "deferred_issue_execution"] as const;
function readRecord(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null ? (value as Record<string, unknown>) : {};
}
function readNonEmptyString(value: unknown) {
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
}
function readGeneration(value: unknown) {
return typeof value === "number" && Number.isFinite(value) ? Math.max(0, Math.floor(value)) : 0;
}
/**
* The two replay lanes spell the wake payload slightly differently — the
* immediate lane writes `rootWakeupRequestId`/`commentIds`, the backstop lane
* writes the run-context spelling. Read both so neither lane goes invisible.
*/
function readReplayRootWakeupRequestId(payload: Record<string, unknown>) {
return (
readNonEmptyString(payload.feedbackDeliveryRootWakeupRequestId)
?? readNonEmptyString(payload.rootWakeupRequestId)
);
}
function readReplayCommentIds(payload: Record<string, unknown>) {
const merged = [...readFeedbackDeliveryWakeCommentIds(payload)];
const legacy = payload.commentIds;
if (Array.isArray(legacy)) {
for (const entry of legacy) {
const normalized = readNonEmptyString(entry);
if (normalized && !merged.includes(normalized)) merged.push(normalized);
}
}
return merged;
}
/**
* Reads the durable feedback-delivery signals for one issue and projects them
* onto the individual human comments they belong to.
*
* Three cheap, index-backed reads, all scoped to the issue:
* - `feedback_delivery` recovery actions (active plus recently resolved),
* - queued replay wakes (`payload->>'issueId'` index),
* - non-terminal runs carrying replay context (`contextSnapshot->>'issueId'`).
*
* The projection itself is pure and unit-tested separately.
*/
export function feedbackDeliveryCommentStateService(db: Db) {
async function loadRecoveryActions(input: {
companyId: string;
issueId: string;
resolvedSince: Date;
}) {
return db
.select({
id: issueRecoveryActions.id,
kind: issueRecoveryActions.kind,
cause: issueRecoveryActions.cause,
status: issueRecoveryActions.status,
fingerprint: issueRecoveryActions.fingerprint,
evidence: issueRecoveryActions.evidence,
attemptCount: issueRecoveryActions.attemptCount,
maxAttempts: issueRecoveryActions.maxAttempts,
lastAttemptAt: issueRecoveryActions.lastAttemptAt,
resolvedAt: issueRecoveryActions.resolvedAt,
createdAt: issueRecoveryActions.createdAt,
})
.from(issueRecoveryActions)
.where(
and(
eq(issueRecoveryActions.companyId, input.companyId),
eq(issueRecoveryActions.sourceIssueId, input.issueId),
eq(issueRecoveryActions.kind, FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND),
or(
inArray(issueRecoveryActions.status, ["active", "escalated"]),
gte(issueRecoveryActions.resolvedAt, input.resolvedSince),
),
),
);
}
async function loadPendingReplays(input: {
companyId: string;
issueId: string;
}): Promise<FeedbackDeliveryPendingReplay[]> {
const [wakes, runs] = await Promise.all([
db
.select({
agentId: agentWakeupRequests.agentId,
payload: agentWakeupRequests.payload,
runId: agentWakeupRequests.runId,
requestedAt: agentWakeupRequests.requestedAt,
})
.from(agentWakeupRequests)
.where(
and(
eq(agentWakeupRequests.companyId, input.companyId),
inArray(agentWakeupRequests.status, [...PENDING_REPLAY_WAKE_STATUSES]),
eq(agentWakeupRequests.reason, FEEDBACK_DELIVERY_RETRY_WAKE_REASON),
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${input.issueId}`,
),
),
db
.select({
id: heartbeatRuns.id,
agentId: heartbeatRuns.agentId,
contextSnapshot: heartbeatRuns.contextSnapshot,
wakeupRequestId: heartbeatRuns.wakeupRequestId,
scheduledRetryAt: heartbeatRuns.scheduledRetryAt,
createdAt: heartbeatRuns.createdAt,
})
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, input.companyId),
inArray(heartbeatRuns.status, [...PENDING_REPLAY_RUN_STATUSES]),
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${input.issueId}`,
sql`${heartbeatRuns.contextSnapshot} ->> 'feedbackDeliveryRootWakeupRequestId' is not null`,
),
),
]);
const replays: FeedbackDeliveryPendingReplay[] = [];
for (const wake of wakes) {
const payload = readRecord(wake.payload);
const rootWakeupRequestId = readReplayRootWakeupRequestId(payload);
if (!rootWakeupRequestId) continue;
replays.push({
fingerprint: buildFeedbackDeliveryFingerprint({
companyId: input.companyId,
issueId: input.issueId,
agentId: wake.agentId,
rootWakeupRequestId,
}),
commentIds: readReplayCommentIds(payload),
generation: readGeneration(payload.feedbackDeliveryRetryGeneration),
runId: wake.runId ?? null,
nextAttemptAt: null,
requestedAt: wake.requestedAt,
});
}
for (const run of runs) {
const context = readRecord(run.contextSnapshot);
const rootWakeupRequestId = readReplayRootWakeupRequestId(context) ?? run.wakeupRequestId;
if (!rootWakeupRequestId) continue;
replays.push({
fingerprint: buildFeedbackDeliveryFingerprint({
companyId: input.companyId,
issueId: input.issueId,
agentId: run.agentId,
rootWakeupRequestId,
}),
commentIds: readReplayCommentIds(context),
generation: readGeneration(context.feedbackDeliveryRetryGeneration),
runId: run.id,
nextAttemptAt: run.scheduledRetryAt,
requestedAt: run.createdAt,
});
}
return replays;
}
return {
/**
* Returns a comment-id keyed map of delivery state, or an empty map when
* this issue has no feedback-delivery history at all (the common case).
*/
async listForIssue(input: {
companyId: string;
issueId: string;
viewerCanRetry: boolean;
viewerCanReadRuns: boolean;
now?: Date;
}): Promise<Map<string, IssueCommentFeedbackDelivery>> {
const now = input.now ?? new Date();
// Issued together: the first automatic replay is `retrying` *before* any
// recovery action exists, so recovery actions cannot gate the replay read.
// All three predicates are index-backed and scoped to one issue.
const [recoveryActions, pendingReplays] = await Promise.all([
loadRecoveryActions({
companyId: input.companyId,
issueId: input.issueId,
resolvedSince: new Date(now.getTime() - FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS),
}),
loadPendingReplays({ companyId: input.companyId, issueId: input.issueId }),
]);
if (recoveryActions.length === 0 && pendingReplays.length === 0) return new Map();
return projectFeedbackDeliveryByComment({
recoveryActions,
pendingReplays,
viewerCanRetry: input.viewerCanRetry,
viewerCanReadRuns: input.viewerCanReadRuns,
now,
});
},
};
}
+756 -30
View File
@@ -151,6 +151,7 @@ import {
WORKTREE_INSTANCE_ROOT_METADATA_KEY,
} from "./workspace-instance-cleanup.js";
import { issueService } from "./issues.js";
import { issueRecoveryActionService } from "./issue-recovery-actions.js";
import { projectService } from "./projects.js";
import { authorizationService, type AuthorizationActor } from "./authorization.js";
import { createToolGatewayService } from "./tool-gateway.js";
@@ -223,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 {
@@ -485,6 +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 { 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
@@ -6739,6 +6751,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
};
const budgets = budgetService(db, budgetHooks);
const recovery = recoveryService(db, { enqueueWakeup });
const recoveryActions = issueRecoveryActionService(db);
function isPlanApprovalConfirmationPayload(payload: unknown) {
const target = parseObject(parseObject(payload).target);
@@ -10065,7 +10078,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return { outcome: "satisfied" as const, queuedRun: null };
}
if (readNonEmptyString(contextSnapshot.retryReason) === "missing_issue_comment") {
const issueCommentRetryReason = readNonEmptyString(contextSnapshot.retryReason);
if (
issueCommentRetryReason === "missing_issue_comment" ||
issueCommentRetryReason === FEEDBACK_DELIVERY_RETRY_REASON
) {
await patchRunIssueCommentStatus(run.id, {
issueCommentStatus: "retry_exhausted",
issueCommentSatisfiedByCommentId: null,
@@ -10074,7 +10091,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
eventType: "lifecycle",
stream: "system",
level: "warn",
message: "Run ended without an issue comment after one retry; no further comment wake will be queued",
message:
issueCommentRetryReason === FEEDBACK_DELIVERY_RETRY_REASON
? "Recovered feedback run ended without handling evidence; no separate missing-comment retry will be queued"
: "Run ended without an issue comment after one retry; no further comment wake will be queued",
});
return { outcome: "retry_exhausted" as const, queuedRun: null };
}
@@ -10123,6 +10143,663 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return { outcome: "retry_exhausted" as const, queuedRun: null };
}
type FeedbackDeliveryContext = {
issueId: string;
rootWakeupRequestId: string;
commentIds: string[];
generation: number;
fingerprint: string;
};
async function readFeedbackDeliveryContext(
run: typeof heartbeatRuns.$inferSelect,
): Promise<FeedbackDeliveryContext | null> {
const context = parseObject(run.contextSnapshot);
const issueId = readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId);
const rootWakeupRequestId =
readNonEmptyString(context.feedbackDeliveryRootWakeupRequestId) ?? run.wakeupRequestId;
if (!issueId || !rootWakeupRequestId) return null;
const wakeReason = readNonEmptyString(context.wakeReason);
const commentIds = mergeWakeCommentIds(context, deriveCommentId(context, null));
const carriesExplicitResume = carriesExplicitFeedbackResume(context);
const eligibleWake = isEligibleFeedbackDeliveryWake({ wakeReason, carriesExplicitResume });
if (!eligibleWake || (commentIds.length === 0 && !carriesExplicitResume)) return null;
const rootWake = await db
.select({ requestedByActorType: agentWakeupRequests.requestedByActorType })
.from(agentWakeupRequests)
.where(and(
eq(agentWakeupRequests.id, rootWakeupRequestId),
eq(agentWakeupRequests.companyId, run.companyId),
eq(agentWakeupRequests.agentId, run.agentId),
))
.then((rows) => rows[0] ?? null);
// Durable feedback delivery is a human-input contract. System-created
// generation-1 wakes retain their original user wake id in context.
if (rootWake?.requestedByActorType !== "user") return null;
return {
issueId,
rootWakeupRequestId,
commentIds,
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,
agentId: run.agentId,
rootWakeupRequestId,
}),
};
}
async function findFeedbackHandlingEvidence(
run: typeof heartbeatRuns.$inferSelect,
issueId: string,
) {
const [comment, documentRevision, workProduct, issueActivity] = await Promise.all([
findRunIssueComment(run.id, run.companyId, issueId),
db
.select({ id: documentRevisions.id })
.from(documentRevisions)
.innerJoin(issueDocuments, eq(issueDocuments.documentId, documentRevisions.documentId))
.where(and(
eq(documentRevisions.companyId, run.companyId),
eq(documentRevisions.createdByRunId, run.id),
eq(issueDocuments.companyId, run.companyId),
eq(issueDocuments.issueId, issueId),
))
.limit(1)
.then((rows) => rows[0] ?? null),
db
.select({ id: issueWorkProducts.id })
.from(issueWorkProducts)
.where(and(
eq(issueWorkProducts.companyId, run.companyId),
eq(issueWorkProducts.issueId, issueId),
eq(issueWorkProducts.createdByRunId, run.id),
))
.limit(1)
.then((rows) => rows[0] ?? null),
db
.select({ id: activityLog.id, action: activityLog.action })
.from(activityLog)
.where(and(
eq(activityLog.companyId, run.companyId),
eq(activityLog.entityType, "issue"),
eq(activityLog.entityId, issueId),
eq(activityLog.runId, run.id),
inArray(activityLog.action, ISSUE_PROGRESS_ACTIVITY_ACTIONS),
))
.limit(1)
.then((rows) => rows[0] ?? null),
]);
if (comment) return { kind: "comment", id: comment.id } as const;
if (documentRevision) return { kind: "document_revision", id: documentRevision.id } as const;
if (workProduct) return { kind: "work_product", id: workProduct.id } as const;
if (issueActivity) return { kind: "issue_activity", id: issueActivity.id, action: issueActivity.action } as const;
return null;
}
async function logFeedbackRetryQueuedOnce(input: {
run: typeof heartbeatRuns.$inferSelect;
delivery: FeedbackDeliveryContext;
retryRunId: string | null;
retryWakeupRequestId: string | null;
disposition: "queued" | "coalesced" | "deferred";
}) {
const existing = await db
.select({ id: activityLog.id })
.from(activityLog)
.where(and(
eq(activityLog.companyId, input.run.companyId),
eq(activityLog.entityType, "issue"),
eq(activityLog.entityId, input.delivery.issueId),
eq(activityLog.action, "issue.feedback_retry_queued"),
sql`${activityLog.details} ->> 'rootWakeupRequestId' = ${input.delivery.rootWakeupRequestId}`,
))
.limit(1)
.then((rows) => rows[0] ?? null);
if (existing) return;
await logActivity(db, {
companyId: input.run.companyId,
actorType: "system",
actorId: "heartbeat",
agentId: input.run.agentId,
runId: input.run.id,
action: "issue.feedback_retry_queued",
entityType: "issue",
entityId: input.delivery.issueId,
details: {
sourceRunId: input.run.id,
sourceWakeupRequestId: input.run.wakeupRequestId,
rootWakeupRequestId: input.delivery.rootWakeupRequestId,
sourceCommentIds: input.delivery.commentIds,
retryRunId: input.retryRunId,
retryWakeupRequestId: input.retryWakeupRequestId,
retryGeneration: 1,
disposition: input.disposition,
},
});
}
async function ensureFeedbackDeliveryRecovery(input: {
run: typeof heartbeatRuns.$inferSelect;
agent: typeof agents.$inferSelect;
delivery: FeedbackDeliveryContext;
gateReason?: string | null;
gateErrorCode?: string | null;
}) {
const issue = await db
.select({
id: issues.id,
status: issues.status,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
.from(issues)
.where(and(eq(issues.id, input.delivery.issueId), eq(issues.companyId, input.run.companyId)))
.then((rows) => rows[0] ?? null);
if (!issue) return { kind: "suppressed" as const, reason: "issue_not_found" };
if (issue.status === "done" || issue.status === "cancelled") {
return { kind: "suppressed" as const, reason: "issue_terminal" };
}
if (issue.assigneeAgentId !== input.run.agentId || issue.assigneeUserId) {
return { kind: "suppressed" as const, reason: "ownership_changed" };
}
const existing = await recoveryActions.getActiveForIssue(input.run.companyId, issue.id);
if (
existing?.kind === FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND
&& existing.cause === FEEDBACK_DELIVERY_RECOVERY_CAUSE
&& existing.fingerprint === input.delivery.fingerprint
) {
return { kind: "recovery" as const, action: existing, reusedExisting: true };
}
const action = await recoveryActions.upsertSourceScoped({
companyId: input.run.companyId,
sourceIssueId: issue.id,
kind: FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND,
ownerType: "agent",
ownerAgentId: input.agent.id,
previousOwnerAgentId: input.agent.id,
returnOwnerAgentId: input.agent.id,
cause: FEEDBACK_DELIVERY_RECOVERY_CAUSE,
fingerprint: input.delivery.fingerprint,
evidence: {
sourceRunId: input.run.id,
sourceWakeupRequestId: input.run.wakeupRequestId,
rootWakeupRequestId: input.delivery.rootWakeupRequestId,
outstandingCommentIds: input.delivery.commentIds,
retryGeneration: input.delivery.generation,
lastFailureCode: input.run.errorCode,
lastFailureReason: input.run.error,
gateErrorCode: input.gateErrorCode ?? null,
gateReason: input.gateReason ?? null,
},
nextAction: [
input.gateReason,
"Restore an invokable current assignee and valid workspace, then retry delivery explicitly or record a valid manual issue disposition.",
].filter(Boolean).join(" "),
wakePolicy: {
type: "manual_repair_required",
reason: FEEDBACK_DELIVERY_RECOVERY_CAUSE,
ownerAgentId: input.agent.id,
},
maxAttempts: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
lastAttemptAt: new Date(),
});
await logActivity(db, {
companyId: input.run.companyId,
actorType: "system",
actorId: "heartbeat",
agentId: input.run.agentId,
runId: input.run.id,
action: "issue.feedback_recovery_exhausted",
entityType: "issue",
entityId: issue.id,
details: {
recoveryActionId: action.id,
sourceRunId: input.run.id,
sourceWakeupRequestId: input.run.wakeupRequestId,
rootWakeupRequestId: input.delivery.rootWakeupRequestId,
sourceCommentIds: input.delivery.commentIds,
retryGeneration: input.delivery.generation,
errorCode: input.run.errorCode,
gateErrorCode: input.gateErrorCode ?? null,
},
});
return { kind: "recovery" as const, action, reusedExisting: false };
}
async function enqueueFeedbackDeliveryRetry(input: {
run: typeof heartbeatRuns.$inferSelect;
agent: typeof agents.$inferSelect;
delivery: FeedbackDeliveryContext;
}) {
const now = new Date();
const sourceContext = parseObject(input.run.contextSnapshot);
const retryContext = withRecoveryModelProfileHint({
...sourceContext,
issueId: input.delivery.issueId,
taskId: input.delivery.issueId,
wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
retryReason: FEEDBACK_DELIVERY_RETRY_REASON,
retryOfRunId: input.run.id,
feedbackDeliveryRootWakeupRequestId: input.delivery.rootWakeupRequestId,
feedbackDeliverySourceRunId: input.run.id,
feedbackDeliveryRetryGeneration: 1,
[WAKE_COMMENT_IDS_KEY]: input.delivery.commentIds,
...(input.delivery.commentIds.length > 0
? {
commentId: input.delivery.commentIds.at(-1),
wakeCommentId: input.delivery.commentIds.at(-1),
}
: {}),
}, "normal_model");
const taskKey = deriveTaskKeyWithHeartbeatFallback(retryContext, null);
const sessionBefore = await resolveSessionBeforeForWakeup(input.agent, taskKey);
const responsibleUserId = await resolveResponsibleUserIdForRunContext(input.run, retryContext);
const idempotencyKey = buildFeedbackDeliveryRetryIdempotencyKey({
fingerprint: input.delivery.fingerprint,
generation: 1,
});
const result = await db.transaction(async (tx) => {
const companyIsActive = await tx
.select({ status: companies.status })
.from(companies)
.where(eq(companies.id, input.run.companyId))
.then((rows) => rows[0]?.status === "active");
if (!companyIsActive) return { kind: "suppressed" as const };
await tx.execute(sql`
select id from issues
where id = ${input.delivery.issueId} and company_id = ${input.run.companyId}
for update
`);
const issue = await tx
.select({
id: issues.id,
status: issues.status,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
executionRunId: issues.executionRunId,
})
.from(issues)
.where(and(
eq(issues.id, input.delivery.issueId),
eq(issues.companyId, input.run.companyId),
))
.then((rows) => rows[0] ?? null);
if (
!issue ||
issue.assigneeAgentId !== input.run.agentId ||
issue.assigneeUserId ||
issue.status === "done" ||
issue.status === "cancelled"
) {
return { kind: "suppressed" as const };
}
const liveRun = await tx
.select()
.from(heartbeatRuns)
.where(and(
eq(heartbeatRuns.companyId, input.run.companyId),
inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]),
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${input.delivery.issueId}`,
ne(heartbeatRuns.id, input.run.id),
))
.orderBy(
sql`case when ${heartbeatRuns.id} = ${issue.executionRunId} then 0 when ${heartbeatRuns.status} = 'running' then 1 else 2 end`,
desc(heartbeatRuns.createdAt),
desc(heartbeatRuns.id),
)
.limit(1)
.then((rows) => rows[0] ?? null);
const orderCommentIds = async (ids: string[]) => {
if (ids.length === 0) return ids;
const rows = await tx
.select({ id: issueComments.id })
.from(issueComments)
.where(and(
eq(issueComments.companyId, input.run.companyId),
eq(issueComments.issueId, input.delivery.issueId),
inArray(issueComments.id, ids),
))
.orderBy(asc(issueComments.createdAt), asc(issueComments.id));
const ordered = rows.map((row) => row.id);
return [...ordered, ...ids.filter((id) => !ordered.includes(id))];
};
const liveContext = parseObject(liveRun?.contextSnapshot);
if (
liveRun &&
readNonEmptyString(liveContext.feedbackDeliveryRootWakeupRequestId) ===
input.delivery.rootWakeupRequestId &&
Number(liveContext.feedbackDeliveryRetryGeneration) >= 1
) {
return {
kind: "coalesced" as const,
run: liveRun,
wakeupRequestId: liveRun.wakeupRequestId,
};
}
if (
liveRun &&
liveRun.agentId === input.run.agentId &&
liveRun.status !== "running"
) {
const mergedContext = mergeCoalescedContextSnapshot(retryContext, parseObject(liveRun.contextSnapshot));
const orderedCommentIds = await orderCommentIds(mergeWakeCommentIds(retryContext, liveRun.contextSnapshot));
if (orderedCommentIds.length > 0) {
mergedContext[WAKE_COMMENT_IDS_KEY] = orderedCommentIds;
mergedContext.commentId = orderedCommentIds.at(-1);
mergedContext.wakeCommentId = orderedCommentIds.at(-1);
}
mergedContext.feedbackDeliveryRootWakeupRequestId = input.delivery.rootWakeupRequestId;
mergedContext.feedbackDeliverySourceRunId = input.run.id;
mergedContext.feedbackDeliveryRetryGeneration = 1;
const mergedRun = await tx
.update(heartbeatRuns)
.set({ contextSnapshot: mergedContext, updatedAt: now })
.where(eq(heartbeatRuns.id, liveRun.id))
.returning()
.then((rows) => rows[0] ?? liveRun);
return {
kind: "coalesced" as const,
run: mergedRun,
wakeupRequestId: mergedRun.wakeupRequestId,
};
}
const existingDeferred = await tx
.select()
.from(agentWakeupRequests)
.where(and(
eq(agentWakeupRequests.companyId, input.run.companyId),
eq(agentWakeupRequests.agentId, input.run.agentId),
eq(agentWakeupRequests.status, "deferred_issue_execution"),
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${input.delivery.issueId}`,
))
.orderBy(asc(agentWakeupRequests.requestedAt), asc(agentWakeupRequests.id))
.limit(1)
.then((rows) => rows[0] ?? null);
if (liveRun) {
const existingPayload = parseObject(existingDeferred?.payload);
const existingDeferredContext = parseObject(existingPayload[DEFERRED_WAKE_CONTEXT_KEY]);
const mergedDeferredContext = mergeCoalescedContextSnapshot(retryContext, existingDeferredContext);
const orderedCommentIds = await orderCommentIds(
mergeWakeCommentIds(retryContext, existingDeferredContext),
);
if (orderedCommentIds.length > 0) {
mergedDeferredContext[WAKE_COMMENT_IDS_KEY] = orderedCommentIds;
mergedDeferredContext.commentId = orderedCommentIds.at(-1);
mergedDeferredContext.wakeCommentId = orderedCommentIds.at(-1);
}
mergedDeferredContext.feedbackDeliveryRootWakeupRequestId = input.delivery.rootWakeupRequestId;
mergedDeferredContext.feedbackDeliverySourceRunId = input.run.id;
mergedDeferredContext.feedbackDeliveryRetryGeneration = 1;
const payload = {
...existingPayload,
issueId: input.delivery.issueId,
retryOfRunId: input.run.id,
retryReason: FEEDBACK_DELIVERY_RETRY_REASON,
[DEFERRED_WAKE_CONTEXT_KEY]: mergedDeferredContext,
};
if (existingDeferred) {
const updated = await tx
.update(agentWakeupRequests)
.set({
payload,
coalescedCount: (existingDeferred.coalescedCount ?? 0) + 1,
updatedAt: now,
})
.where(eq(agentWakeupRequests.id, existingDeferred.id))
.returning()
.then((rows) => rows[0] ?? existingDeferred);
return { kind: "deferred" as const, run: liveRun, wakeupRequestId: updated.id };
}
const deferred = await tx
.insert(agentWakeupRequests)
.values({
companyId: input.run.companyId,
agentId: input.run.agentId,
source: "automation",
triggerDetail: "system",
reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
payload,
status: "deferred_issue_execution",
requestedByActorType: "system",
requestedByActorId: null,
idempotencyKey,
updatedAt: now,
})
.returning()
.then((rows) => rows[0]);
return { kind: "deferred" as const, run: liveRun, wakeupRequestId: deferred.id };
}
const wakeupRequest = await tx
.insert(agentWakeupRequests)
.values({
companyId: input.run.companyId,
agentId: input.run.agentId,
source: "automation",
triggerDetail: "system",
reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
payload: {
issueId: input.delivery.issueId,
retryOfRunId: input.run.id,
retryReason: FEEDBACK_DELIVERY_RETRY_REASON,
rootWakeupRequestId: input.delivery.rootWakeupRequestId,
commentIds: input.delivery.commentIds,
},
status: "queued",
requestedByActorType: "system",
requestedByActorId: null,
idempotencyKey,
updatedAt: now,
})
.returning()
.then((rows) => rows[0]);
const retryRun = await tx
.insert(heartbeatRuns)
.values({
companyId: input.run.companyId,
agentId: input.run.agentId,
invocationSource: "automation",
triggerDetail: "system",
status: "queued",
wakeupRequestId: wakeupRequest.id,
contextSnapshot: retryContext,
responsibleUserId,
sessionIdBefore: sessionBefore,
retryOfRunId: input.run.id,
scheduledRetryAttempt: 1,
scheduledRetryReason: FEEDBACK_DELIVERY_RETRY_REASON,
updatedAt: now,
})
.returning()
.then((rows) => rows[0]);
await tx
.update(agentWakeupRequests)
.set({ runId: retryRun.id, updatedAt: now })
.where(eq(agentWakeupRequests.id, wakeupRequest.id));
await tx
.update(issues)
.set({
executionRunId: retryRun.id,
executionAgentNameKey: normalizeAgentNameKey(input.agent.name),
executionLockedAt: now,
updatedAt: now,
})
.where(and(eq(issues.id, issue.id), eq(issues.companyId, input.run.companyId)));
return { kind: "queued" as const, run: retryRun, wakeupRequestId: wakeupRequest.id };
});
if (result.kind === "queued") {
publishLiveEvent({
companyId: result.run.companyId,
type: "heartbeat.run.queued",
payload: {
runId: result.run.id,
agentId: result.run.agentId,
invocationSource: result.run.invocationSource,
triggerDetail: result.run.triggerDetail,
wakeupRequestId: result.run.wakeupRequestId,
},
});
await startNextQueuedRunForAgent(result.run.agentId);
}
return result;
}
async function handleFeedbackDeliveryDisposition(input: {
run: typeof heartbeatRuns.$inferSelect;
agent: typeof agents.$inferSelect;
existingRetryRun?: typeof heartbeatRuns.$inferSelect | null;
}) {
const delivery = await readFeedbackDeliveryContext(input.run);
if (!delivery) return { kind: "not_applicable" as const };
const handlingEvidence = await findFeedbackHandlingEvidence(input.run, delivery.issueId);
if (handlingEvidence) {
return { kind: "handled" as const, delivery, handlingEvidence };
}
const preLaunchProcessLoss =
input.run.errorCode === "process_lost" &&
!input.run.processPid &&
!input.run.processGroupId;
const automaticReplayExhausted = delivery.generation >= FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES;
if (automaticReplayExhausted) {
const recovery = await ensureFeedbackDeliveryRecovery({
run: input.run,
agent: input.agent,
delivery,
});
return { kind: "recovery" as const, delivery, recovery };
}
if (!preLaunchProcessLoss && !input.existingRetryRun) {
return { kind: "not_applicable" as const };
}
const gate = await evaluateScheduledRetryGate({
run: input.run,
agent: input.agent,
contextSnapshot: parseObject(input.run.contextSnapshot),
retryReason: FEEDBACK_DELIVERY_RETRY_REASON,
});
if (!gate.allowed) {
if (
gate.errorCode === "issue_not_found" ||
gate.errorCode === "issue_reassigned" ||
gate.errorCode === "issue_cancelled" ||
gate.errorCode === "issue_terminal_status"
) {
return { kind: "suppressed" as const, delivery, reason: gate.errorCode };
}
const recovery = await ensureFeedbackDeliveryRecovery({
run: input.run,
agent: input.agent,
delivery,
gateReason: gate.reason,
gateErrorCode: gate.errorCode,
});
return { kind: "recovery" as const, delivery, recovery };
}
if (input.existingRetryRun) {
const retryContext: Record<string, unknown> = {
...parseObject(input.existingRetryRun.contextSnapshot),
feedbackDeliveryRootWakeupRequestId: delivery.rootWakeupRequestId,
feedbackDeliverySourceRunId: input.run.id,
feedbackDeliveryRetryGeneration: 1,
[WAKE_COMMENT_IDS_KEY]: delivery.commentIds,
};
if (delivery.commentIds.length > 0) {
retryContext.commentId = delivery.commentIds.at(-1);
retryContext.wakeCommentId = delivery.commentIds.at(-1);
}
const updatedRetry = await db
.update(heartbeatRuns)
.set({ contextSnapshot: retryContext, updatedAt: new Date() })
.where(eq(heartbeatRuns.id, input.existingRetryRun.id))
.returning()
.then((rows) => rows[0] ?? input.existingRetryRun!);
await logFeedbackRetryQueuedOnce({
run: input.run,
delivery,
retryRunId: updatedRetry.id,
retryWakeupRequestId: updatedRetry.wakeupRequestId,
disposition: "queued",
});
return { kind: "retry" as const, delivery, run: updatedRetry, disposition: "queued" as const };
}
const retry = await enqueueFeedbackDeliveryRetry({ run: input.run, agent: input.agent, delivery });
if (retry.kind === "suppressed") {
return { kind: "suppressed" as const, delivery, reason: "state_changed" };
}
await logFeedbackRetryQueuedOnce({
run: input.run,
delivery,
retryRunId: retry.run.id,
retryWakeupRequestId: retry.wakeupRequestId,
disposition: retry.kind,
});
return { kind: "retry" as const, delivery, run: retry.run, disposition: retry.kind };
}
async function applyPostRunDisposition(input: {
run: typeof heartbeatRuns.$inferSelect;
agent: typeof agents.$inferSelect;
existingRetryRun?: typeof heartbeatRuns.$inferSelect | null;
scheduleInteractionRetry?: boolean;
suppressImmediateRecovery?: boolean;
}) {
const feedback = await handleFeedbackDeliveryDisposition({
run: input.run,
agent: input.agent,
existingRetryRun: input.existingRetryRun,
});
let interactionRetry = null;
if (
input.scheduleInteractionRetry !== false &&
feedback.kind === "not_applicable" &&
UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES.includes(
input.run.status as (typeof UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES)[number],
)
) {
interactionRetry = await scheduleInteractionContinuationInfrastructureRetryIfEligible(
input.run,
input.agent,
);
}
await releaseIssueExecutionAndPromote(input.run, {
suppressImmediateRecovery:
input.suppressImmediateRecovery === true ||
feedback.kind === "retry" ||
feedback.kind === "recovery" ||
interactionRetry?.outcome === "scheduled",
});
await handleIssueReviewPathDisposition(input.run);
return { feedback, interactionRetry };
}
async function enqueueProcessLossRetry(
run: typeof heartbeatRuns.$inferSelect,
agent: typeof agents.$inferSelect,
@@ -10454,6 +11131,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.select({
run: heartbeatRuns,
adapterType: agents.adapterType,
agent: agents,
})
.from(heartbeatRuns)
.innerJoin(agents, eq(heartbeatRuns.agentId, agents.id))
@@ -10611,6 +11289,24 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
classify(candidate, "adopted", processPidAlive ? "process_pid_alive" : "process_group_alive", patch);
}
// A run can finalize while the server is down (for example through an
// external callback). Re-run the same idempotent disposition pipeline used
// by live execution and startup reaping so restart reconciliation cannot
// strand feedback or a consumed review path.
for (const runId of new Set(finalizedWhileDownRunIds)) {
const current = currentByRunId.get(runId);
if (!current || !isHeartbeatRunTerminalStatus(current.run.status)) continue;
await applyPostRunDisposition({
run: current.run,
agent: current.agent,
}).catch((error) => {
logger.error(
{ err: error, runId },
"failed to apply post-run disposition during hot-restart reconciliation",
);
});
}
const report = await writeHotRestartReport({
version: 1,
requestedAt: intent.requestedAt,
@@ -10727,11 +11423,15 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
});
const retry = await enqueueProcessLossRetry(interrupted, agent, now);
if (!retry) {
await releaseIssueExecutionAndPromote(interrupted);
} else {
if (retry) {
retryRunIds.push(retry.id);
}
await applyPostRunDisposition({
run: interrupted,
agent,
existingRetryRun: retry,
scheduleInteractionRetry: false,
});
await appendRunEvent(interrupted, await nextRunEventSeq(interrupted.id), {
eventType: "lifecycle",
@@ -13507,13 +14207,23 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
if (retryAgent) {
retriedRun = await enqueueProcessLossRetry(finalizedRun, retryAgent, now);
}
} else if (retryAgent) {
const scheduled = await scheduleInteractionContinuationInfrastructureRetryIfEligible(finalizedRun, retryAgent);
retriedRun = scheduled?.outcome === "scheduled" ? scheduled.run : null;
}
if (!retriedRun) {
if (retryAgent) {
const disposition = await applyPostRunDisposition({
run: finalizedRun,
agent: retryAgent,
existingRetryRun: retriedRun,
scheduleInteractionRetry: !shouldRetry,
});
if (!retriedRun && disposition.feedback.kind === "retry") {
retriedRun = disposition.feedback.run;
} else if (!retriedRun && disposition.interactionRetry?.outcome === "scheduled") {
retriedRun = disposition.interactionRetry.run;
}
} else {
await releaseIssueExecutionAndPromote(finalizedRun);
await handleIssueReviewPathDisposition(finalizedRun);
}
await appendRunEvent(finalizedRun, await nextRunEventSeq(finalizedRun.id), {
@@ -16253,9 +16963,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
await scheduleBoundedRetryForRun(livenessRun, agent);
}
const issueCommentPolicyResult = await finalizeIssueCommentPolicy(livenessRun, agent);
await releaseIssueExecutionAndPromote(livenessRun);
await applyPostRunDisposition({ run: livenessRun, agent });
await handleRunLivenessContinuation(livenessRun);
await handleIssueReviewPathDisposition(livenessRun);
await handleSuccessfulRunHandoff(
issueCommentPolicyResult.outcome === "retry_queued" || issueCommentPolicyResult.outcome === "retry_exhausted"
? {
@@ -16426,9 +17135,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
if (!isWorkspaceValidationFailedRun(livenessRun) && !isConfigurationIncompleteFailedRun(livenessRun)) {
await finalizeIssueCommentPolicy(livenessRun, agent);
}
await scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, agent);
await releaseIssueExecutionAndPromote(livenessRun);
await handleIssueReviewPathDisposition(livenessRun);
await applyPostRunDisposition({ run: livenessRun, agent });
await updateRuntimeState(agent, livenessRun, {
exitCode: null,
@@ -16554,25 +17261,27 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
if (!isWorkspaceValidationFailedRun(livenessRun) && !isConfigurationIncompleteFailedRun(livenessRun)) {
await finalizeIssueCommentPolicy(livenessRun, failedAgent).catch(() => undefined);
}
await scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, failedAgent).catch((retryError) => {
logger.warn(
{ err: retryError, runId: livenessRun.id },
"failed to schedule interaction continuation retry after setup failure",
await applyPostRunDisposition({ run: livenessRun, agent: failedAgent }).catch((dispositionError) => {
logger.error(
{ err: dispositionError, runId: livenessRun.id },
"failed to apply post-run disposition after setup failure",
);
});
}
if (!failedAgent) {
await releaseIssueExecutionAndPromote(livenessRun).catch((releaseError) => {
logger.error(
{ err: releaseError, runId },
"failed to release issue execution after heartbeat setup failure",
);
});
await handleIssueReviewPathDisposition(livenessRun).catch((reviewPathError) => {
logger.error(
{ err: reviewPathError, runId },
"failed to evaluate review-path disposition after heartbeat setup failure",
);
});
}
await releaseIssueExecutionAndPromote(livenessRun).catch((releaseError) => {
logger.error(
{ err: releaseError, runId },
"failed to release issue execution after heartbeat setup failure",
);
});
await handleIssueReviewPathDisposition(livenessRun).catch((reviewPathError) => {
logger.error(
{ err: reviewPathError, runId },
"failed to evaluate review-path disposition after heartbeat setup failure",
);
});
}
// Ensure the agent is not left stuck in "running" if the setup-failure
// path owned the terminal transition. If another path already finalized
@@ -16585,6 +17294,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
}
} finally {
let latestRun = await getRun(run.id).catch(() => null);
const dispositionNeededAfterLeaseTerminalization =
latestRun?.status === "running" || latestRun?.status === "queued";
// Close the invariant "environment lease released implies the run is
// terminal". When the teardown reaches this point with the run still
// running or queued, force a terminal status before the lease is
@@ -16598,6 +17309,21 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return latestRun;
});
}
if (
dispositionNeededAfterLeaseTerminalization
&& latestRun
&& isHeartbeatRunTerminalStatus(latestRun.status)
) {
const dispositionAgent = await getAgent(latestRun.agentId).catch(() => null);
if (dispositionAgent) {
await applyPostRunDisposition({ run: latestRun, agent: dispositionAgent }).catch((dispositionError) => {
logger.error(
{ err: dispositionError, runId: latestRun.id },
"failed to apply post-run disposition after lease-release terminalization",
);
});
}
}
await releaseEnvironmentLeasesForRun({
runId: run.id,
companyId: run.companyId,
+1
View File
@@ -49,6 +49,7 @@ export { issueTreeControlService } from "./issue-tree-control.js";
export { issueApprovalService } from "./issue-approvals.js";
export { issueReferenceService } from "./issue-references.js";
export { issueRecoveryActionService } from "./issue-recovery-actions.js";
export { feedbackDeliveryCommentStateService } from "./feedback-delivery-comment-states.js";
export {
stalledReviewDecisionService,
type DecideStalledReviewInput,
@@ -0,0 +1,293 @@
import { describe, expect, it } from "vitest";
import {
FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS,
projectFeedbackDeliveryByComment,
toSafeFeedbackDeliveryFailureReason,
type FeedbackDeliveryPendingReplay,
type FeedbackDeliveryRecoveryActionRow,
} from "./feedback-delivery-projection.js";
import { FEEDBACK_DELIVERY_RECOVERY_CAUSE } from "./feedback-delivery.js";
const NOW = new Date("2026-08-14T18:00:00.000Z");
const FINGERPRINT = "feedback_delivery:company-1:issue-1:agent-1:root-wake-1";
function action(
overrides: Partial<FeedbackDeliveryRecoveryActionRow> = {},
): FeedbackDeliveryRecoveryActionRow {
return {
id: "action-1",
kind: "feedback_delivery",
cause: FEEDBACK_DELIVERY_RECOVERY_CAUSE,
status: "active",
fingerprint: FINGERPRINT,
evidence: {
outstandingCommentIds: ["comment-1"],
rootWakeupRequestId: "root-wake-1",
retryGeneration: 1,
maxAutomaticRetries: 1,
sourceRunId: "run-1",
lastFailureCode: "process_lost",
// Deliberately present: the projection must never surface it.
lastFailureReason: "Error: ECONNREFUSED /home/agent/.secret/token.json",
},
attemptCount: 1,
maxAttempts: 1,
lastAttemptAt: new Date("2026-08-14T17:58:00.000Z"),
resolvedAt: null,
createdAt: new Date("2026-08-14T17:57:00.000Z"),
...overrides,
};
}
function replay(
overrides: Partial<FeedbackDeliveryPendingReplay> = {},
): FeedbackDeliveryPendingReplay {
return {
fingerprint: FINGERPRINT,
commentIds: ["comment-1"],
generation: 1,
runId: "run-2",
nextAttemptAt: new Date("2026-08-14T18:00:12.000Z"),
requestedAt: new Date("2026-08-14T17:59:30.000Z"),
...overrides,
};
}
function project(input: {
recoveryActions?: FeedbackDeliveryRecoveryActionRow[];
pendingReplays?: FeedbackDeliveryPendingReplay[];
viewerCanRetry?: boolean;
viewerCanReadRuns?: boolean;
now?: Date;
}) {
return projectFeedbackDeliveryByComment({
recoveryActions: input.recoveryActions ?? [],
pendingReplays: input.pendingReplays ?? [],
viewerCanRetry: input.viewerCanRetry ?? true,
viewerCanReadRuns: input.viewerCanReadRuns ?? true,
now: input.now ?? NOW,
});
}
describe("projectFeedbackDeliveryByComment", () => {
it("returns nothing for an issue with no delivery history", () => {
expect(project({}).size).toBe(0);
});
it("projects a queued replay as quiet retrying state with no operator action", () => {
const states = project({ pendingReplays: [replay()] });
const state = states.get("comment-1");
expect(state).toMatchObject({
state: "retrying",
sourceCommentId: "comment-1",
attempt: 1,
maxAttempts: 1,
canRetry: false,
retryInFlight: true,
safeFailureReason: null,
nextAttemptAt: "2026-08-14T18:00:12.000Z",
targetRunId: "run-2",
});
});
it("projects an active recovery action as exhausted and offers retry", () => {
const state = project({ recoveryActions: [action()] }).get("comment-1");
expect(state).toMatchObject({
state: "exhausted_needs_attention",
attempt: 1,
maxAttempts: 1,
canRetry: true,
retryInFlight: false,
targetRunId: "run-1",
safeFailureReason: "The agent's process stopped before it read the comment.",
batchedCommentCount: 1,
});
});
it("never leaks a raw failure reason into the projection", () => {
const state = project({
recoveryActions: [
action({
evidence: {
outstandingCommentIds: ["comment-1"],
lastFailureCode: "some_unmapped_adapter_code",
lastFailureReason: "Error: ECONNREFUSED /home/agent/.secret/token.json",
gateReason: "workspace /srv/secret/path is dirty",
},
}),
],
}).get("comment-1");
expect(state?.safeFailureReason).toBeNull();
expect(JSON.stringify(state)).not.toContain("ECONNREFUSED");
expect(JSON.stringify(state)).not.toContain("/srv/secret/path");
});
it("keeps the actionable banner while an explicit retry is in flight", () => {
const state = project({
recoveryActions: [action()],
pendingReplays: [replay()],
}).get("comment-1");
// Still exhausted: clearing it here would drop the attention item before the
// replay has actually delivered anything.
expect(state).toMatchObject({
state: "exhausted_needs_attention",
retryInFlight: true,
canRetry: false,
});
});
it("projects a recently resolved action as recovered after attention", () => {
const state = project({
recoveryActions: [
action({
status: "resolved",
resolvedAt: new Date("2026-08-14T17:59:00.000Z"),
}),
],
}).get("comment-1");
expect(state).toMatchObject({
state: "recovered_after_attention",
deliveredAt: "2026-08-14T17:59:00.000Z",
canRetry: false,
retryInFlight: false,
safeFailureReason: null,
});
});
it("drops the recovered notice once it ages out of the window", () => {
const states = project({
recoveryActions: [
action({
status: "resolved",
resolvedAt: new Date(NOW.getTime() - FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS - 1000),
}),
],
});
expect(states.size).toBe(0);
});
it("lets a re-opened obligation win over its own earlier resolution", () => {
const state = project({
recoveryActions: [
action({
id: "action-old",
status: "resolved",
resolvedAt: new Date("2026-08-14T17:30:00.000Z"),
}),
action({ id: "action-new", status: "active" }),
],
}).get("comment-1");
expect(state?.state).toBe("exhausted_needs_attention");
});
it("keeps a newer comment's obligation from clearing an older one", () => {
const states = project({
recoveryActions: [
action({
id: "action-old",
fingerprint: `${FINGERPRINT}:old`,
status: "active",
evidence: { outstandingCommentIds: ["comment-1"], retryGeneration: 1 },
}),
action({
id: "action-new",
fingerprint: `${FINGERPRINT}:new`,
status: "resolved",
resolvedAt: new Date("2026-08-14T17:59:00.000Z"),
evidence: { outstandingCommentIds: ["comment-2"], retryGeneration: 1 },
}),
],
});
expect(states.get("comment-1")?.state).toBe("exhausted_needs_attention");
expect(states.get("comment-2")?.state).toBe("recovered_after_attention");
});
it("rolls one batched obligation onto every comment it covers", () => {
const states = project({
recoveryActions: [
action({
evidence: {
outstandingCommentIds: ["comment-1", "comment-2", "comment-3"],
retryGeneration: 1,
},
}),
],
});
expect([...states.keys()]).toEqual(["comment-1", "comment-2", "comment-3"]);
for (const state of states.values()) {
expect(state.batchedCommentCount).toBe(3);
expect(state.state).toBe("exhausted_needs_attention");
}
});
it("omits the run reference for a viewer who cannot read runs", () => {
const state = project({
recoveryActions: [action()],
pendingReplays: [replay()],
viewerCanReadRuns: false,
}).get("comment-1");
expect(state?.targetRunId).toBeNull();
// The recovery action stays available even without a run link.
expect(state?.state).toBe("exhausted_needs_attention");
});
it("withholds retry from a viewer without operator authority", () => {
const state = project({
recoveryActions: [action()],
viewerCanRetry: false,
}).get("comment-1");
expect(state?.canRetry).toBe(false);
expect(state?.state).toBe("exhausted_needs_attention");
});
it("ignores recovery actions of other kinds and causes", () => {
const states = project({
recoveryActions: [
action({ kind: "stranded_assigned_issue" }),
action({ id: "action-2", cause: "other_cause" }),
],
});
expect(states.size).toBe(0);
});
it("keeps the highest generation when several replays share a fingerprint", () => {
const state = project({
pendingReplays: [
replay({ generation: 0, runId: "run-old" }),
replay({ generation: 1, runId: "run-new" }),
],
}).get("comment-1");
expect(state).toMatchObject({ attempt: 1, targetRunId: "run-new" });
});
it("ignores replays that carry no fingerprint", () => {
expect(project({ pendingReplays: [replay({ fingerprint: null })] }).size).toBe(0);
});
});
describe("toSafeFeedbackDeliveryFailureReason", () => {
it("maps known codes to plain operator language", () => {
expect(toSafeFeedbackDeliveryFailureReason("agent_not_invokable")).toBe(
"The assigned agent can't be started right now.",
);
expect(toSafeFeedbackDeliveryFailureReason(" budget_blocked ")).toBe(
"The agent is over its budget.",
);
});
it("collapses anything unrecognized rather than echoing it", () => {
expect(toSafeFeedbackDeliveryFailureReason("Error: connect ECONNREFUSED 127.0.0.1:5432")).toBeNull();
expect(toSafeFeedbackDeliveryFailureReason("")).toBeNull();
expect(toSafeFeedbackDeliveryFailureReason(null)).toBeNull();
expect(toSafeFeedbackDeliveryFailureReason(undefined)).toBeNull();
});
it("uses no transport or harness jargon in operator copy", () => {
const jargon = ["wake", "adapter", "heartbeat", "pid", "stack"];
for (const code of ["process_lost", "server_shutdown_interrupted", "agent_not_invokable"]) {
const copy = toSafeFeedbackDeliveryFailureReason(code)?.toLowerCase() ?? "";
for (const term of jargon) expect(copy).not.toContain(term);
}
});
});
@@ -0,0 +1,258 @@
import type { IssueCommentFeedbackDelivery, IssueRecoveryActionStatus } from "@paperclipai/shared";
import {
FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
FEEDBACK_DELIVERY_RECOVERY_CAUSE,
} from "./feedback-delivery.js";
/**
* How long a recovered-after-attention notice keeps its own projected state.
* After this the comment falls back to no delivery state at all; the audit
* trail lives in activity history, not in a permanently decorated comment.
*/
export const FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS = 24 * 60 * 60 * 1000;
/**
* Operator-safe failure vocabulary. Run/gate failure codes are mapped through
* this whitelist so no adapter message, stack frame, request body, credential,
* or filesystem path can reach a comment surface. An unrecognized code is
* deliberately collapsed to `null` rather than echoed: the banner copy already
* states the operator-relevant fact ("Paperclip stopped retrying").
*/
const SAFE_FAILURE_REASONS: Record<string, string> = {
agent_not_found: "The assigned agent no longer exists.",
agent_not_invokable: "The assigned agent can't be started right now.",
budget_blocked: "The agent is over its budget.",
issue_paused: "Work on this task is paused.",
issue_dependencies_blocked: "This task is waiting on a blocker.",
issue_assignee_changed: "The assignee changed while delivering.",
issue_reassigned: "The assignee changed while delivering.",
issue_execution_lock_changed: "Another run took over this task.",
issue_review_participant_changed: "The review owner changed while delivering.",
issue_terminal_status: "The task was closed while delivering.",
process_lost: "The agent's process stopped before it read the comment.",
server_shutdown_interrupted: "Paperclip restarted before the agent read the comment.",
workspace_validation_failed: "The task's workspace couldn't be prepared.",
configuration_validation_failed: "Required agent configuration is missing.",
};
export function toSafeFeedbackDeliveryFailureReason(
code: string | null | undefined,
): string | null {
if (typeof code !== "string") return null;
const normalized = code.trim();
if (!normalized) return null;
return SAFE_FAILURE_REASONS[normalized] ?? null;
}
/** Minimal recovery-action shape this projection needs. */
export type FeedbackDeliveryRecoveryActionRow = {
id: string;
kind: string;
cause: string;
/** Raw DB column: narrowed against the active-status list rather than cast. */
status: string;
fingerprint: string;
evidence: unknown;
attemptCount: number;
maxAttempts: number | null;
lastAttemptAt: Date | string | null;
resolvedAt: Date | string | null;
createdAt: Date | string;
};
/**
* A replay this issue is currently waiting on: either a queued wake or a
* queued/scheduled run that carries feedback-delivery context.
*/
export type FeedbackDeliveryPendingReplay = {
fingerprint: string | null;
commentIds: readonly string[];
generation: number;
runId: string | null;
nextAttemptAt: Date | string | null;
requestedAt: Date | string | null;
};
export type FeedbackDeliveryProjectionInput = {
recoveryActions: readonly FeedbackDeliveryRecoveryActionRow[];
pendingReplays: readonly FeedbackDeliveryPendingReplay[];
/** Whether this viewer may request one more explicit delivery attempt. */
viewerCanRetry: boolean;
/** Whether this viewer may open run detail. Gates `targetRunId` entirely. */
viewerCanReadRuns: boolean;
now: Date;
recoveredWindowMs?: number;
};
const ACTIVE_STATUSES: readonly string[] = [
"active",
"escalated",
] satisfies readonly IssueRecoveryActionStatus[];
function toIso(value: Date | string | null | undefined): string | null {
if (!value) return null;
const date = value instanceof Date ? value : new Date(value);
return Number.isNaN(date.getTime()) ? null : date.toISOString();
}
function toMillis(value: Date | string | null | undefined): number | null {
if (!value) return null;
const date = value instanceof Date ? value : new Date(value);
return Number.isNaN(date.getTime()) ? null : date.getTime();
}
function readEvidenceCommentIds(evidence: unknown): string[] {
if (typeof evidence !== "object" || evidence === null) return [];
const raw = (evidence as Record<string, unknown>).outstandingCommentIds;
if (!Array.isArray(raw)) return [];
const ids: string[] = [];
for (const entry of raw) {
if (typeof entry !== "string") continue;
const trimmed = entry.trim();
if (trimmed && !ids.includes(trimmed)) ids.push(trimmed);
}
return ids;
}
function readEvidenceString(evidence: unknown, key: string): string | null {
if (typeof evidence !== "object" || evidence === null) return null;
const raw = (evidence as Record<string, unknown>)[key];
return typeof raw === "string" && raw.trim().length > 0 ? raw.trim() : null;
}
function readEvidenceNumber(evidence: unknown, key: string): number | null {
if (typeof evidence !== "object" || evidence === null) return null;
const raw = (evidence as Record<string, unknown>)[key];
return typeof raw === "number" && Number.isFinite(raw) ? raw : null;
}
function isFeedbackDeliveryAction(action: FeedbackDeliveryRecoveryActionRow) {
return action.kind === "feedback_delivery" && action.cause === FEEDBACK_DELIVERY_RECOVERY_CAUSE;
}
/**
* Projects feedback-delivery state onto the individual human comments it
* belongs to.
*
* Precedence is deliberate:
* 1. An active/escalated recovery action always wins, because the operator must
* keep seeing the actionable banner until the obligation is discharged — an
* in-flight explicit retry surfaces as `retryInFlight`, not as a state change
* that would make the banner (and its attention item) disappear.
* 2. Otherwise a queued replay renders the quiet `retrying` footer line.
* 3. Otherwise a recently resolved action renders the historical recovered
* notice, so attention disappearing is explained rather than silent.
*
* A newer feedback comment never clears an older obligation: each comment id is
* keyed independently off the evidence of the action that owns it.
*/
export function projectFeedbackDeliveryByComment(
input: FeedbackDeliveryProjectionInput,
): Map<string, IssueCommentFeedbackDelivery> {
const byComment = new Map<string, IssueCommentFeedbackDelivery>();
const recoveredWindowMs = input.recoveredWindowMs ?? FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS;
const nowMs = input.now.getTime();
const actions = input.recoveryActions.filter(isFeedbackDeliveryAction);
const activeActions = actions.filter((action) => ACTIVE_STATUSES.includes(action.status));
const activeFingerprints = new Set(activeActions.map((action) => action.fingerprint));
// Recovered notices come first so an active action or queued replay for the
// same comment overwrites them, never the other way round.
const resolvedActions = actions
.filter((action) => !ACTIVE_STATUSES.includes(action.status))
// A fingerprint that resolved and then re-opened is owned by the open one.
.filter((action) => !activeFingerprints.has(action.fingerprint))
.filter((action) => {
const resolvedMs = toMillis(action.resolvedAt);
return resolvedMs !== null && nowMs - resolvedMs <= recoveredWindowMs;
})
// Oldest first: the freshest resolution wins a shared comment.
.sort((a, b) => (toMillis(a.resolvedAt) ?? 0) - (toMillis(b.resolvedAt) ?? 0));
for (const action of resolvedActions) {
const commentIds = readEvidenceCommentIds(action.evidence);
const targetRunId = input.viewerCanReadRuns
? readEvidenceString(action.evidence, "sourceRunId")
: null;
for (const commentId of commentIds) {
byComment.set(commentId, {
state: "recovered_after_attention",
sourceCommentId: commentId,
attempt: readEvidenceNumber(action.evidence, "retryGeneration"),
maxAttempts: action.maxAttempts ?? FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
nextAttemptAt: null,
lastAttemptAt: toIso(action.lastAttemptAt),
deliveredAt: toIso(action.resolvedAt),
targetRunId,
canRetry: false,
retryInFlight: false,
safeFailureReason: null,
batchedCommentCount: commentIds.length,
});
}
}
const replaysByFingerprint = new Map<string, FeedbackDeliveryPendingReplay>();
for (const replay of input.pendingReplays) {
if (!replay.fingerprint) continue;
const existing = replaysByFingerprint.get(replay.fingerprint);
if (!existing || replay.generation > existing.generation) {
replaysByFingerprint.set(replay.fingerprint, replay);
}
}
for (const replay of replaysByFingerprint.values()) {
if (activeFingerprints.has(replay.fingerprint as string)) continue;
for (const commentId of replay.commentIds) {
byComment.set(commentId, {
state: "retrying",
sourceCommentId: commentId,
attempt: replay.generation,
maxAttempts: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
nextAttemptAt: toIso(replay.nextAttemptAt),
lastAttemptAt: toIso(replay.requestedAt),
deliveredAt: null,
targetRunId: input.viewerCanReadRuns ? replay.runId : null,
canRetry: false,
retryInFlight: true,
safeFailureReason: null,
batchedCommentCount: replay.commentIds.length,
});
}
}
for (const action of activeActions) {
const commentIds = readEvidenceCommentIds(action.evidence);
const replay = replaysByFingerprint.get(action.fingerprint) ?? null;
const retryInFlight = replay !== null;
const targetRunId = input.viewerCanReadRuns
? replay?.runId ?? readEvidenceString(action.evidence, "sourceRunId")
: null;
const safeFailureReason =
toSafeFeedbackDeliveryFailureReason(readEvidenceString(action.evidence, "lastFailureCode"))
?? toSafeFeedbackDeliveryFailureReason(readEvidenceString(action.evidence, "gateErrorCode"));
for (const commentId of commentIds) {
byComment.set(commentId, {
state: "exhausted_needs_attention",
sourceCommentId: commentId,
attempt: readEvidenceNumber(action.evidence, "retryGeneration"),
maxAttempts:
readEvidenceNumber(action.evidence, "maxAutomaticRetries")
?? action.maxAttempts
?? FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES,
nextAttemptAt: toIso(replay?.nextAttemptAt ?? null),
lastAttemptAt: toIso(action.lastAttemptAt ?? action.createdAt),
deliveredAt: null,
targetRunId,
canRetry: input.viewerCanRetry && !retryInFlight,
retryInFlight,
safeFailureReason,
batchedCommentCount: commentIds.length,
});
}
}
return byComment;
}
@@ -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,362 @@
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}`;
}
/**
* Explicit operator "Retry delivery" replays get their own key namespace: the
* automatic `:generation:N` keys are already spent by the time delivery is
* exhausted, so reusing them would silently swallow the operator's request.
* Keying on the recovery action's attempt count still collapses a double-click
* into one replay while allowing a fresh attempt after a failure.
*/
export function buildFeedbackDeliveryManualRetryIdempotencyKey(input: {
fingerprint: string;
attempt: number;
}) {
return `${input.fingerprint}:manual:${input.attempt}`;
}
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"),
};
}
+23
View File
@@ -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";
+391 -2
View File
@@ -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;