Revert "Split run progress from liveness disposition"

This reverts commit 55518170ad.
This commit is contained in:
Dotta committed 2026-04-29 16:02:22 -05:00
1 parent caf37a355d
commit dd57bfba21
40 files changed
+141 -3563

No files matched your search

@@ -260,48 +260,6 @@ describe("approval routes idempotent retries", () => {
expect(mockApprovalService.reject).toHaveBeenCalledWith("approval-5", "user-1", "not now");
});
it("wakes the requesting agent when an approval is rejected", async () => {
mockApprovalService.getById.mockResolvedValue({
id: "approval-7",
companyId: "company-1",
type: "request_board_approval",
status: "pending",
payload: {},
requestedByAgentId: "agent-1",
});
mockApprovalService.reject.mockResolvedValue({
approval: {
id: "approval-7",
companyId: "company-1",
type: "request_board_approval",
status: "rejected",
payload: {},
requestedByAgentId: "agent-1",
},
applied: true,
});
mockIssueApprovalService.listIssuesForApproval.mockResolvedValue([{ id: "issue-1" }, { id: "issue-2" }]);
const res = await request(await createApp())
.post("/api/approvals/approval-7/reject")
.send({ decisionNote: "not approved" });
expect(res.status).toBe(200);
expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith("agent-1", expect.objectContaining({
reason: "approval_rejected",
payload: expect.objectContaining({
approvalId: "approval-7",
approvalStatus: "rejected",
issueId: "issue-1",
issueIds: ["issue-1", "issue-2"],
}),
contextSnapshot: expect.objectContaining({
wakeReason: "approval_rejected",
taskId: "issue-1",
}),
}));
});
it("derives approval attribution from the authenticated actor on request revision", async () => {
mockApprovalService.getById.mockResolvedValue({
id: "approval-6",
@@ -7,7 +7,6 @@ import {
agents,
agentRuntimeState,
agentWakeupRequests,
approvals,
budgetPolicies,
companySkills,
companies,
@@ -18,10 +17,8 @@ import {
heartbeatRunEvents,
heartbeatRuns,
issueComments,
issueApprovals,
issueDocuments,
issueRelations,
issueThreadInteractions,
issueTreeHoldMembers,
issueTreeHolds,
issues,
@@ -313,19 +310,14 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
await db.delete(companySkills);
await db.delete(costEvents);
await db.delete(issueComments);
await db.delete(issueThreadInteractions);
await db.delete(issueApprovals);
await db.delete(issueDocuments);
await db.delete(documentRevisions);
await db.delete(documents);
await db.delete(approvals);
await db.delete(issueRelations);
await db.delete(issueTreeHoldMembers);
await db.delete(issueTreeHolds);
for (let attempt = 0; attempt < 5; attempt += 1) {
await db.delete(issueComments);
await db.delete(issueThreadInteractions);
await db.delete(issueApprovals);
await db.delete(issueDocuments);
try {
await db.delete(issues);
@@ -348,7 +340,6 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
}
await db.delete(agentWakeupRequests);
await db.delete(budgetPolicies);
await db.delete(approvals);
for (let attempt = 0; attempt < 5; attempt += 1) {
await db.delete(agentRuntimeState);
try {
@@ -482,9 +473,6 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
assignToUser?: boolean;
activePauseHold?: boolean;
livenessState?: "completed" | "advanced" | "plan_only" | "empty_response" | "blocked" | "failed" | "needs_followup" | null;
livenessReason?: string | null;
nextAction?: string | null;
continuationAttempt?: number | null;
runErrorCode?: string | null;
runError?: string | null;
}) {
@@ -559,9 +547,6 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
? null
: ("runError" in input ? input.runError : "run failed before issue advanced"),
livenessState: input.livenessState ?? null,
livenessReason: input.livenessReason ?? null,
nextAction: input.nextAction ?? null,
continuationAttempt: input.continuationAttempt ?? 0,
});
await db.insert(issues).values([
@@ -2125,126 +2110,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
expect(wakeups).toHaveLength(1);
});
it("queues a bounded liveness continuation for productive successful stranded in-progress work", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "advanced",
livenessReason: "Run made progress and left follow-up work",
nextAction: "Implement the remaining recovery branch.",
});
const heartbeat = heartbeatService(db);
const result = await heartbeat.reconcileStrandedAssignedIssues();
expect(result.continuationRequeued).toBe(1);
expect(result.escalated).toBe(0);
expect(result.issueIds).toEqual([issueId]);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(1);
expect(wakeups[0]?.payload).toMatchObject({
issueId,
sourceRunId: runId,
livenessState: "advanced",
continuationAttempt: 1,
instruction: "Implement the remaining recovery branch.",
});
expect(wakeups[0]?.idempotencyKey).toBe(`run_liveness_continuation:${issueId}:${runId}:advanced:1`);
const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null);
expect(sourceRun?.continuationAttempt).toBe(1);
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
expect(issue?.status).toBe("in_progress");
const recoveryIssues = await db
.select()
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery")));
expect(recoveryIssues).toHaveLength(0);
});
it("blocks productive successful stranded work when bounded liveness continuation is exhausted", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "advanced",
livenessReason: "Run still has follow-up work",
nextAction: "Try the remaining implementation step.",
continuationAttempt: 2,
});
const heartbeat = heartbeatService(db);
const result = await heartbeat.reconcileStrandedAssignedIssues();
expect(result.continuationRequeued).toBe(0);
expect(result.escalated).toBe(1);
expect(result.issueIds).toEqual([issueId]);
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
expect(issue?.status).toBe("blocked");
const recovery = await db
.select()
.from(issues)
.where(and(
eq(issues.companyId, companyId),
eq(issues.originKind, "stranded_issue_recovery"),
eq(issues.originId, issueId),
))
.then((rows) => rows[0] ?? null);
if (!recovery) throw new Error("Expected stranded recovery issue");
expect(recovery).toMatchObject({
companyId,
parentId: issueId,
assigneeAgentId: agentId,
originKind: "stranded_issue_recovery",
originId: issueId,
originRunId: runId,
});
await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([recovery.id]);
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
expect(comments).toHaveLength(1);
expect(comments[0]?.body).toContain("Bounded liveness continuation exhausted");
expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`);
});
it("does not duplicate an existing bounded liveness continuation wake for productive successful work", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "advanced",
nextAction: "Resume implementation from the recorded next action.",
});
await db.insert(agentWakeupRequests).values({
companyId,
agentId,
source: "automation",
triggerDetail: "system",
reason: "run_liveness_continuation",
payload: { issueId, sourceRunId: runId, livenessState: "advanced", continuationAttempt: 1 },
status: "queued",
idempotencyKey: `run_liveness_continuation:${issueId}:${runId}:advanced:1`,
});
const heartbeat = heartbeatService(db);
const result = await heartbeat.reconcileStrandedAssignedIssues();
expect(result.continuationRequeued).toBe(0);
expect(result.escalated).toBe(0);
expect(result.skipped).toBe(1);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(1);
});
it("escalates a successful automatic continuation when no live post-run path remains", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
it("records productive continuation instead of recovery when the latest automatic continuation succeeded", async () => {
const { agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
retryReason: "issue_continuation_needed",
@@ -2254,36 +2121,28 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
const result = await heartbeat.reconcileStrandedAssignedIssues();
expect(result.continuationRequeued).toBe(0);
expect(result.productiveContinuationObserved).toBe(0);
expect(result.productiveContinuationObserved).toBe(1);
expect(result.successfulContinuationObserved).toBe(0);
expect(result.escalated).toBe(1);
expect(result.issueIds).toEqual([issueId]);
expect(result.escalated).toBe(0);
expect(result.issueIds).toEqual([]);
const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
expect(issue?.status).toBe("blocked");
expect(issue?.status).toBe("in_progress");
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId));
expect(comments).toHaveLength(1);
expect(comments[0]?.body).toContain("successful automatic continuation");
expect(comments).toHaveLength(0);
const runs = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.agentId, agentId));
const sourceIssueRuns = runs.filter((row) => (row.contextSnapshot as Record<string, unknown> | null)?.issueId === issueId);
expect(sourceIssueRuns.map((row) => row.id)).toEqual([runId]);
const recoveryIssues = await db
.select()
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery")));
expect(recoveryIssues).toHaveLength(1);
expect(runs.map((row) => row.id)).toEqual([runId]);
const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId));
expect(wakeups.some((row) => (row.payload as Record<string, unknown> | null)?.issueId === recoveryIssues[0]?.id)).toBe(true);
expect(wakeups).toHaveLength(1);
});
it("does not treat a productive terminal run as healthy when in-progress work has no live path", async () => {
it.fails("does not treat a productive terminal run as healthy when in-progress work has no live path", async () => {
const { companyId, agentId, issueId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
@@ -2348,408 +2207,4 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
expect(runs).toHaveLength(1);
});
it("suppresses bounded liveness continuation wakes when the issue is held by an active pause hold", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "plan_only",
livenessReason: "Run made no concrete progress",
nextAction: "Take the first concrete action now.",
activePauseHold: true,
});
const heartbeat = heartbeatService(db);
const sourceRun = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null);
if (!sourceRun) throw new Error("Expected source run");
await heartbeat.handleRunLivenessContinuationForTest(sourceRun);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(0);
const refreshedRun = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null);
expect(refreshedRun?.continuationAttempt).toBe(0);
expect(refreshedRun?.livenessReason).toContain("continuation suppressed by active pause hold");
const recoveryIssues = await db
.select()
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery")));
expect(recoveryIssues).toHaveLength(0);
const refreshedIssue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null);
expect(refreshedIssue?.status).toBe("in_progress");
expect(refreshedIssue?.assigneeAgentId).toBe(agentId);
});
it("suppresses bounded liveness continuation wakes when the issue has a pending confirmation", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "plan_only",
livenessReason: "Run produced a plan and needs approval",
nextAction: "Wait for the board to confirm the plan.",
});
await db.insert(issueThreadInteractions).values({
companyId,
issueId,
kind: "request_confirmation",
status: "pending",
payload: {
version: 1,
prompt: "Approve the plan?",
},
createdByAgentId: agentId,
sourceRunId: runId,
});
const heartbeat = heartbeatService(db);
const sourceRun = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null);
if (!sourceRun) throw new Error("Expected source run");
await heartbeat.handleRunLivenessContinuationForTest(sourceRun);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(0);
const refreshedRun = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null);
expect(refreshedRun?.continuationAttempt).toBe(0);
expect(refreshedRun?.livenessReason).toContain("continuation suppressed by explicit interaction or approval wait");
});
it("does not suppress bounded liveness continuation for stale agent-authored confirmations", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "plan_only",
livenessReason: "Run produced a plan and needs approval",
nextAction: "Continue after recovering from the stale confirmation.",
});
await db.insert(issueThreadInteractions).values({
companyId,
issueId,
kind: "request_confirmation",
status: "pending",
payload: {
version: 1,
prompt: "Approve the stale plan?",
},
createdByAgentId: agentId,
sourceRunId: runId,
createdAt: new Date("2026-03-19T00:06:00.000Z"),
updatedAt: new Date("2026-03-19T00:06:00.000Z"),
});
const heartbeat = heartbeatService(db);
const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null);
if (!sourceRun) throw new Error("Expected source run");
await heartbeat.handleRunLivenessContinuationForTest(sourceRun);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(1);
});
it("does not suppress bounded liveness continuation for stale agent-requested approvals", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "plan_only",
livenessReason: "Run requested approval before continuing",
nextAction: "Continue after recovering from the stale approval.",
});
const approvalId = randomUUID();
await db.insert(approvals).values({
id: approvalId,
companyId,
type: "request_board_approval",
requestedByAgentId: agentId,
status: "pending",
payload: {},
createdAt: new Date("2026-03-19T00:06:00.000Z"),
updatedAt: new Date("2026-03-19T00:06:00.000Z"),
});
await db.insert(issueApprovals).values({
companyId,
issueId,
approvalId,
linkedByAgentId: agentId,
createdAt: new Date("2026-03-19T00:06:00.000Z"),
});
const heartbeat = heartbeatService(db);
const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null);
if (!sourceRun) throw new Error("Expected source run");
await heartbeat.handleRunLivenessContinuationForTest(sourceRun);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(1);
});
it("does not re-dispatch assigned work while a live pending confirmation owns the next step", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "todo",
runStatus: "failed",
});
await db.insert(issueThreadInteractions).values({
companyId,
issueId,
kind: "request_confirmation",
status: "pending",
payload: {
version: 1,
prompt: "Approve the next action?",
},
createdByAgentId: agentId,
sourceRunId: runId,
});
const heartbeat = heartbeatService(db);
const result = await heartbeat.reconcileStrandedAssignedIssues();
expect(result.dispatchRequeued).toBe(0);
expect(result.continuationRequeued).toBe(0);
expect(result.skipped).toBeGreaterThan(0);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "issue_assignment_recovery")));
expect(wakeups).toHaveLength(0);
});
it("expires a stale confirmation target and then queues bounded liveness continuation", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "plan_only",
livenessReason: "Run produced a plan and needs approval",
nextAction: "Continue from the current plan revision.",
});
const documentId = randomUUID();
const oldRevisionId = randomUUID();
const currentRevisionId = randomUUID();
const interactionId = randomUUID();
await db.insert(documents).values({
id: documentId,
companyId,
title: "Plan",
format: "markdown",
latestBody: "current plan",
latestRevisionId: currentRevisionId,
latestRevisionNumber: 2,
});
await db.insert(issueDocuments).values({
companyId,
issueId,
documentId,
key: "plan",
});
await db.insert(documentRevisions).values([
{
id: oldRevisionId,
companyId,
documentId,
revisionNumber: 1,
title: "Plan",
format: "markdown",
body: "old plan",
},
{
id: currentRevisionId,
companyId,
documentId,
revisionNumber: 2,
title: "Plan",
format: "markdown",
body: "current plan",
},
]);
await db.insert(issueThreadInteractions).values({
id: interactionId,
companyId,
issueId,
kind: "request_confirmation",
status: "pending",
continuationPolicy: "wake_assignee_on_accept",
payload: {
version: 1,
prompt: "Approve the plan?",
target: {
type: "issue_document",
issueId,
documentId,
key: "plan",
revisionId: oldRevisionId,
revisionNumber: 1,
},
},
createdByAgentId: agentId,
sourceRunId: runId,
});
const heartbeat = heartbeatService(db);
const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null);
if (!sourceRun) throw new Error("Expected source run");
await heartbeat.handleRunLivenessContinuationForTest(sourceRun);
const [interaction] = await db
.select()
.from(issueThreadInteractions)
.where(eq(issueThreadInteractions.id, interactionId));
expect(interaction?.status).toBe("expired");
expect((interaction?.result as Record<string, unknown> | null)?.outcome).toBe("stale_target");
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(1);
});
it("expires a confirmation superseded by a user comment and then queues bounded liveness continuation", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "plan_only",
livenessReason: "Run asked the board to confirm the next step",
nextAction: "Continue after incorporating the board comment.",
});
const interactionId = randomUUID();
await db.insert(issueThreadInteractions).values({
id: interactionId,
companyId,
issueId,
kind: "request_confirmation",
status: "pending",
payload: {
version: 1,
prompt: "Approve the next step?",
supersedeOnUserComment: true,
},
createdByAgentId: agentId,
sourceRunId: runId,
createdAt: new Date("2026-03-19T00:06:00.000Z"),
});
const commentId = randomUUID();
await db.insert(issueComments).values({
id: commentId,
companyId,
issueId,
authorUserId: "board",
body: "Use the narrower implementation path.",
createdAt: new Date("2026-03-19T00:07:00.000Z"),
});
const heartbeat = heartbeatService(db);
const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null);
if (!sourceRun) throw new Error("Expected source run");
await heartbeat.handleRunLivenessContinuationForTest(sourceRun);
const [interaction] = await db
.select()
.from(issueThreadInteractions)
.where(eq(issueThreadInteractions.id, interactionId));
expect(interaction?.status).toBe("expired");
expect(interaction?.result).toMatchObject({
outcome: "superseded_by_comment",
commentId,
});
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(1);
});
it("queues bounded liveness continuation once a linked approval is resolved", async () => {
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "plan_only",
livenessReason: "Run requested approval before continuing",
nextAction: "Continue after the approval decision.",
});
const approvalId = randomUUID();
await db.insert(approvals).values({
id: approvalId,
companyId,
type: "request_board_approval",
requestedByAgentId: agentId,
status: "rejected",
payload: {},
decidedByUserId: "board",
decidedAt: new Date("2026-03-19T00:06:00.000Z"),
});
await db.insert(issueApprovals).values({
companyId,
issueId,
approvalId,
});
const heartbeat = heartbeatService(db);
const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null);
if (!sourceRun) throw new Error("Expected source run");
await heartbeat.handleRunLivenessContinuationForTest(sourceRun);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(1);
});
it("queues a bounded liveness continuation when no pause hold suppresses it", async () => {
const { companyId, runId } = await seedStrandedIssueFixture({
status: "in_progress",
runStatus: "succeeded",
livenessState: "plan_only",
livenessReason: "Run made no concrete progress",
nextAction: "Take the first concrete action now.",
});
const heartbeat = heartbeatService(db);
const sourceRun = await db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.then((rows) => rows[0] ?? null);
if (!sourceRun) throw new Error("Expected source run");
await heartbeat.handleRunLivenessContinuationForTest(sourceRun);
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation")));
expect(wakeups).toHaveLength(1);
});
});
@@ -2,7 +2,6 @@ import { randomUUID } from "node:crypto";
import { eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
activityLog,
agents,
agentRuntimeState,
agentWakeupRequests,
@@ -11,8 +10,6 @@ import {
environmentLeases,
heartbeatRunEvents,
heartbeatRuns,
issueTreeHoldMembers,
issueTreeHolds,
issues,
} from "@paperclipai/db";
import {
@@ -45,11 +42,8 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => {
}, 20_000);
afterEach(async () => {
await db.delete(activityLog);
await db.delete(heartbeatRunEvents);
await db.delete(environmentLeases);
await db.delete(issueTreeHoldMembers);
await db.delete(issueTreeHolds);
await db.delete(issues);
await db.delete(heartbeatRuns);
await db.delete(agentWakeupRequests);
@@ -840,132 +834,4 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => {
retryNotBefore.toISOString(),
);
});
it("cancels a due scheduled retry whose issue is held by an active pause hold", async () => {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const sourceRunId = randomUUID();
const now = new Date("2026-04-20T20:00:00.000Z");
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "active",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {
heartbeat: {
wakeOnDemand: true,
maxConcurrentRuns: 1,
},
},
permissions: {},
});
await db.insert(heartbeatRuns).values({
id: sourceRunId,
companyId,
agentId,
invocationSource: "assignment",
triggerDetail: "system",
status: "failed",
error: "upstream overload",
errorCode: "adapter_failed",
finishedAt: now,
contextSnapshot: {
issueId,
wakeReason: "issue_assigned",
},
updatedAt: now,
createdAt: now,
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Pause-hold scheduled retry",
status: "todo",
priority: "medium",
assigneeAgentId: agentId,
executionRunId: sourceRunId,
executionAgentNameKey: "codexcoder",
executionLockedAt: now,
issueNumber: 1,
identifier: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-1`,
});
const scheduled = await heartbeat.scheduleBoundedRetry(sourceRunId, {
now,
random: () => 0.5,
});
expect(scheduled.outcome).toBe("scheduled");
if (scheduled.outcome !== "scheduled") return;
await db.insert(issueTreeHolds).values({
companyId,
rootIssueId: issueId,
mode: "pause",
status: "active",
reason: "pause scheduled retry promotion",
releasePolicy: { strategy: "manual" },
});
const promotion = await heartbeat.promoteDueScheduledRetries(scheduled.dueAt);
expect(promotion).toEqual({ promoted: 0, runIds: [] });
const cancelled = await db
.select({
status: heartbeatRuns.status,
errorCode: heartbeatRuns.errorCode,
error: heartbeatRuns.error,
wakeupRequestId: heartbeatRuns.wakeupRequestId,
})
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, scheduled.run.id))
.then((rows) => rows[0] ?? null);
expect(cancelled).toMatchObject({
status: "cancelled",
errorCode: "issue_paused",
});
expect(cancelled?.error).toContain("active subtree pause hold");
const wakeupRequestId = cancelled?.wakeupRequestId ?? null;
if (wakeupRequestId) {
const wakeup = await db
.select({ status: agentWakeupRequests.status })
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.id, wakeupRequestId))
.then((rows) => rows[0] ?? null);
expect(wakeup?.status).toBe("cancelled");
}
const issue = await db
.select({ executionRunId: issues.executionRunId })
.from(issues)
.where(eq(issues.id, issueId))
.then((rows) => rows[0] ?? null);
expect(issue?.executionRunId).toBeNull();
const event = await db
.select({
message: heartbeatRunEvents.message,
payload: heartbeatRunEvents.payload,
})
.from(heartbeatRunEvents)
.where(eq(heartbeatRunEvents.runId, scheduled.run.id))
.orderBy(sql`${heartbeatRunEvents.id} desc`)
.then((rows) => rows[0] ?? null);
expect(event?.message).toContain("active subtree pause hold");
expect(event?.payload).toMatchObject({ issueId });
});
});
@@ -4,13 +4,10 @@ import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
agents,
agentWakeupRequests,
approvals,
companies,
createDb,
heartbeatRuns,
issueApprovals,
issueRelations,
issueThreadInteractions,
issues,
} from "@paperclipai/db";
import {
@@ -42,9 +39,6 @@ describeEmbeddedPostgres("issue blocker attention", () => {
afterEach(async () => {
await db.delete(heartbeatRuns);
await db.delete(agentWakeupRequests);
await db.delete(issueThreadInteractions);
await db.delete(issueApprovals);
await db.delete(approvals);
await db.delete(issueRelations);
await db.delete(issues);
await db.delete(agents);
@@ -121,56 +115,6 @@ describeEmbeddedPostgres("issue blocker attention", () => {
return runId;
}
async function pendingInteraction(input: {
companyId: string;
issueId: string;
kind?: "request_confirmation" | "ask_user_questions";
agentId?: string | null;
userId?: string | null;
createdAt?: Date;
}) {
await db.insert(issueThreadInteractions).values({
companyId: input.companyId,
issueId: input.issueId,
kind: input.kind ?? "request_confirmation",
status: "pending",
createdByAgentId: input.agentId ?? null,
createdByUserId: input.userId ?? null,
createdAt: input.createdAt,
updatedAt: input.createdAt,
payload: {},
});
}
async function linkedPendingApproval(input: {
companyId: string;
issueId: string;
agentId: string;
userId?: string | null;
createdAt?: Date;
}) {
const approvalId = randomUUID();
await db.insert(approvals).values({
id: approvalId,
companyId: input.companyId,
type: "request_board_approval",
requestedByAgentId: input.userId ? null : input.agentId,
requestedByUserId: input.userId ?? null,
status: "pending",
payload: {},
createdAt: input.createdAt,
updatedAt: input.createdAt,
});
await db.insert(issueApprovals).values({
companyId: input.companyId,
issueId: input.issueId,
approvalId,
linkedByAgentId: input.agentId,
linkedByUserId: input.userId ?? null,
createdAt: input.createdAt,
});
}
it("classifies a blocked parent as covered when its child has a running execution path", async () => {
const { companyId, agentId } = await createCompany("PBC");
const parentId = await insertIssue({ companyId, identifier: "PBC-1", title: "Parent", status: "blocked" });
@@ -412,212 +356,6 @@ describeEmbeddedPostgres("issue blocker attention", () => {
});
});
async function recordedRun(input: {
companyId: string;
agentId: string;
issueId: string;
livenessState?: string | null;
continuationAttempt?: number;
}) {
const runId = randomUUID();
await db.insert(heartbeatRuns).values({
id: runId,
companyId: input.companyId,
agentId: input.agentId,
status: "succeeded",
contextSnapshot: { issueId: input.issueId },
livenessState: (input.livenessState ?? "advanced") as never,
continuationAttempt: input.continuationAttempt ?? 0,
});
return runId;
}
it("classifies a chain whose leaf had a productive run that exited without a continuation as recovery_needed/productive_run_stopped", async () => {
const { companyId, agentId } = await createCompany("PRR");
const parentId = await insertIssue({ companyId, identifier: "PRR-1", title: "Parent", status: "blocked" });
const leafId = await insertIssue({
companyId,
identifier: "PRR-2",
title: "Recovery leaf",
status: "in_progress",
assigneeAgentId: agentId,
});
await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId });
await recordedRun({ companyId, agentId, issueId: leafId, livenessState: "advanced" });
const issuesList = await svc.list(companyId, { status: "blocked,in_progress" });
const parent = issuesList.find((issue) => issue.id === parentId);
const leaf = issuesList.find((issue) => issue.id === leafId);
expect(parent?.blockerAttention).toMatchObject({
state: "recovery_needed",
reason: "productive_run_stopped",
sampleBlockerIdentifier: "PRR-2",
nextActionHint: "wake_to_continue",
});
expect(parent?.blockerAttention?.nextActionOwner).toMatchObject({ type: "agent", agentId });
// Leaf surface gets the same recovery_needed classification on its own
// blockerAttention so the leaf-perspective notice can render.
expect(leaf?.blockerAttention).toMatchObject({
state: "recovery_needed",
reason: "productive_run_stopped",
nextActionHint: "wake_to_continue",
});
expect(leaf?.blockerAttention?.nextActionOwner).toMatchObject({ type: "agent", agentId });
});
it("treats a pending plan confirmation on an in-progress leaf as explicit waiting instead of recovery", async () => {
const { companyId, agentId } = await createCompany("PRW");
const parentId = await insertIssue({ companyId, identifier: "PRW-1", title: "Parent", status: "blocked" });
const leafId = await insertIssue({
companyId,
identifier: "PRW-2",
title: "Plan confirmation leaf",
status: "in_progress",
assigneeAgentId: agentId,
});
await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId });
await recordedRun({ companyId, agentId, issueId: leafId, livenessState: "advanced" });
await pendingInteraction({ companyId, issueId: leafId, kind: "request_confirmation" });
const issuesList = await svc.list(companyId, { status: "blocked,in_progress" });
const parent = issuesList.find((issue) => issue.id === parentId);
const leaf = issuesList.find((issue) => issue.id === leafId);
expect(parent?.blockerAttention).toMatchObject({
state: "covered",
reason: "explicit_waiting",
coveredBlockerCount: 1,
sampleBlockerIdentifier: "PRW-2",
nextActionHint: "needs_human_review",
});
expect(parent?.blockerAttention?.nextActionOwner).toMatchObject({ type: "user" });
expect(leaf?.blockerAttention).toMatchObject({
state: "covered",
reason: "explicit_waiting",
sampleBlockerIdentifier: "PRW-2",
nextActionHint: "needs_human_review",
});
expect(leaf?.blockerAttention?.nextActionOwner).toMatchObject({ type: "user" });
});
it("treats linked pending approvals on blocked leaves as explicit waiting", async () => {
const { companyId, agentId } = await createCompany("PRA");
const parentId = await insertIssue({ companyId, identifier: "PRA-1", title: "Parent", status: "blocked" });
const leafId = await insertIssue({
companyId,
identifier: "PRA-2",
title: "Approval leaf",
status: "blocked",
assigneeAgentId: agentId,
});
await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId });
await linkedPendingApproval({ companyId, issueId: leafId, agentId });
const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "covered",
reason: "explicit_waiting",
coveredBlockerCount: 1,
sampleBlockerIdentifier: "PRA-2",
nextActionHint: "needs_human_review",
});
expect(parent?.blockerAttention?.nextActionOwner).toMatchObject({ type: "user" });
});
it("does not let stale agent-authored confirmations suppress recovery attention", async () => {
const { companyId, agentId } = await createCompany("PRS");
const parentId = await insertIssue({ companyId, identifier: "PRS-1", title: "Parent", status: "blocked" });
const leafId = await insertIssue({
companyId,
identifier: "PRS-2",
title: "Stale confirmation leaf",
status: "in_progress",
assigneeAgentId: agentId,
});
await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId });
await recordedRun({ companyId, agentId, issueId: leafId, livenessState: "advanced" });
await pendingInteraction({
companyId,
issueId: leafId,
kind: "request_confirmation",
agentId,
createdAt: new Date(Date.now() - 25 * 60 * 60 * 1000),
});
const issuesList = await svc.list(companyId, { status: "blocked,in_progress" });
const parent = issuesList.find((issue) => issue.id === parentId);
const leaf = issuesList.find((issue) => issue.id === leafId);
expect(parent?.blockerAttention).toMatchObject({
state: "recovery_needed",
reason: "productive_run_stopped",
sampleBlockerIdentifier: "PRS-2",
});
expect(leaf?.blockerAttention).toMatchObject({
state: "recovery_needed",
reason: "productive_run_stopped",
});
});
it("does not let stale agent-requested approvals suppress recovery attention", async () => {
const { companyId, agentId } = await createCompany("PRZ");
const parentId = await insertIssue({ companyId, identifier: "PRZ-1", title: "Parent", status: "blocked" });
const leafId = await insertIssue({
companyId,
identifier: "PRZ-2",
title: "Stale approval leaf",
status: "in_progress",
assigneeAgentId: agentId,
});
await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId });
await recordedRun({ companyId, agentId, issueId: leafId, livenessState: "advanced" });
await linkedPendingApproval({
companyId,
issueId: leafId,
agentId,
createdAt: new Date(Date.now() - 25 * 60 * 60 * 1000),
});
const issuesList = await svc.list(companyId, { status: "blocked,in_progress" });
const parent = issuesList.find((issue) => issue.id === parentId);
const leaf = issuesList.find((issue) => issue.id === leafId);
expect(parent?.blockerAttention).toMatchObject({
state: "recovery_needed",
reason: "productive_run_stopped",
sampleBlockerIdentifier: "PRZ-2",
});
expect(leaf?.blockerAttention).toMatchObject({
state: "recovery_needed",
reason: "productive_run_stopped",
});
});
it("escalates to continuation_exhausted when the latest run already used its bounded continuations", async () => {
const { companyId, agentId } = await createCompany("PRX");
const parentId = await insertIssue({ companyId, identifier: "PRX-1", title: "Parent", status: "blocked" });
const leafId = await insertIssue({
companyId,
identifier: "PRX-2",
title: "Exhausted leaf",
status: "in_progress",
assigneeAgentId: agentId,
});
await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId });
await recordedRun({ companyId, agentId, issueId: leafId, livenessState: "advanced", continuationAttempt: 2 });
const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "recovery_needed",
reason: "continuation_exhausted",
sampleBlockerIdentifier: "PRX-2",
nextActionHint: "create_recovery_issue",
});
});
it("does not treat a scheduled retry as actively covered work", async () => {
const { companyId, agentId } = await createCompany("PBY");
const parentId = await insertIssue({ companyId, identifier: "PBY-1", title: "Parent", status: "blocked" });
+2 -13
View File
@@ -365,23 +365,12 @@ describe("issue graph liveness classifier", () => {
{
name: "pending interaction",
issue: baseReviewIssue,
pendingInteractions: [{
companyId,
issueId: reviewIssueId,
status: "pending",
createdAt: new Date(),
}],
pendingInteractions: [{ companyId, issueId: reviewIssueId, status: "pending" }],
},
{
name: "pending approval",
issue: baseReviewIssue,
pendingApprovals: [{
companyId,
issueId: reviewIssueId,
status: "pending",
createdAt: new Date(),
updatedAt: new Date(),
}],
pendingApprovals: [{ companyId, issueId: reviewIssueId, status: "pending" }],
},
{
name: "open recovery issue",
@@ -49,13 +49,10 @@ function registerModuleMocks() {
})),
}),
clampIssueListLimit: (value: number) => value,
companyService: () => ({}),
ISSUE_LIST_DEFAULT_LIMIT: 500,
ISSUE_LIST_MAX_LIMIT: 1000,
documentService: () => ({}),
environmentService: () => ({}),
executionWorkspaceService: () => ({}),
executionWorkspaceServiceDirect: () => ({}),
feedbackService: () => ({
listIssueVotesForUser: vi.fn(async () => []),
saveIssueVote: vi.fn(async () => ({ vote: null, consentEnabledNow: false, sharingEnabled: false })),
@@ -573,108 +570,6 @@ describe.sequential("issue thread interaction routes", () => {
expect(mockHeartbeatService.wakeup).not.toHaveBeenCalled();
});
it("wakes the current assignee when a rejected confirmation uses wake-assignee continuation", async () => {
mockInteractionService.rejectInteraction.mockResolvedValueOnce({
id: "interaction-7",
companyId: "company-1",
issueId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
kind: "request_confirmation",
status: "rejected",
continuationPolicy: "wake_assignee",
idempotencyKey: null,
sourceCommentId: null,
sourceRunId: "run-7",
payload: {
version: 1,
prompt: "Apply this plan?",
},
result: {
version: 1,
outcome: "rejected",
reason: "Needs changes",
},
createdAt: "2026-04-20T12:00:00.000Z",
updatedAt: "2026-04-20T12:05:00.000Z",
resolvedAt: "2026-04-20T12:05:00.000Z",
});
const app = await createApp();
const res = await request(app)
.post("/api/issues/aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa/interactions/interaction-7/reject")
.send({ reason: "Needs changes" });
expect(res.status).toBe(200);
expect(mockHeartbeatService.wakeup).toHaveBeenCalledTimes(1);
expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(
ASSIGNEE_AGENT_ID,
expect.objectContaining({
reason: "issue_commented",
payload: expect.objectContaining({
interactionId: "interaction-7",
interactionKind: "request_confirmation",
interactionStatus: "rejected",
}),
}),
);
});
it("wakes the current assignee when an accepted confirmation expires because its target is stale", async () => {
mockInteractionService.acceptInteraction.mockResolvedValueOnce({
interaction: {
id: "interaction-6",
companyId: "company-1",
issueId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
kind: "request_confirmation",
status: "expired",
continuationPolicy: "wake_assignee_on_accept",
idempotencyKey: null,
sourceCommentId: null,
sourceRunId: "run-6",
payload: {
version: 1,
prompt: "Approve this plan?",
target: {
type: "issue_document",
key: "plan",
revisionId: "old-revision",
},
},
result: {
version: 1,
outcome: "stale_target",
staleTarget: {
type: "issue_document",
key: "plan",
revisionId: "old-revision",
},
},
createdAt: "2026-04-20T12:00:00.000Z",
updatedAt: "2026-04-20T12:05:00.000Z",
resolvedAt: "2026-04-20T12:05:00.000Z",
},
createdIssues: [],
});
const app = await createApp();
const res = await request(app)
.post("/api/issues/aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa/interactions/interaction-6/accept")
.send({});
expect(res.status).toBe(200);
expect(mockHeartbeatService.wakeup).toHaveBeenCalledTimes(1);
expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(
ASSIGNEE_AGENT_ID,
expect.objectContaining({
reason: "issue_commented",
payload: expect.objectContaining({
interactionId: "interaction-6",
interactionKind: "request_confirmation",
interactionStatus: "expired",
}),
}),
);
});
it("does not emit an accept-only continuation wake for rejected suggested tasks", async () => {
mockInteractionService.rejectInteraction.mockResolvedValueOnce({
id: "interaction-5",
@@ -768,206 +768,6 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => {
});
});
it("moves agent-authored plan confirmations into review while preserving the return assignee", async () => {
const companyId = randomUUID();
const goalId = randomUUID();
const issueId = randomUUID();
const agentId = randomUUID();
const documentId = randomUUID();
const revisionId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: false });
await db.insert(goals).values({
id: goalId,
companyId,
title: "Confirm a plan",
level: "task",
status: "active",
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "active",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
goalId,
title: "Plan the implementation",
status: "in_progress",
priority: "medium",
assigneeAgentId: agentId,
});
await db.insert(documents).values({
id: documentId,
companyId,
title: "Plan",
format: "markdown",
latestBody: "v1",
latestRevisionId: revisionId,
latestRevisionNumber: 1,
});
await db.insert(issueDocuments).values({
companyId,
issueId,
documentId,
key: "plan",
});
await db.insert(documentRevisions).values({
id: revisionId,
companyId,
documentId,
revisionNumber: 1,
title: "Plan",
format: "markdown",
body: "v1",
});
const created = await interactionsSvc.create({
id: issueId,
companyId,
}, {
kind: "request_confirmation",
idempotencyKey: `confirmation:${issueId}:plan:${revisionId}`,
continuationPolicy: "wake_assignee",
payload: {
version: 1,
prompt: "Approve this plan?",
target: {
type: "issue_document",
issueId,
documentId,
key: "plan",
revisionId,
revisionNumber: 1,
},
},
}, {
agentId,
});
expect(created).toMatchObject({
kind: "request_confirmation",
status: "pending",
continuationPolicy: "wake_assignee",
});
const [reviewIssue] = await db.select().from(issues);
expect(reviewIssue).toMatchObject({
id: issueId,
status: "in_review",
assigneeAgentId: agentId,
assigneeUserId: null,
});
const idempotent = await interactionsSvc.create({
id: issueId,
companyId,
}, {
kind: "request_confirmation",
idempotencyKey: `confirmation:${issueId}:plan:${revisionId}`,
continuationPolicy: "wake_assignee",
payload: {
version: 1,
prompt: "Approve this plan?",
target: {
type: "issue_document",
issueId,
documentId,
key: "plan",
revisionId,
revisionNumber: 1,
},
},
}, {
agentId,
});
expect(idempotent.id).toBe(created.id);
expect(await db.select().from(issueThreadInteractions)).toHaveLength(1);
});
it("moves agent-authored questions into review while preserving the return assignee", async () => {
const companyId = randomUUID();
const goalId = randomUUID();
const issueId = randomUUID();
const agentId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: false });
await db.insert(goals).values({
id: goalId,
companyId,
title: "Ask a question",
level: "task",
status: "active",
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "active",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
goalId,
title: "Clarify the implementation",
status: "in_progress",
priority: "medium",
assigneeAgentId: agentId,
});
await interactionsSvc.create({
id: issueId,
companyId,
}, {
kind: "ask_user_questions",
continuationPolicy: "wake_assignee",
payload: {
version: 1,
questions: [{
id: "scope",
prompt: "Which scope?",
selectionMode: "single",
options: [{ id: "minimal", label: "Minimal" }],
}],
},
}, {
agentId,
});
const [reviewIssue] = await db.select().from(issues);
expect(reviewIssue).toMatchObject({
id: issueId,
status: "in_review",
assigneeAgentId: agentId,
assigneeUserId: null,
});
});
it("expires supersedable request confirmations when a user comments", async () => {
const companyId = randomUUID();
const goalId = randomUUID();
@@ -106,7 +106,7 @@ describe("run liveness continuations", () => {
expect(decision.nextAttempt).toBe(2);
});
it("treats an advanced terminal run as progress evidence, not a live continuation path", () => {
it.fails("treats an advanced terminal run as progress evidence, not a live continuation path", () => {
const decision = decideRunLivenessContinuation({
run: run(),
issue: issue(),
-67
View File
@@ -58,57 +58,6 @@ describe("run liveness classifier", () => {
expect(classification.lastUsefulActionAt).toBe(latestEvidenceAt);
});
it("preserves runnable next actions when concrete progress still leaves non-terminal work", () => {
const classification = classifyRunLiveness({
...baseInput,
resultJson: {
summary: "Updated the service and tests.\nNext action: run pnpm vitest server/src/__tests__/run-liveness.test.ts.",
},
evidence: {
issueCommentsCreated: 1,
latestEvidenceAt: new Date("2026-04-18T12:00:00Z"),
},
});
expect(classification.livenessState).toBe("advanced");
expect(classification.actionability).toBe("runnable");
expect(classification.nextAction).toBe("run pnpm vitest server/src/__tests__/run-liveness.test.ts.");
});
it("routes concrete progress with manager-review next action to visible follow-up", () => {
const classification = classifyRunLiveness({
...baseInput,
resultJson: {
summary: "Prepared the release notes.\nNext action: deploy to production and verify live traffic.",
},
evidence: {
issueCommentsCreated: 1,
latestEvidenceAt: new Date("2026-04-18T12:00:00Z"),
},
});
expect(classification.livenessState).toBe("needs_followup");
expect(classification.actionability).toBe("manager_review");
expect(classification.nextAction).toBe("deploy to production and verify live traffic.");
});
it("routes concrete progress with ambiguous next action to visible follow-up", () => {
const classification = classifyRunLiveness({
...baseInput,
resultJson: {
summary: "Captured the current state.\nNext action: decide what to do with the remaining uncertainty.",
},
evidence: {
issueCommentsCreated: 1,
latestEvidenceAt: new Date("2026-04-18T12:00:00Z"),
},
});
expect(classification.livenessState).toBe("needs_followup");
expect(classification.actionability).toBe("unknown");
expect(classification.nextAction).toBe("decide what to do with the remaining uncertainty.");
});
it("does not treat workspace operations alone as concrete progress", () => {
const classification = classifyRunLiveness({
...baseInput,
@@ -157,22 +106,6 @@ describe("run liveness classifier", () => {
expect(classification.livenessState).toBe("advanced");
});
it("keeps in-review work on the explicit waiting path", () => {
const classification = classifyRunLiveness({
...baseInput,
issue: {
...baseInput.issue,
status: "in_review",
},
resultJson: {
summary: "Next action: wait for manager review.",
},
});
expect(classification.livenessState).toBe("needs_followup");
expect(classification.nextAction).toBe("wait for manager review.");
});
it("classifies done issues as completed", () => {
const classification = classifyRunLiveness({
...baseInput,
+1 -72
View File
@@ -240,10 +240,6 @@ export function approvalRoutes(
const { approval, applied } = await svc.reject(id, decidedByUserId, req.body.decisionNote);
if (applied) {
const linkedIssues = await issueApprovalsSvc.listIssuesForApproval(approval.id);
const linkedIssueIds = linkedIssues.map((issue) => issue.id);
const primaryIssueId = linkedIssueIds[0] ?? null;
await logActivity(db, {
companyId: approval.companyId,
actorType: "user",
@@ -251,75 +247,8 @@ export function approvalRoutes(
action: "approval.rejected",
entityType: "approval",
entityId: approval.id,
details: {
type: approval.type,
requestedByAgentId: approval.requestedByAgentId,
linkedIssueIds,
},
details: { type: approval.type },
});
if (approval.requestedByAgentId) {
try {
const wakeRun = await heartbeat.wakeup(approval.requestedByAgentId, {
source: "automation",
triggerDetail: "system",
reason: "approval_rejected",
payload: {
approvalId: approval.id,
approvalStatus: approval.status,
issueId: primaryIssueId,
issueIds: linkedIssueIds,
},
requestedByActorType: "user",
requestedByActorId: req.actor.userId ?? "board",
contextSnapshot: {
source: "approval.rejected",
approvalId: approval.id,
approvalStatus: approval.status,
issueId: primaryIssueId,
issueIds: linkedIssueIds,
taskId: primaryIssueId,
wakeReason: "approval_rejected",
},
});
await logActivity(db, {
companyId: approval.companyId,
actorType: "user",
actorId: req.actor.userId ?? "board",
action: "approval.requester_wakeup_queued",
entityType: "approval",
entityId: approval.id,
details: {
requesterAgentId: approval.requestedByAgentId,
wakeRunId: wakeRun?.id ?? null,
linkedIssueIds,
},
});
} catch (err) {
logger.warn(
{
err,
approvalId: approval.id,
requestedByAgentId: approval.requestedByAgentId,
},
"failed to queue requester wakeup after approval rejection",
);
await logActivity(db, {
companyId: approval.companyId,
actorType: "user",
actorId: req.actor.userId ?? "board",
action: "approval.requester_wakeup_failed",
entityType: "approval",
entityId: approval.id,
details: {
requesterAgentId: approval.requestedByAgentId,
linkedIssueIds,
error: err instanceof Error ? err.message : String(err),
},
});
}
}
}
res.json(redactApprovalPayload(approval));
+1 -54
View File
@@ -208,13 +208,6 @@ function isExplicitResumeCapableStatus(status: string | null | undefined) {
return status === "done" || status === "blocked" || status === "todo" || status === "in_progress";
}
function getRequestConfirmationResolutionOutcome(interaction: { kind: string; result?: unknown }) {
if (interaction.kind !== "request_confirmation") return null;
if (typeof interaction.result !== "object" || interaction.result === null) return null;
const outcome = (interaction.result as { outcome?: unknown }).outcome;
return typeof outcome === "string" ? outcome : null;
}
function queueResolvedInteractionContinuationWakeup(input: {
heartbeat: ReturnType<typeof heartbeatService>;
issue: { id: string; assigneeAgentId: string | null; status: string };
@@ -225,7 +218,6 @@ function queueResolvedInteractionContinuationWakeup(input: {
continuationPolicy: string;
sourceCommentId?: string | null;
sourceRunId?: string | null;
result?: unknown;
};
actor: { actorType: "user" | "agent"; actorId: string };
source: string;
@@ -237,18 +229,8 @@ function queueResolvedInteractionContinuationWakeup(input: {
if (
input.interaction.continuationPolicy === "wake_assignee_on_accept"
&& input.interaction.status !== "accepted"
&& !(input.interaction.status === "expired" && getRequestConfirmationResolutionOutcome(input.interaction) === "stale_target")
) return;
if (
input.interaction.status === "expired"
&& !(
getRequestConfirmationResolutionOutcome(input.interaction) === "stale_target"
|| (
input.interaction.continuationPolicy === "wake_assignee"
&& getRequestConfirmationResolutionOutcome(input.interaction) === "superseded_by_comment"
)
)
) return;
if (input.interaction.status === "expired") return;
if (!input.issue.assigneeAgentId || isClosedIssueStatus(input.issue.status)) return;
void input.heartbeat.wakeup(input.issue.assigneeAgentId, {
@@ -3060,41 +3042,6 @@ export function issueRoutes(
},
});
const refreshedIssue = await svc.getById(id);
if (
refreshedIssue
&& (
refreshedIssue.status !== issue.status
|| refreshedIssue.assigneeAgentId !== issue.assigneeAgentId
|| refreshedIssue.assigneeUserId !== issue.assigneeUserId
)
) {
await logActivity(db, {
companyId: issue.companyId,
actorType: actor.actorType,
actorId: actor.actorId,
agentId: actor.agentId,
runId: actor.runId,
action: "issue.updated",
entityType: "issue",
entityId: issue.id,
details: {
identifier: issue.identifier,
status: refreshedIssue.status,
assigneeAgentId: refreshedIssue.assigneeAgentId ?? null,
assigneeUserId: refreshedIssue.assigneeUserId ?? null,
source: "issue.thread_interaction_created",
interactionId: interaction.id,
interactionKind: interaction.kind,
_previous: {
status: issue.status,
assigneeAgentId: issue.assigneeAgentId ?? null,
assigneeUserId: issue.assigneeUserId ?? null,
},
},
});
}
res.status(201).json(interaction);
});
+1 -171
View File
@@ -26,11 +26,8 @@ import {
issueDocuments,
heartbeatRunEvents,
heartbeatRuns,
approvals,
issueApprovals,
issueComments,
issueRelations,
issueThreadInteractions,
issues,
issueWorkProducts,
projects,
@@ -80,7 +77,6 @@ import {
sanitizeRuntimeServiceBaseEnv,
} from "./workspace-runtime.js";
import { issueService } from "./issues.js";
import { issueThreadInteractionService } from "./issue-thread-interactions.js";
import { parseIssueExecutionState } from "./issue-execution-policy.js";
import {
ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS,
@@ -113,10 +109,6 @@ import {
readContinuationAttempt,
} from "./recovery/index.js";
import { isAutomaticRecoverySuppressedByPauseHold } from "./recovery/pause-hold-guard.js";
import {
isLiveExplicitApprovalWaitingPath,
isLiveExplicitInteractionWaitingPath,
} from "./recovery/explicit-waiting-paths.js";
import { recoveryService } from "./recovery/service.js";
import { productivityReviewService } from "./productivity-review.js";
import { withAgentStartLock } from "./agent-start-lock.js";
@@ -136,8 +128,6 @@ import { environmentRuntimeService } from "./environment-runtime.js";
import { environmentRunOrchestrator } from "./environment-run-orchestrator.js";
import type { PluginWorkerManager } from "./plugin-worker-manager.js";
const EXPLICIT_WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions"];
const MAX_LIVE_LOG_CHUNK_BYTES = 8 * 1024;
const MAX_PERSISTED_LOG_CHUNK_CHARS = 64 * 1024;
const MAX_RUN_EVENT_PAYLOAD_STRING_CHARS = 16 * 1024;
@@ -2778,68 +2768,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
});
}
async function hasExplicitIssueWait(issue: {
companyId: string;
id: string;
assigneeUserId?: string | null;
executionState?: Record<string, unknown> | null;
}) {
await issueThreadInteractionService(db).expireInvalidPendingRequestConfirmationsForIssue(
{ id: issue.id, companyId: issue.companyId },
{ agentId: null, userId: null },
);
const [interactionRows, approvalRows] = await Promise.all([
db
.select({
issueId: issueThreadInteractions.issueId,
companyId: issueThreadInteractions.companyId,
status: issueThreadInteractions.status,
createdByUserId: issueThreadInteractions.createdByUserId,
createdAt: issueThreadInteractions.createdAt,
})
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.companyId, issue.companyId),
eq(issueThreadInteractions.issueId, issue.id),
eq(issueThreadInteractions.status, "pending"),
inArray(issueThreadInteractions.kind, EXPLICIT_WAITING_INTERACTION_KINDS),
),
)
.limit(5),
db
.select({
issueId: issueApprovals.issueId,
companyId: issueApprovals.companyId,
status: approvals.status,
requestedByUserId: approvals.requestedByUserId,
linkedByUserId: issueApprovals.linkedByUserId,
createdAt: approvals.createdAt,
updatedAt: approvals.updatedAt,
linkedAt: issueApprovals.createdAt,
})
.from(issueApprovals)
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
.where(
and(
eq(issueApprovals.companyId, issue.companyId),
eq(issueApprovals.issueId, issue.id),
inArray(approvals.status, ["pending", "revision_requested"]),
),
)
.limit(5),
]);
return interactionRows.some((row) => isLiveExplicitInteractionWaitingPath(issue, row)) ||
approvalRows.some((row) => isLiveExplicitApprovalWaitingPath(issue, row));
}
async function handleRunLivenessContinuation(run: typeof heartbeatRuns.$inferSelect) {
const livenessState = run.livenessState as RunLivenessState | null;
if (!livenessState) return;
const hasProgressFollowup = livenessState === "advanced" && Boolean(run.nextAction?.trim());
if (livenessState !== "plan_only" && livenessState !== "empty_response" && !hasProgressFollowup) return;
if (livenessState !== "plan_only" && livenessState !== "empty_response") return;
const context = parseObject(run.contextSnapshot);
const issueId = readNonEmptyString(context.issueId);
@@ -2854,7 +2785,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
title: issues.title,
status: issues.status,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
executionState: issues.executionState,
projectId: issues.projectId,
})
@@ -2880,14 +2810,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
})
: null;
if (issue) {
if (await hasExplicitIssueWait(issue)) {
await setRunStatus(run.id, run.status, {
livenessReason:
`${run.livenessReason ?? "Run ended without concrete progress"}; continuation suppressed by explicit interaction or approval wait`,
});
return;
}
const productivityHold = await productivityReviews.isProductivityReviewContinuationHoldActive({
companyId: issue.companyId,
issueId: issue.id,
@@ -2909,14 +2831,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
});
return;
}
if (await isAutomaticRecoverySuppressedByPauseHold(db, issue.companyId, issue.id, treeControlSvc)) {
await setRunStatus(run.id, run.status, {
livenessReason:
`${run.livenessReason ?? "Run ended without concrete progress"}; continuation suppressed by active pause hold`,
});
return;
}
}
const nextAttempt = readContinuationAttempt(run.continuationAttempt) + 1;
@@ -3741,88 +3655,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
});
continue;
}
const activePauseHold = await treeControlSvc.getActivePauseHoldGate(dueRun.companyId, dueRunIssueId);
if (activePauseHold) {
const reason = "Cancelled because issue is held by an active subtree pause hold";
const cancelled = await db
.update(heartbeatRuns)
.set({
status: "cancelled",
finishedAt: now,
error: reason,
errorCode: "issue_paused",
updatedAt: now,
})
.where(
and(
eq(heartbeatRuns.id, dueRun.id),
eq(heartbeatRuns.status, "scheduled_retry"),
lte(heartbeatRuns.scheduledRetryAt, now),
),
)
.returning()
.then((rows) => rows[0] ?? null);
if (!cancelled) continue;
if (cancelled.wakeupRequestId) {
await db
.update(agentWakeupRequests)
.set({
status: "cancelled",
finishedAt: now,
error: reason,
updatedAt: now,
})
.where(eq(agentWakeupRequests.id, cancelled.wakeupRequestId));
}
if (issue && issue.executionRunId === cancelled.id) {
await db
.update(issues)
.set({
executionRunId: null,
executionAgentNameKey: null,
executionLockedAt: null,
updatedAt: now,
})
.where(and(eq(issues.id, issue.id), eq(issues.executionRunId, cancelled.id)));
}
await appendRunEvent(cancelled, await nextRunEventSeq(cancelled.id), {
eventType: "lifecycle",
stream: "system",
level: "warn",
message: "Scheduled retry cancelled because issue is held by an active subtree pause hold",
payload: {
issueId: dueRunIssueId,
holdId: activePauseHold.holdId,
rootIssueId: activePauseHold.rootIssueId,
scheduledRetryAttempt: cancelled.scheduledRetryAttempt,
scheduledRetryAt: cancelled.scheduledRetryAt ? new Date(cancelled.scheduledRetryAt).toISOString() : null,
scheduledRetryReason: cancelled.scheduledRetryReason,
},
});
await logActivity(db, {
companyId: cancelled.companyId,
actorType: "system",
actorId: "system",
agentId: cancelled.agentId,
runId: cancelled.id,
action: "issue.tree_hold_run_interrupted",
entityType: "heartbeat_run",
entityId: cancelled.id,
details: {
issueId: dueRunIssueId,
holdId: activePauseHold.holdId,
rootIssueId: activePauseHold.rootIssueId,
source: "heartbeat.promote_due_scheduled_retries",
},
});
continue;
}
}
const promoted = await db
@@ -7663,8 +7495,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
reapOrphanedRuns,
handleRunLivenessContinuationForTest: handleRunLivenessContinuation,
promoteDueScheduledRetries,
resumeQueuedRuns,
@@ -1,5 +1,5 @@
import { isDeepStrictEqual } from "node:util";
import { and, asc, eq, gt, inArray, sql } from "drizzle-orm";
import { and, asc, eq, inArray } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
documents,
@@ -68,45 +68,6 @@ type IssueResolutionContext = {
assigneeUserId: string | null;
};
function isAgentAuthoredWaitingInteraction(input: CreateIssueThreadInteraction, actor: InteractionActor) {
return Boolean(actor.agentId) && (input.kind === "request_confirmation" || input.kind === "ask_user_questions");
}
async function moveIssueToReviewForAgentWaitingInteraction(
db: Db,
issue: { id: string; companyId: string },
input: CreateIssueThreadInteraction,
actor: InteractionActor,
) {
if (!actor.agentId || !isAgentAuthoredWaitingInteraction(input, actor)) return null;
const issueContext = await db
.select({
id: issues.id,
companyId: issues.companyId,
status: issues.status,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
.from(issues)
.where(eq(issues.id, issue.id))
.then((rows: IssueResolutionContext[]) => rows[0] ?? null);
if (!issueContext || issueContext.companyId !== issue.companyId) {
throw notFound("Issue not found");
}
if (issueContext.status !== "todo" && issueContext.status !== "in_progress") return null;
if (issueContext.assigneeUserId) return null;
return issueService(db).update(issue.id, {
status: "in_review",
assigneeAgentId: issueContext.assigneeAgentId ?? actor.agentId,
assigneeUserId: null,
actorAgentId: actor.agentId ?? null,
actorUserId: actor.userId ?? null,
});
}
function isIssueThreadInteractionIdempotencyConflict(error: unknown): boolean {
if (typeof error !== "object" || error === null) return false;
const err = error as { code?: string; constraint?: string; constraint_name?: string };
@@ -463,55 +424,6 @@ async function expireStaleRequestConfirmationTarget(db: Db | any, args: {
return hydrateInteraction(updated);
}
async function expireRequestConfirmationSupersededByLatestUserComment(db: Db | any, args: {
row: IssueThreadInteractionRow;
actor: InteractionActor;
}): Promise<IssueThreadInteraction | null> {
if (args.row.kind !== "request_confirmation" || args.row.status !== "pending") return null;
const interaction = hydrateInteraction(args.row) as RequestConfirmationInteraction;
if (interaction.payload.supersedeOnUserComment !== true) return null;
const [comment] = await db
.select({ id: issueComments.id })
.from(issueComments)
.where(and(
eq(issueComments.companyId, args.row.companyId),
eq(issueComments.issueId, args.row.issueId),
gt(issueComments.createdAt, args.row.createdAt),
sql`${issueComments.authorUserId} is not null`,
))
.orderBy(asc(issueComments.createdAt))
.limit(1);
if (!comment) return null;
const now = new Date();
const [updated] = await db
.update(issueThreadInteractions)
.set({
status: "expired",
result: {
version: 1,
outcome: "superseded_by_comment",
commentId: comment.id,
},
resolvedByAgentId: args.actor.agentId ?? null,
resolvedByUserId: args.actor.userId ?? null,
resolvedAt: now,
updatedAt: now,
})
.where(and(
eq(issueThreadInteractions.id, args.row.id),
eq(issueThreadInteractions.status, "pending"),
))
.returning();
if (!updated) {
throw conflict("Interaction has already been resolved");
}
await touchIssue(db, args.row.issueId);
return hydrateInteraction(updated);
}
export function issueThreadInteractionService(db: Db) {
async function getIdempotentInteraction(args: {
issueId: string;
@@ -728,7 +640,6 @@ export function issueThreadInteractionService(db: Db) {
idempotencyKey: data.idempotencyKey,
});
}
await moveIssueToReviewForAgentWaitingInteraction(db, issue, data, actor);
return hydrateInteraction(existing);
}
}
@@ -806,7 +717,6 @@ export function issueThreadInteractionService(db: Db) {
return hydrateInteraction(existing);
}
await moveIssueToReviewForAgentWaitingInteraction(db, issue, data, actor);
await touchIssue(db, issue.id);
return hydrateInteraction(created);
},
@@ -1182,33 +1092,6 @@ export function issueThreadInteractionService(db: Db) {
return expired;
},
expireInvalidPendingRequestConfirmationsForIssue: async (
issue: { id: string; companyId: string },
actor: InteractionActor,
) => {
const rows = await db
.select()
.from(issueThreadInteractions)
.where(and(
eq(issueThreadInteractions.companyId, issue.companyId),
eq(issueThreadInteractions.issueId, issue.id),
eq(issueThreadInteractions.kind, "request_confirmation"),
eq(issueThreadInteractions.status, "pending"),
));
const expired: IssueThreadInteraction[] = [];
for (const row of rows) {
const staleTarget = await expireStaleRequestConfirmationTarget(db, { row, actor });
if (staleTarget) {
expired.push(staleTarget);
continue;
}
const superseded = await expireRequestConfirmationSupersededByLatestUserComment(db, { row, actor });
if (superseded) expired.push(superseded);
}
return expired;
},
answerQuestions: async (
issue: { id: string; companyId: string },
interactionId: string,
+40 -462
View File
@@ -29,10 +29,6 @@ import {
} from "@paperclipai/db";
import type {
IssueBlockerAttention,
IssueBlockerAttentionReason,
IssueBlockerAttentionState,
IssueBlockerNextActionHint,
IssueBlockerNextActionOwner,
IssueProductivityReview,
IssueProductivityReviewTrigger,
IssueRelationIssueSummary,
@@ -55,10 +51,6 @@ import {
issueTreeControlService,
type ActiveIssueTreePauseHoldGate,
} from "./issue-tree-control.js";
import {
isLiveExplicitApprovalWaitingPath,
isLiveExplicitInteractionWaitingPath,
} from "./recovery/explicit-waiting-paths.js";
const ALL_ISSUE_STATUSES = ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"];
const MAX_ISSUE_COMMENT_PAGE_LIMIT = 500;
@@ -678,7 +670,6 @@ const ACTIVE_RUN_STATUSES = ["queued", "running"];
const BLOCKER_ATTENTION_ACTIVE_RUN_STATUSES = ["queued", "running"];
const BLOCKER_ATTENTION_ACTIVE_WAKE_STATUSES = ["queued", "deferred_issue_execution"];
const BLOCKER_ATTENTION_PENDING_INTERACTION_STATUSES = ["pending"];
const BLOCKER_ATTENTION_WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions"];
const BLOCKER_ATTENTION_PENDING_APPROVAL_STATUSES = ["pending", "revision_requested"];
const BLOCKER_ATTENTION_OPEN_RECOVERY_ORIGIN_KIND = "harness_liveness_escalation";
const PRODUCTIVITY_REVIEW_ORIGIN_KIND = "issue_productivity_review";
@@ -696,18 +687,6 @@ const BLOCKER_ATTENTION_OPEN_RECOVERY_TERMINAL_STATUSES = ["done", "cancelled"];
const BLOCKER_ATTENTION_MAX_DEPTH = 8;
const BLOCKER_ATTENTION_MAX_NODES = 2000;
const BLOCKER_ATTENTION_INVOKABLE_AGENT_STATUSES = new Set(["active", "idle", "running", "error"]);
// Leaves whose latest issue-linked run looks "productive but stranded" — the
// run produced concrete progress or described future work, but the issue is
// non-terminal and no continuation/wake/interaction took ownership of the next
// step. These leaves drive the recovery_needed surface.
const BLOCKER_ATTENTION_RECOVERY_LIVENESS_STATES = new Set([
"advanced",
"needs_followup",
"plan_only",
"empty_response",
]);
const BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES = new Set(["todo", "in_progress"]);
const BLOCKER_ATTENTION_DEFAULT_MAX_CONTINUATION_ATTEMPTS = 2;
type IssueBlockerAttentionNode = {
id: string;
@@ -717,7 +696,6 @@ type IssueBlockerAttentionNode = {
title: string;
status: string;
executionRunId?: string | null;
executionState?: Record<string, unknown> | null;
assigneeAgentId: string | null;
assigneeUserId: string | null;
};
@@ -744,10 +722,6 @@ type IssueBlockerAttentionAgentRow = {
companyId: string;
status: string;
};
type IssueBlockerExplicitWaitingPath = {
owner: IssueBlockerNextActionOwner;
hint: IssueBlockerNextActionHint;
};
async function activeRunMapForIssues(
dbOrTx: any,
@@ -796,113 +770,9 @@ function createIssueBlockerAttention(input: Partial<IssueBlockerAttention> = {})
attentionBlockerCount: input.attentionBlockerCount ?? 0,
sampleBlockerIdentifier: input.sampleBlockerIdentifier ?? null,
sampleStalledBlockerIdentifier: input.sampleStalledBlockerIdentifier ?? null,
nextActionOwner: input.nextActionOwner ?? null,
nextActionHint: input.nextActionHint ?? null,
};
}
type LeafRecoveryClassification = {
reason: Extract<
IssueBlockerAttentionReason,
"productive_run_stopped" | "continuation_exhausted" | "continuation_suppressed"
>;
hint: IssueBlockerNextActionHint;
owner: IssueBlockerNextActionOwner;
};
function leafRecoveryHintForReason(
reason: LeafRecoveryClassification["reason"],
owner: IssueBlockerNextActionOwner,
): IssueBlockerNextActionHint {
if (owner.type === "user") return "reassign";
if (owner.type === "none") return "reassign";
if (reason === "productive_run_stopped") return "wake_to_continue";
if (reason === "continuation_exhausted") return "create_recovery_issue";
return "needs_human_review";
}
function ownerForLeafRecovery(node: IssueBlockerAttentionNode): IssueBlockerNextActionOwner {
if (node.assigneeAgentId) {
return { type: "agent", agentId: node.assigneeAgentId, userId: null };
}
if (node.assigneeUserId) {
return { type: "user", agentId: null, userId: node.assigneeUserId };
}
return { type: "none", agentId: null, userId: null };
}
function ownerForExplicitWait(input: { assigneeAgentId?: string | null; assigneeUserId?: string | null }): IssueBlockerNextActionOwner {
if (input.assigneeUserId) {
return { type: "user", agentId: null, userId: input.assigneeUserId };
}
if (input.assigneeAgentId) {
return { type: "agent", agentId: input.assigneeAgentId, userId: null };
}
return { type: "user", agentId: null, userId: null };
}
function explicitWaitingPathForOwner(owner: IssueBlockerNextActionOwner): IssueBlockerExplicitWaitingPath {
return {
owner,
hint: "needs_human_review",
};
}
type IssueBlockerAttentionLatestRunRow = {
issueId: string | null;
status: string;
livenessState: string | null;
continuationAttempt: number | null;
finishedAt: Date | null;
};
type IssueBlockerAttentionLatestRun = {
status: string;
livenessState: string | null;
continuationAttempt: number | null;
finishedAt: Date | null;
};
function classifyLeafRecovery(input: {
node: IssueBlockerAttentionNode;
latestRun: IssueBlockerAttentionLatestRun | null;
hasActivePath: boolean;
hasWaitingPath: boolean;
hasProductivityReview: boolean;
agentInvokable: boolean;
maxContinuationAttempts: number;
}): LeafRecoveryClassification | null {
const { node, latestRun, hasActivePath, hasWaitingPath, hasProductivityReview, agentInvokable, maxContinuationAttempts } = input;
if (!BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES.has(node.status)) return null;
if (hasActivePath || hasWaitingPath) return null;
if (!latestRun) return null;
if (latestRun.status !== "succeeded") return null;
if (!latestRun.livenessState || !BLOCKER_ATTENTION_RECOVERY_LIVENESS_STATES.has(latestRun.livenessState)) return null;
const owner = ownerForLeafRecovery(node);
// If the leaf has no agent owner at all, fall through to needs_attention so
// the existing "no owner" surface continues to apply.
if (owner.type === "none") return null;
let reason: LeafRecoveryClassification["reason"];
if (hasProductivityReview) {
reason = "continuation_suppressed";
} else if ((latestRun.continuationAttempt ?? 0) >= maxContinuationAttempts) {
reason = "continuation_exhausted";
} else {
reason = "productive_run_stopped";
}
// An invokable agent who can be woken keeps the wake_to_continue affordance;
// otherwise the user needs to reassign or escalate.
let hint = leafRecoveryHintForReason(reason, owner);
if (owner.type === "agent" && !agentInvokable) {
hint = reason === "continuation_exhausted" ? "create_recovery_issue" : "needs_human_review";
}
return { reason, hint, owner };
}
function blockerSampleIdentifier(node: IssueBlockerAttentionNode | null | undefined) {
return node?.identifier ?? node?.id ?? null;
}
@@ -1136,27 +1006,18 @@ async function listIssueBlockerAttentionMap(
companyId: string,
issueRows: IssueBlockerAttentionInputNode[],
): Promise<Map<string, IssueBlockerAttention>> {
const sameCompanyRows = issueRows.filter((row) => row.companyId === companyId);
const roots = sameCompanyRows.filter((row) => row.status === "blocked");
// Non-blocked, non-terminal input issues with an agent owner are eligible to
// surface as invalid leaves on their own detail/inbox surfaces.
const recoverySeedRows = sameCompanyRows.filter((row) =>
BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES.has(row.status) && Boolean(row.assigneeAgentId),
);
const roots = issueRows.filter((row) => row.companyId === companyId && row.status === "blocked");
const attentionMap = new Map<string, IssueBlockerAttention>();
for (const row of issueRows) {
if (row.status !== "blocked") {
attentionMap.set(row.id, createIssueBlockerAttention());
}
}
if (roots.length === 0 && recoverySeedRows.length === 0) return attentionMap;
if (roots.length === 0) return attentionMap;
const nodesById = new Map<string, IssueBlockerAttentionNode>();
const edgesByIssueId = new Map<string, IssueBlockerAttentionEdge[]>();
for (const root of roots) nodesById.set(root.id, { ...root });
for (const seed of recoverySeedRows) {
if (!nodesById.has(seed.id)) nodesById.set(seed.id, { ...seed });
}
let frontier = roots.map((root) => root.id);
let truncated = false;
@@ -1175,7 +1036,6 @@ async function listIssueBlockerAttentionMap(
title: issues.title,
status: issues.status,
executionRunId: issues.executionRunId,
executionState: issues.executionState,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
@@ -1201,7 +1061,6 @@ async function listIssueBlockerAttentionMap(
title: issues.title,
status: issues.status,
executionRunId: issues.executionRunId,
executionState: issues.executionState,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
@@ -1301,73 +1160,26 @@ async function listIssueBlockerAttentionMap(
}
}
const nonTerminalNodeIds = [...nodesById.values()]
.filter((node) => node.status !== "done" && node.status !== "cancelled")
const reviewNodeIds = [...nodesById.values()]
.filter((node) => node.status === "in_review")
.map((node) => node.id);
const explicitWaitingByIssueId = new Map<string, IssueBlockerExplicitWaitingPath>();
const setExplicitWaitingPath = (issueId: string | null | undefined, path: IssueBlockerExplicitWaitingPath) => {
if (!issueId || explicitWaitingByIssueId.has(issueId)) return;
explicitWaitingByIssueId.set(issueId, path);
};
for (const node of nodesById.values()) {
if (node.status === "done" || node.status === "cancelled") continue;
if (node.assigneeUserId) {
setExplicitWaitingPath(node.id, explicitWaitingPathForOwner(ownerForExplicitWait(node)));
}
}
if (nonTerminalNodeIds.length > 0) {
for (const chunk of chunkList(nonTerminalNodeIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
const interactionRows: Array<{
issueId: string;
companyId: string;
status: string;
createdByUserId: string | null;
createdAt: Date;
}> = await dbOrTx
.select({
issueId: issueThreadInteractions.issueId,
companyId: issueThreadInteractions.companyId,
status: issueThreadInteractions.status,
createdByUserId: issueThreadInteractions.createdByUserId,
createdAt: issueThreadInteractions.createdAt,
})
const explicitWaitingIssueIds = new Set<string>();
if (reviewNodeIds.length > 0) {
for (const chunk of chunkList(reviewNodeIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
const interactionRows: Array<{ issueId: string }> = await dbOrTx
.select({ issueId: issueThreadInteractions.issueId })
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.companyId, companyId),
inArray(issueThreadInteractions.status, BLOCKER_ATTENTION_PENDING_INTERACTION_STATUSES),
inArray(issueThreadInteractions.kind, BLOCKER_ATTENTION_WAITING_INTERACTION_KINDS),
inArray(issueThreadInteractions.issueId, chunk),
),
);
for (const row of interactionRows) {
const node = nodesById.get(row.issueId);
if (!node || !isLiveExplicitInteractionWaitingPath(node, row)) continue;
setExplicitWaitingPath(row.issueId, explicitWaitingPathForOwner(ownerForExplicitWait({
assigneeUserId: node?.assigneeUserId ?? null,
})));
}
for (const row of interactionRows) explicitWaitingIssueIds.add(row.issueId);
const approvalRows: Array<{
issueId: string;
companyId: string;
status: string;
requestedByUserId: string | null;
linkedByUserId: string | null;
createdAt: Date;
updatedAt: Date;
linkedAt: Date;
}> = await dbOrTx
.select({
issueId: issueApprovals.issueId,
companyId: issueApprovals.companyId,
status: approvals.status,
requestedByUserId: approvals.requestedByUserId,
linkedByUserId: issueApprovals.linkedByUserId,
createdAt: approvals.createdAt,
updatedAt: approvals.updatedAt,
linkedAt: issueApprovals.createdAt,
})
const approvalRows: Array<{ issueId: string }> = await dbOrTx
.select({ issueId: issueApprovals.issueId })
.from(issueApprovals)
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
.where(
@@ -1377,24 +1189,10 @@ async function listIssueBlockerAttentionMap(
inArray(issueApprovals.issueId, chunk),
),
);
for (const row of approvalRows) {
const node = nodesById.get(row.issueId);
if (!node || !isLiveExplicitApprovalWaitingPath(node, row)) continue;
setExplicitWaitingPath(row.issueId, explicitWaitingPathForOwner(ownerForExplicitWait({
assigneeUserId: node?.assigneeUserId ?? null,
})));
}
for (const row of approvalRows) explicitWaitingIssueIds.add(row.issueId);
const recoveryRows: Array<{
originId: string | null;
assigneeAgentId: string | null;
assigneeUserId: string | null;
}> = await dbOrTx
.select({
originId: issues.originId,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
const recoveryRows: Array<{ originId: string | null }> = await dbOrTx
.select({ originId: issues.originId })
.from(issues)
.where(
and(
@@ -1406,7 +1204,7 @@ async function listIssueBlockerAttentionMap(
),
);
for (const row of recoveryRows) {
setExplicitWaitingPath(row.originId, explicitWaitingPathForOwner(ownerForExplicitWait(row)));
if (row.originId) explicitWaitingIssueIds.add(row.originId);
}
}
}
@@ -1423,165 +1221,40 @@ async function listIssueBlockerAttentionMap(
: [];
const agentsById = new Map(agentRows.map((agent) => [agent.id, agent]));
// Recovery-leaf inputs: latest issue-linked run + productivity-review hold.
// We only need this for non-blocked agent-owned non-terminal nodes; gate the
// queries on that subset so we do not pull runs for every blocker chain.
const recoveryCandidateIds = [...nodesById.values()]
.filter((node) =>
BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES.has(node.status) &&
Boolean(node.assigneeAgentId) &&
!activeIssueIds.has(node.id) &&
!explicitWaitingByIssueId.has(node.id),
)
.map((node) => node.id);
const latestRunByIssueId = new Map<string, IssueBlockerAttentionLatestRun>();
if (recoveryCandidateIds.length > 0) {
for (const chunk of chunkList(recoveryCandidateIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
const runRows: IssueBlockerAttentionLatestRunRow[] = await dbOrTx
.select({
issueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`,
status: heartbeatRuns.status,
livenessState: heartbeatRuns.livenessState,
continuationAttempt: heartbeatRuns.continuationAttempt,
finishedAt: heartbeatRuns.finishedAt,
})
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, companyId),
inArray(sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, chunk),
),
)
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id));
for (const row of runRows) {
if (!row.issueId) continue;
if (latestRunByIssueId.has(row.issueId)) continue;
latestRunByIssueId.set(row.issueId, {
status: row.status,
livenessState: row.livenessState,
continuationAttempt: row.continuationAttempt,
finishedAt: row.finishedAt,
});
}
}
}
const productivityReviewIssueIds = new Set<string>();
if (recoveryCandidateIds.length > 0) {
for (const chunk of chunkList(recoveryCandidateIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
const productivityRows: Array<{ originId: string | null }> = await dbOrTx
.select({ originId: issues.originId })
.from(issues)
.where(
and(
eq(issues.companyId, companyId),
eq(issues.originKind, PRODUCTIVITY_REVIEW_ORIGIN_KIND),
isNull(issues.hiddenAt),
inArray(issues.originId, chunk),
notInArray(issues.status, PRODUCTIVITY_REVIEW_TERMINAL_STATUSES),
),
);
for (const row of productivityRows) {
if (row.originId) productivityReviewIssueIds.add(row.originId);
}
}
}
const leafRecoveryByNodeId = new Map<string, LeafRecoveryClassification>();
for (const nodeId of recoveryCandidateIds) {
const node = nodesById.get(nodeId);
if (!node) continue;
const assignee = node.assigneeAgentId ? agentsById.get(node.assigneeAgentId) : null;
const agentInvokable = Boolean(
assignee && assignee.companyId === companyId && BLOCKER_ATTENTION_INVOKABLE_AGENT_STATUSES.has(assignee.status),
);
const classification = classifyLeafRecovery({
node,
latestRun: latestRunByIssueId.get(nodeId) ?? null,
hasActivePath: activeIssueIds.has(nodeId),
hasWaitingPath: explicitWaitingByIssueId.has(nodeId),
hasProductivityReview: productivityReviewIssueIds.has(nodeId),
agentInvokable,
maxContinuationAttempts: BLOCKER_ATTENTION_DEFAULT_MAX_CONTINUATION_ATTEMPTS,
});
if (classification) leafRecoveryByNodeId.set(nodeId, classification);
}
type PathClassification = {
covered: boolean;
coveredReason: IssueBlockerAttentionReason | null;
stalled: boolean;
sampleBlockerIdentifier: string | null;
sampleStalledBlockerIdentifier: string | null;
recoveryLeafIdentifier: string | null;
recovery: LeafRecoveryClassification | null;
waitingPath: IssueBlockerExplicitWaitingPath | null;
};
const emptyPath = (sample: string | null): PathClassification => ({
covered: false,
coveredReason: null,
stalled: false,
sampleBlockerIdentifier: sample,
sampleStalledBlockerIdentifier: null,
recoveryLeafIdentifier: null,
recovery: null,
waitingPath: null,
});
const coveredPath = (
sample: string | null,
coveredReason: IssueBlockerAttentionReason = null,
waitingPath: IssueBlockerExplicitWaitingPath | null = null,
): PathClassification => ({
covered: true,
coveredReason,
stalled: false,
sampleBlockerIdentifier: sample,
sampleStalledBlockerIdentifier: null,
recoveryLeafIdentifier: null,
recovery: null,
waitingPath,
});
const classifyPath = (
nodeId: string,
seen: Set<string>,
): PathClassification => {
const sample = blockerSampleIdentifier(nodesById.get(nodeId));
if (truncated || seen.has(nodeId)) {
return emptyPath(sample);
return { covered: false, stalled: false, sampleBlockerIdentifier: sample, sampleStalledBlockerIdentifier: null };
}
const node = nodesById.get(nodeId);
if (!node || node.companyId !== companyId) {
return emptyPath(nodeId);
return { covered: false, stalled: false, sampleBlockerIdentifier: nodeId, sampleStalledBlockerIdentifier: null };
}
const nodeSample = blockerSampleIdentifier(node);
if (node.status === "done") {
return coveredPath(nodeSample);
}
const explicitWaitingPath = explicitWaitingByIssueId.get(node.id) ?? null;
if (explicitWaitingPath) {
return coveredPath(nodeSample, "explicit_waiting", explicitWaitingPath);
return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null };
}
if (node.status === "in_review") {
if (activeIssueIds.has(node.id)) {
return coveredPath(nodeSample);
const hasWaitingPath = activeIssueIds.has(node.id) || Boolean(node.assigneeUserId) || explicitWaitingIssueIds.has(node.id);
if (hasWaitingPath) {
return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null };
}
return {
covered: false,
coveredReason: null,
stalled: true,
sampleBlockerIdentifier: nodeSample,
sampleStalledBlockerIdentifier: nodeSample,
recoveryLeafIdentifier: null,
recovery: null,
waitingPath: null,
};
return { covered: false, stalled: true, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: nodeSample };
}
if (activeIssueIds.has(node.id)) {
return coveredPath(nodeSample);
return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null };
}
if (node.status === "cancelled") {
return emptyPath(nodeSample);
return { covered: false, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null };
}
const downstream = (edgesByIssueId.get(node.id) ?? []).filter((edge) => nodesById.get(edge.blockerIssueId)?.status !== "done");
@@ -1591,95 +1264,48 @@ async function listIssueBlockerAttentionMap(
const classified = downstream.map((edge) => classifyPath(edge.blockerIssueId, nextSeen));
const stalledChild = classified.find((result) => result.stalled || result.sampleStalledBlockerIdentifier);
const sampleStalled = stalledChild?.sampleStalledBlockerIdentifier ?? null;
const recoveryChild = classified.find((result) => result.recovery && result.recoveryLeafIdentifier);
const recoveryLeafIdentifier = recoveryChild?.recoveryLeafIdentifier ?? null;
const recovery = recoveryChild?.recovery ?? null;
const waitingEntry = classified.find((result) => result.coveredReason === "explicit_waiting" && result.waitingPath);
const waitingPath = waitingEntry?.waitingPath ?? null;
const hardAttention = classified.find((result) => !result.covered && !result.stalled && !result.recovery);
const hardAttention = classified.find((result) => !result.covered && !result.stalled);
if (hardAttention) {
return {
covered: false,
coveredReason: null,
stalled: false,
sampleBlockerIdentifier: hardAttention.sampleBlockerIdentifier,
sampleStalledBlockerIdentifier: sampleStalled,
recoveryLeafIdentifier,
recovery,
waitingPath,
};
}
if (recoveryChild) {
return {
covered: false,
coveredReason: null,
stalled: false,
sampleBlockerIdentifier: recoveryLeafIdentifier ?? recoveryChild.sampleBlockerIdentifier,
sampleStalledBlockerIdentifier: sampleStalled,
recoveryLeafIdentifier,
recovery,
waitingPath,
};
}
const stalledEntry = classified.find((result) => result.stalled);
if (stalledEntry) {
return {
covered: false,
coveredReason: null,
stalled: true,
sampleBlockerIdentifier: stalledEntry.sampleBlockerIdentifier,
sampleStalledBlockerIdentifier: sampleStalled,
recoveryLeafIdentifier: null,
recovery: null,
waitingPath,
};
}
return {
covered: true,
coveredReason: waitingPath ? "explicit_waiting" : null,
stalled: false,
sampleBlockerIdentifier: waitingEntry?.sampleBlockerIdentifier ?? classified[0]?.sampleBlockerIdentifier ?? nodeSample,
sampleBlockerIdentifier: classified[0]?.sampleBlockerIdentifier ?? nodeSample,
sampleStalledBlockerIdentifier: null,
recoveryLeafIdentifier: null,
recovery: null,
waitingPath,
};
}
const recovery = leafRecoveryByNodeId.get(node.id) ?? null;
if (recovery) {
return {
covered: false,
coveredReason: null,
stalled: false,
sampleBlockerIdentifier: nodeSample,
sampleStalledBlockerIdentifier: null,
recoveryLeafIdentifier: nodeSample,
recovery,
waitingPath: null,
};
}
if (node.assigneeAgentId) {
const assignee = agentsById.get(node.assigneeAgentId);
if (!assignee || assignee.companyId !== companyId || !BLOCKER_ATTENTION_INVOKABLE_AGENT_STATUSES.has(assignee.status)) {
return emptyPath(nodeSample);
return { covered: false, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null };
}
}
return emptyPath(nodeSample);
return { covered: false, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null };
};
for (const root of roots) {
const topLevelEdges = (edgesByIssueId.get(root.id) ?? []).filter((edge) => nodesById.get(edge.blockerIssueId)?.status !== "done");
if (topLevelEdges.length === 0) {
const waitingPath = explicitWaitingByIssueId.get(root.id) ?? null;
attentionMap.set(root.id, createIssueBlockerAttention({
state: waitingPath ? "covered" : "needs_attention",
reason: waitingPath ? "explicit_waiting" : "attention_required",
sampleBlockerIdentifier: waitingPath ? blockerSampleIdentifier(nodesById.get(root.id)) : null,
nextActionOwner: waitingPath?.owner ?? null,
nextActionHint: waitingPath?.hint ?? null,
state: "needs_attention",
reason: "attention_required",
}));
continue;
}
@@ -1690,48 +1316,28 @@ async function listIssueBlockerAttentionMap(
}));
const coveredBlockerCount = classified.filter((entry) => entry.result.covered).length;
const stalledBlockerCount = classified.filter((entry) => entry.result.stalled).length;
const recoveryEntry = classified.find((entry) => entry.result.recovery && entry.result.recoveryLeafIdentifier);
const recoveryBlockerCount = classified.filter((entry) => entry.result.recovery).length;
const attentionBlockerCount = classified.length - coveredBlockerCount - stalledBlockerCount - recoveryBlockerCount;
const hardAttentionEntry = classified.find((entry) => !entry.result.covered && !entry.result.stalled && !entry.result.recovery);
const attentionBlockerCount = classified.length - coveredBlockerCount - stalledBlockerCount;
const hardAttentionEntry = classified.find((entry) => !entry.result.covered && !entry.result.stalled);
const stalledEntry = classified.find((entry) => entry.result.stalled);
const waitingEntry = classified.find((entry) => entry.result.coveredReason === "explicit_waiting" && entry.result.waitingPath);
const sampleEntry = hardAttentionEntry ?? recoveryEntry ?? stalledEntry ?? classified[0] ?? null;
const sampleEntry = hardAttentionEntry ?? stalledEntry ?? classified[0] ?? null;
const sampleNode = sampleEntry ? nodesById.get(sampleEntry.edge.blockerIssueId) : null;
const sampleStalledFromChain = classified
.map((entry) => entry.result.sampleStalledBlockerIdentifier)
.find((value) => value);
const recoveryLeafIdentifier = recoveryEntry?.result.recoveryLeafIdentifier ?? null;
const recoveryClassification = recoveryEntry?.result.recovery ?? null;
let state: IssueBlockerAttentionState;
let reason: IssueBlockerAttentionReason;
let nextActionOwner: IssueBlockerNextActionOwner | null = null;
let nextActionHint: IssueBlockerNextActionHint = null;
let sampleBlockerIdentifier =
sampleEntry?.result.sampleBlockerIdentifier ?? blockerSampleIdentifier(sampleNode);
let state: IssueBlockerAttention["state"];
let reason: IssueBlockerAttention["reason"];
if (attentionBlockerCount > 0) {
state = "needs_attention";
reason = "attention_required";
} else if (recoveryClassification) {
state = "recovery_needed";
reason = recoveryClassification.reason;
nextActionOwner = recoveryClassification.owner;
nextActionHint = recoveryClassification.hint;
sampleBlockerIdentifier = recoveryLeafIdentifier ?? sampleBlockerIdentifier;
} else if (stalledBlockerCount > 0) {
state = "stalled";
reason = "stalled_review";
} else {
state = "covered";
reason = waitingEntry
? "explicit_waiting"
: topLevelEdges.every((edge) => nodesById.get(edge.blockerIssueId)?.parentId === root.id)
? "active_child"
: "active_dependency";
nextActionOwner = waitingEntry?.result.waitingPath?.owner ?? null;
nextActionHint = waitingEntry?.result.waitingPath?.hint ?? null;
sampleBlockerIdentifier = waitingEntry?.result.sampleBlockerIdentifier ?? sampleBlockerIdentifier;
reason = topLevelEdges.every((edge) => nodesById.get(edge.blockerIssueId)?.parentId === root.id)
? "active_child"
: "active_dependency";
}
attentionMap.set(root.id, createIssueBlockerAttention({
@@ -1741,37 +1347,9 @@ async function listIssueBlockerAttentionMap(
coveredBlockerCount,
stalledBlockerCount,
attentionBlockerCount,
sampleBlockerIdentifier,
sampleBlockerIdentifier: sampleEntry?.result.sampleBlockerIdentifier ?? blockerSampleIdentifier(sampleNode),
sampleStalledBlockerIdentifier:
stalledEntry?.result.sampleStalledBlockerIdentifier ?? sampleStalledFromChain ?? null,
nextActionOwner,
nextActionHint,
}));
}
// Non-blocked recovery seeds: a leaf that is itself the invalid endpoint
// gets its own recovery_needed surface (e.g. opening the leaf directly from
// an inbox wake).
for (const seed of recoverySeedRows) {
const waitingPath = explicitWaitingByIssueId.get(seed.id) ?? null;
if (waitingPath) {
attentionMap.set(seed.id, createIssueBlockerAttention({
state: "covered",
reason: "explicit_waiting",
sampleBlockerIdentifier: blockerSampleIdentifier(nodesById.get(seed.id)) ?? seed.identifier ?? seed.id,
nextActionOwner: waitingPath.owner,
nextActionHint: waitingPath.hint,
}));
continue;
}
const recovery = leafRecoveryByNodeId.get(seed.id);
if (!recovery) continue;
attentionMap.set(seed.id, createIssueBlockerAttention({
state: "recovery_needed",
reason: recovery.reason,
sampleBlockerIdentifier: blockerSampleIdentifier(nodesById.get(seed.id)) ?? seed.identifier ?? seed.id,
nextActionOwner: recovery.owner,
nextActionHint: recovery.hint,
}));
}
@@ -1,85 +0,0 @@
const EXPLICIT_WAITING_PATH_FRESHNESS_MS = 24 * 60 * 60 * 1000;
type MaybeDate = Date | string | null | undefined;
export type ExplicitWaitingIssueInput = {
companyId: string;
id: string;
assigneeUserId?: string | null;
executionState?: Record<string, unknown> | null;
};
export type ExplicitWaitingInteractionInput = {
companyId: string;
issueId: string;
status?: string | null;
createdByUserId?: string | null;
createdAt?: MaybeDate;
};
export type ExplicitWaitingApprovalInput = {
companyId: string;
issueId: string;
status?: string | null;
requestedByUserId?: string | null;
linkedByUserId?: string | null;
createdAt?: MaybeDate;
updatedAt?: MaybeDate;
linkedAt?: MaybeDate;
};
function parseDate(value: MaybeDate) {
if (!value) return null;
if (value instanceof Date) return Number.isNaN(value.getTime()) ? null : value;
const parsed = new Date(value);
return Number.isNaN(parsed.getTime()) ? null : parsed;
}
function latestDate(...values: MaybeDate[]) {
return values
.map(parseDate)
.filter((value): value is Date => Boolean(value))
.sort((left, right) => right.getTime() - left.getTime())[0] ?? null;
}
function currentParticipantIsUser(executionState: Record<string, unknown> | null | undefined) {
const participant = executionState?.currentParticipant;
if (!participant || typeof participant !== "object") return false;
const value = participant as Record<string, unknown>;
return value.type === "user" && typeof value.userId === "string" && value.userId.trim().length > 0;
}
function issueHasHumanOwner(issue: ExplicitWaitingIssueInput) {
return Boolean(issue.assigneeUserId) || currentParticipantIsUser(issue.executionState);
}
function isFreshWaitingPath(createdAt: MaybeDate, now: Date) {
const created = parseDate(createdAt);
if (!created) return false;
return now.getTime() - created.getTime() <= EXPLICIT_WAITING_PATH_FRESHNESS_MS;
}
export function isLiveExplicitInteractionWaitingPath(
issue: ExplicitWaitingIssueInput,
interaction: ExplicitWaitingInteractionInput,
now = new Date(),
) {
if (interaction.companyId !== issue.companyId || interaction.issueId !== issue.id) return false;
if (interaction.status && interaction.status !== "pending") return false;
if (issueHasHumanOwner(issue)) return true;
if (interaction.createdByUserId) return true;
return isFreshWaitingPath(interaction.createdAt, now);
}
export function isLiveExplicitApprovalWaitingPath(
issue: ExplicitWaitingIssueInput,
approval: ExplicitWaitingApprovalInput,
now = new Date(),
) {
if (approval.companyId !== issue.companyId || approval.issueId !== issue.id) return false;
if (approval.status && approval.status !== "pending" && approval.status !== "revision_requested") return false;
if (issueHasHumanOwner(issue)) return true;
if (approval.requestedByUserId || approval.linkedByUserId) return true;
return isFreshWaitingPath(latestDate(approval.updatedAt, approval.linkedAt, approval.createdAt), now);
}
@@ -1,8 +1,4 @@
import { buildIssueGraphLivenessIncidentKey } from "./origins.js";
import {
isLiveExplicitApprovalWaitingPath,
isLiveExplicitInteractionWaitingPath,
} from "./explicit-waiting-paths.js";
export type IssueLivenessSeverity = "warning" | "critical";
@@ -56,12 +52,6 @@ export interface IssueLivenessWaitingPathInput {
companyId: string;
issueId: string;
status: string;
createdByUserId?: string | null;
requestedByUserId?: string | null;
linkedByUserId?: string | null;
createdAt?: Date | string | null;
updatedAt?: Date | string | null;
linkedAt?: Date | string | null;
}
export interface IssueLivenessDependencyPathEntry {
@@ -143,16 +133,11 @@ function hasActiveExecutionPath(
}
function hasWaitingPath(
issue: IssueLivenessIssueInput,
companyId: string,
issueId: string,
waitingPaths: IssueLivenessWaitingPathInput[],
kind: "interaction" | "approval",
now: Date,
) {
return waitingPaths.some((entry) =>
kind === "interaction"
? isLiveExplicitInteractionWaitingPath(issue, entry, now)
: isLiveExplicitApprovalWaitingPath(issue, entry, now)
);
return waitingPaths.some((entry) => entry.companyId === companyId && entry.issueId === issueId);
}
function readPrincipalAgentId(principal: unknown): string | null {
@@ -323,7 +308,6 @@ function finding(input: {
}
export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): IssueLivenessFinding[] {
const now = new Date();
const issuesById = new Map(input.issues.map((issue) => [issue.id, issue]));
const agentsById = new Map(input.agents.map((agent) => [agent.id, agent]));
const blockersByBlockedIssueId = new Map<string, IssueLivenessRelationInput[]>();
@@ -368,9 +352,9 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
function hasExplicitWaitingPath(issue: IssueLivenessIssueInput) {
return Boolean(issue.assigneeUserId) ||
hasActiveExecutionPath(issue.companyId, issue.id, activeRuns, queuedWakeRequests) ||
hasWaitingPath(issue, pendingInteractions, "interaction", now) ||
hasWaitingPath(issue, pendingApprovals, "approval", now) ||
openRecoveryIssues.some((entry) => entry.companyId === issue.companyId && entry.issueId === issue.id);
hasWaitingPath(issue.companyId, issue.id, pendingInteractions) ||
hasWaitingPath(issue.companyId, issue.id, pendingApprovals) ||
hasWaitingPath(issue.companyId, issue.id, openRecoveryIssues);
}
function reviewFinding(
@@ -8,14 +8,13 @@ export const RUN_LIVENESS_CONTINUATION_REASON = RECOVERY_REASON_KINDS.runLivenes
export const DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS = 2;
const ACTIONABLE_LIVENESS_STATES = new Set<RunLivenessState>(["plan_only", "empty_response"]);
const PROGRESS_WITH_FOLLOWUP_LIVENESS_STATES = new Set<RunLivenessState>(["advanced"]);
const CONTINUATION_ACTIVE_ISSUE_STATUSES = new Set(["todo", "in_progress"]);
// A prior adapter error should not permanently suppress bounded liveness
// continuations; the max-attempt/idempotency guards prevent unbounded retries.
const CONTINUATION_AGENT_STATUSES = new Set(["active", "idle", "running", "error"]);
const IDEMPOTENT_WAKE_STATUSES = ["queued", "deferred_issue_execution", "completed"];
type HeartbeatRunRow = Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "continuationAttempt">;
type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect;
type IssueRow = Pick<
typeof issues.$inferSelect,
"id" | "companyId" | "identifier" | "title" | "status" | "assigneeAgentId" | "executionState" | "projectId"
@@ -105,15 +104,7 @@ export function decideRunLivenessContinuation(input: {
} = input;
const maxAttempts = input.maxAttempts ?? DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS;
if (!livenessState) {
return { kind: "skip", reason: "liveness state is not actionable for continuation" };
}
const hasExplicitRunnableFollowup = Boolean(nextAction?.trim());
const livenessCanContinue =
ACTIONABLE_LIVENESS_STATES.has(livenessState) ||
PROGRESS_WITH_FOLLOWUP_LIVENESS_STATES.has(livenessState) && hasExplicitRunnableFollowup;
if (!livenessCanContinue) {
if (!livenessState || !ACTIONABLE_LIVENESS_STATES.has(livenessState)) {
return { kind: "skip", reason: "liveness state is not actionable for continuation" };
}
if (!issue) return { kind: "skip", reason: "issue not found" };
+14 -281
View File
@@ -6,7 +6,6 @@ import {
MIN_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS,
type IssueGraphLivenessAutoRecoveryPreview,
type IssueGraphLivenessAutoRecoveryPreviewItem,
type RunLivenessState,
} from "@paperclipai/shared";
import {
agents,
@@ -32,7 +31,6 @@ import { budgetService } from "../budgets.js";
import { instanceSettingsService } from "../instance-settings.js";
import { issueTreeControlService } from "../issue-tree-control.js";
import { issueService } from "../issues.js";
import { issueThreadInteractionService } from "../issue-thread-interactions.js";
import { getRunLogStore } from "../run-log-store.js";
import {
RECOVERY_ORIGIN_KINDS,
@@ -44,18 +42,7 @@ import {
classifyIssueGraphLiveness,
type IssueLivenessFinding,
} from "./issue-graph-liveness.js";
import {
isLiveExplicitApprovalWaitingPath,
isLiveExplicitInteractionWaitingPath,
} from "./explicit-waiting-paths.js";
import { isAutomaticRecoverySuppressedByPauseHold } from "./pause-hold-guard.js";
import {
RUN_LIVENESS_CONTINUATION_REASON,
buildRunLivenessContinuationIdempotencyKey,
decideRunLivenessContinuation,
findExistingRunLivenessContinuationWake,
readContinuationAttempt,
} from "./run-liveness-continuations.js";
const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;
const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const;
@@ -66,7 +53,6 @@ const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024;
const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.strandedIssueRecovery;
const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation;
const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
const EXPLICIT_WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions"];
type RecoveryWakeupOptions = {
source?: "timer" | "assignment" | "on_demand" | "automation";
@@ -86,21 +72,9 @@ type RecoveryWakeup = (
type LatestIssueRun = Pick<
typeof heartbeatRuns.$inferSelect,
| "id"
| "companyId"
| "agentId"
| "status"
| "error"
| "errorCode"
| "contextSnapshot"
| "livenessState"
| "livenessReason"
| "nextAction"
| "continuationAttempt"
"id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" | "livenessState"
> | null;
type PostRunIssueDisposition = "terminal" | "explicitly_waiting" | "explicitly_live" | "invalid";
type WatchdogDecisionActor =
| { type: "board"; userId?: string | null; runId?: string | null }
| { type: "agent"; agentId?: string | null; runId?: string | null }
@@ -214,14 +188,16 @@ function isUnsuccessfulTerminalIssueRun(latestRun: LatestIssueRun) {
);
}
function isProductiveProgressRun(latestRun: LatestIssueRun) {
function isSuccessfulInProgressContinuationRun(latestRun: LatestIssueRun) {
return latestRun?.status === "succeeded";
}
function isProductiveContinuationRun(latestRun: LatestIssueRun) {
return latestRun?.status === "succeeded" &&
(
latestRun.livenessState === "advanced" ||
(latestRun.livenessState === "advanced" ||
latestRun.livenessState === "completed" ||
latestRun.livenessState === "blocked" ||
latestRun.livenessState === "needs_followup"
);
latestRun.livenessState === "needs_followup");
}
function parseLivenessIncidentKey(incidentKey: string | null | undefined) {
@@ -330,16 +306,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
return db
.select({
id: heartbeatRuns.id,
companyId: heartbeatRuns.companyId,
agentId: heartbeatRuns.agentId,
status: heartbeatRuns.status,
error: heartbeatRuns.error,
errorCode: heartbeatRuns.errorCode,
contextSnapshot: heartbeatRuns.contextSnapshot,
livenessState: heartbeatRuns.livenessState,
livenessReason: heartbeatRuns.livenessReason,
nextAction: heartbeatRuns.nextAction,
continuationAttempt: heartbeatRuns.continuationAttempt,
})
.from(heartbeatRuns)
.where(
@@ -399,103 +371,6 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
.then((rows) => Boolean(rows[0]));
}
async function hasExplicitWaitingPath(
issue: Pick<
typeof issues.$inferSelect,
"companyId" | "id" | "status" | "assigneeUserId" | "executionState"
>,
) {
if (issue.status === "done" || issue.status === "cancelled") return true;
if (issue.status === "blocked" || issue.status === "in_review") return true;
if (issue.executionState) return true;
await issueThreadInteractionService(db).expireInvalidPendingRequestConfirmationsForIssue(
{ id: issue.id, companyId: issue.companyId },
{ agentId: null, userId: null },
);
const [interactionRows, approvalRows, recoveryIssue] = await Promise.all([
db
.select({
issueId: issueThreadInteractions.issueId,
companyId: issueThreadInteractions.companyId,
status: issueThreadInteractions.status,
createdByUserId: issueThreadInteractions.createdByUserId,
createdAt: issueThreadInteractions.createdAt,
})
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.companyId, issue.companyId),
eq(issueThreadInteractions.issueId, issue.id),
eq(issueThreadInteractions.status, "pending"),
inArray(issueThreadInteractions.kind, EXPLICIT_WAITING_INTERACTION_KINDS),
),
)
.limit(5),
db
.select({
issueId: issueApprovals.issueId,
companyId: issueApprovals.companyId,
status: approvals.status,
requestedByUserId: approvals.requestedByUserId,
linkedByUserId: issueApprovals.linkedByUserId,
createdAt: approvals.createdAt,
updatedAt: approvals.updatedAt,
linkedAt: issueApprovals.createdAt,
})
.from(issueApprovals)
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
.where(
and(
eq(issueApprovals.companyId, issue.companyId),
eq(issueApprovals.issueId, issue.id),
inArray(approvals.status, ["pending", "revision_requested"]),
),
)
.limit(5),
db
.select({ id: issues.id })
.from(issues)
.where(
and(
eq(issues.companyId, issue.companyId),
eq(issues.originKind, STRANDED_ISSUE_RECOVERY_ORIGIN_KIND),
eq(issues.originId, issue.id),
notInArray(issues.status, ["done", "cancelled"]),
),
)
.limit(1)
.then((rows) => rows[0] ?? null),
]);
return Boolean(
interactionRows.some((row) => isLiveExplicitInteractionWaitingPath(issue, row)) ||
approvalRows.some((row) => isLiveExplicitApprovalWaitingPath(issue, row)) ||
recoveryIssue,
);
}
async function classifyPostRunIssueDisposition(
issue: typeof issues.$inferSelect,
latestRun: LatestIssueRun,
): Promise<PostRunIssueDisposition> {
if (issue.status === "done" || issue.status === "cancelled") return "terminal";
if (await hasActiveExecutionPath(issue.companyId, issue.id)) return "explicitly_live";
if (await hasExplicitWaitingPath(issue)) return "explicitly_waiting";
if (latestRun && EXECUTION_PATH_HEARTBEAT_RUN_STATUSES.includes(
latestRun.status as (typeof EXECUTION_PATH_HEARTBEAT_RUN_STATUSES)[number],
)) {
return "explicitly_live";
}
return "invalid";
}
function didAutomaticContinuationFinishWithoutLivePath(latestRun: LatestIssueRun) {
if (latestRun?.status !== "succeeded") return false;
const latestContext = parseObject(latestRun.contextSnapshot);
return readNonEmptyString(latestContext.retryReason) === "issue_continuation_needed";
}
async function enqueueStrandedIssueRecovery(input: {
issueId: string;
agentId: string;
@@ -559,89 +434,6 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
});
}
async function continueOrEscalateSuccessfulStrandedRun(input: {
issue: typeof issues.$inferSelect;
agent: typeof agents.$inferSelect;
latestRun: NonNullable<LatestIssueRun>;
}) {
const { issue, agent, latestRun } = input;
const livenessState = latestRun.livenessState as RunLivenessState | null;
const nextAttempt = readContinuationAttempt(latestRun.continuationAttempt) + 1;
const idempotencyKey = buildRunLivenessContinuationIdempotencyKey({
issueId: issue.id,
sourceRunId: latestRun.id,
livenessState: livenessState ?? "failed",
nextAttempt,
});
const existingWake = await findExistingRunLivenessContinuationWake(db, {
companyId: issue.companyId,
idempotencyKey,
});
const budgetBlocked = await isInvocationBudgetBlocked(issue, agent.id);
if (budgetBlocked) return "skipped" as const;
const decision = decideRunLivenessContinuation({
run: latestRun,
issue,
agent,
livenessState,
livenessReason: latestRun.livenessReason,
nextAction: latestRun.nextAction,
budgetBlocked,
idempotentWakeExists: Boolean(existingWake),
});
if (decision.kind === "enqueue") {
const queued = await deps.enqueueWakeup(agent.id, {
source: "automation",
triggerDetail: "system",
reason: RUN_LIVENESS_CONTINUATION_REASON,
payload: decision.payload,
contextSnapshot: decision.contextSnapshot,
idempotencyKey: decision.idempotencyKey,
requestedByActorType: "system",
requestedByActorId: "recovery.reconcile_stranded_assigned_issue",
});
if (!queued) return "skipped" as const;
await db
.update(heartbeatRuns)
.set({
continuationAttempt: decision.nextAttempt,
updatedAt: new Date(),
})
.where(eq(heartbeatRuns.id, latestRun.id));
return "queued" as const;
}
if (decision.kind === "skip" && decision.reason === "continuation wake already exists for this source run and attempt") {
return "skipped" as const;
}
const continuationDetail = decision.kind === "exhausted"
? `\n\n${decision.comment}`
: [
"",
"",
"Paperclip did not queue an automatic continuation because the latest successful run did not leave a safe runnable action path.",
"",
`- Liveness state: \`${livenessState ?? "unknown"}\``,
`- Reason: ${decision.reason}`,
].join("\n");
const updated = await escalateStrandedAssignedIssue({
issue,
previousStatus: "in_progress",
latestRun,
comment:
"Paperclip observed a successful terminal run for this assigned `in_progress` issue, " +
"but no live execution, queued wake, waiting interaction, approval, or recovery owner now holds the next action." +
continuationDetail +
"\n\nMoving it to `blocked` so it is visible for intervention.",
});
return updated ? "escalated" as const : "skipped" as const;
}
async function isInvocationBudgetBlocked(issue: typeof issues.$inferSelect, agentId: string) {
const budgetBlock = await budgets.getInvocationBlock(issue.companyId, agentId, {
issueId: issue.id,
@@ -1824,11 +1616,6 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
continue;
}
if (await hasExplicitWaitingPath(issue)) {
result.skipped += 1;
continue;
}
const latestRun = await getLatestIssueRun(issue.companyId, issue.id);
if (isStrandedIssueRecoveryIssue(issue) && isUnsuccessfulTerminalIssueRun(latestRun)) {
const updated = await escalateStrandedRecoveryIssueInPlace({
@@ -1918,57 +1705,15 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
result.skipped += 1;
continue;
}
const postRunDisposition = await classifyPostRunIssueDisposition(issue, latestRun);
if (postRunDisposition !== "invalid") {
if (postRunDisposition === "explicitly_live" && latestRun?.status === "succeeded") {
if (isProductiveProgressRun(latestRun)) {
result.productiveContinuationObserved += 1;
} else {
result.successfulContinuationObserved += 1;
}
if (isSuccessfulInProgressContinuationRun(latestRun)) {
if (isProductiveContinuationRun(latestRun)) {
result.productiveContinuationObserved += 1;
} else {
result.successfulContinuationObserved += 1;
}
result.skipped += 1;
continue;
}
if (didAutomaticContinuationFinishWithoutLivePath(latestRun)) {
const updated = await escalateStrandedAssignedIssue({
issue,
previousStatus: "in_progress",
latestRun,
comment:
"Paperclip observed a successful automatic continuation for this assigned `in_progress` issue, " +
"but the run has finished and no live execution, queued wake, waiting interaction, approval, or recovery owner now holds the next action. " +
"Moving it to `blocked` so it is visible for intervention.",
});
if (updated) {
result.escalated += 1;
result.issueIds.push(issue.id);
} else {
result.skipped += 1;
}
continue;
}
if (latestRun?.status === "succeeded") {
const successfulRunRecovery = await continueOrEscalateSuccessfulStrandedRun({
issue,
agent,
latestRun,
});
if (successfulRunRecovery === "queued") {
result.continuationRequeued += 1;
result.issueIds.push(issue.id);
} else if (successfulRunRecovery === "escalated") {
result.escalated += 1;
result.issueIds.push(issue.id);
} else {
result.skipped += 1;
}
continue;
}
if (didAutomaticRecoveryFail(latestRun, "issue_continuation_needed")) {
const failureSummary = summarizeRunFailureForIssueComment(latestRun);
const updated = await escalateStrandedAssignedIssue({
@@ -2111,26 +1856,14 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
companyId: issueThreadInteractions.companyId,
issueId: issueThreadInteractions.issueId,
status: issueThreadInteractions.status,
createdByUserId: issueThreadInteractions.createdByUserId,
createdAt: issueThreadInteractions.createdAt,
})
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.status, "pending"),
inArray(issueThreadInteractions.kind, EXPLICIT_WAITING_INTERACTION_KINDS),
),
),
.where(eq(issueThreadInteractions.status, "pending")),
db
.select({
companyId: issueApprovals.companyId,
issueId: issueApprovals.issueId,
status: approvals.status,
requestedByUserId: approvals.requestedByUserId,
linkedByUserId: issueApprovals.linkedByUserId,
createdAt: approvals.createdAt,
updatedAt: approvals.updatedAt,
linkedAt: issueApprovals.createdAt,
})
.from(issueApprovals)
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
+4 -16
View File
@@ -317,10 +317,6 @@ export function classifyRunLiveness(input: RunLivenessClassificationInput): RunL
return output("completed", `Issue is ${issueStatus}`);
}
if (issueStatus === "in_review") {
return output("needs_followup", "Issue is in review and waiting on an explicit review path", nextAction);
}
if (declaredBlocker(input)) {
return output("blocked", issueStatus === "blocked" ? "Issue status is blocked" : "Run output declared a concrete blocker", nextAction);
}
@@ -329,20 +325,12 @@ export function classifyRunLiveness(input: RunLivenessClassificationInput): RunL
return output("empty_response", "Run succeeded without useful output or concrete action evidence");
}
if (planExempt && usefulOutput) {
return output("advanced", "Planning/document task produced useful output and is exempt from plan-only classification");
if (concreteEvidence) {
return output("advanced", `Run produced concrete action evidence: ${evidenceReason(evidence)}`);
}
if (concreteEvidence) {
const reason = `Run produced concrete action evidence: ${evidenceReason(evidence)}`;
if (nextAction && actionability === "runnable") {
return output("advanced", `${reason}; runnable next action remains`, nextAction);
}
if (nextAction && (actionability === "manager_review" || actionability === "unknown")) {
const disposition = actionability === "manager_review" ? "manager review" : "ambiguous review";
return output("needs_followup", `${reason}; next action requires ${disposition}`, nextAction);
}
return output("advanced", reason);
if (planExempt && usefulOutput) {
return output("advanced", "Planning/document task produced useful output and is exempt from plan-only classification");
}
if (looksLikePlanningOnly(input) || nextAction) {