Merge origin/master into master

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-04-28 13:19:24 -05:00
commit ec2c5537fc
11 files changed
+278 -26

No files matched your search

@@ -1472,6 +1472,42 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
},
);
it("still re-enqueues stranded assigned todo recovery when an old queued wake exists", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "todo",
runStatus: "failed",
});
await db.insert(agentWakeupRequests).values({
companyId,
agentId,
source: "assignment",
triggerDetail: "system",
reason: "issue_assigned",
payload: { issueId },
status: "queued",
});
const heartbeat = heartbeatService(db);
const result = await heartbeat.reconcileStrandedAssignedIssues();
expect(result.assignmentDispatched).toBe(0);
expect(result.dispatchRequeued).toBe(1);
expect(result.continuationRequeued).toBe(0);
expect(result.escalated).toBe(0);
expect(result.issueIds).toEqual([issueId]);
const runs = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.agentId, agentId));
expect(runs).toHaveLength(2);
const retryRun = runs.find((row) => row.id !== runId);
expect((retryRun?.contextSnapshot as Record<string, unknown>)?.retryReason).toBe("assignment_recovery");
if (retryRun) {
await waitForRunToSettle(heartbeat, retryRun.id);
}
});
it("blocks assigned todo work after the one automatic dispatch recovery was already used", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "todo",
@@ -460,6 +460,89 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
expect(inboxIssues.map((issue) => issue.id)).toContain(previousIssue.id);
});
it("touches a skipped active routine issue for the manual runner's inbox", async () => {
const { agentId, companyId, issueSvc, routine, svc } = await seedFixture();
const userId = randomUUID();
const previousRunId = randomUUID();
const liveHeartbeatRunId = randomUUID();
await db
.update(routines)
.set({ concurrencyPolicy: "skip_if_active" })
.where(eq(routines.id, routine.id));
const previousIssue = await issueSvc.create(companyId, {
projectId: routine.projectId,
title: routine.title,
description: routine.description,
status: "in_progress",
priority: routine.priority,
assigneeAgentId: routine.assigneeAgentId,
originKind: "routine_execution",
originId: routine.id,
originRunId: previousRunId,
});
await db.insert(routineRuns).values({
id: previousRunId,
companyId,
routineId: routine.id,
triggerId: null,
source: "manual",
status: "issue_created",
triggeredAt: new Date("2026-03-20T12:00:00.000Z"),
linkedIssueId: previousIssue.id,
});
await db.insert(heartbeatRuns).values({
id: liveHeartbeatRunId,
companyId,
agentId,
invocationSource: "assignment",
triggerDetail: "system",
status: "running",
contextSnapshot: { issueId: previousIssue.id },
startedAt: new Date("2026-03-20T12:01:00.000Z"),
});
await db
.update(issues)
.set({
checkoutRunId: liveHeartbeatRunId,
executionRunId: liveHeartbeatRunId,
executionLockedAt: new Date("2026-03-20T12:01:00.000Z"),
})
.where(eq(issues.id, previousIssue.id));
await db.insert(issueInboxArchives).values({
companyId,
issueId: previousIssue.id,
userId,
archivedAt: new Date("2026-03-20T12:02:00.000Z"),
});
const run = await svc.runRoutine(routine.id, { source: "manual" }, { userId });
expect(run.status).toBe("skipped");
expect(run.linkedIssueId).toBe(previousIssue.id);
await expect(
db.select().from(issueInboxArchives).where(eq(issueInboxArchives.issueId, previousIssue.id)),
).resolves.toHaveLength(0);
await expect(
db.select().from(issueReadStates).where(eq(issueReadStates.issueId, previousIssue.id)),
).resolves.toEqual([
expect.objectContaining({
companyId,
issueId: previousIssue.id,
userId,
}),
]);
const inboxIssues = await issueSvc.list(companyId, {
touchedByUserId: userId,
inboxArchivedByUserId: userId,
includeRoutineExecutions: true,
});
expect(inboxIssues.map((issue) => issue.id)).toContain(previousIssue.id);
});
it("does not coalesce live routine runs with different resolved variables", async () => {
const { companyId, agentId, projectId, svc } = await seedFixture();
const variableRoutine = await svc.create(
+21 -1
View File
@@ -345,7 +345,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
.where(
and(
eq(agentWakeupRequests.companyId, companyId),
inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
eq(agentWakeupRequests.status, "deferred_issue_execution"),
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
),
)
@@ -356,6 +356,21 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
return Boolean(run || deferredWake);
}
async function hasQueuedIssueWake(companyId: string, issueId: string) {
return db
.select({ id: agentWakeupRequests.id })
.from(agentWakeupRequests)
.where(
and(
eq(agentWakeupRequests.companyId, companyId),
eq(agentWakeupRequests.status, "queued"),
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
),
)
.limit(1)
.then((rows) => Boolean(rows[0]));
}
async function enqueueStrandedIssueRecovery(input: {
issueId: string;
agentId: string;
@@ -1619,6 +1634,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
if (issue.status === "todo") {
if (!latestRun) {
if (await hasQueuedIssueWake(issue.companyId, issue.id)) {
result.skipped += 1;
continue;
}
if (await isInvocationBudgetBlocked(issue, agentId)) {
result.skipped += 1;
continue;