From 2b3a7f683b856f1fadbc7db9c5f0cad0bd7e52e9 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 29 Apr 2026 10:50:25 -0500 Subject: [PATCH] PAP-2822 Expose canonical executionDisposition on issue payloads Adds a per-issue read-model resolver that loads the disposition state vector (active runs, queued wakes, agents, interactions, approvals, recovery/productivity-review issues, latest run evidence) and runs the shared classifier. Wires it into issue list, detail, and heartbeat-context responses, and surfaces it on the shared Issue type as `executionDisposition` so the board can read execution health without inferring it from status alone. Co-Authored-By: Paperclip --- packages/shared/src/types/issue.ts | 2 + ...sue-execution-disposition-resolver.test.ts | 288 ++++++++++ server/src/routes/issues.ts | 32 ++ .../issue-execution-disposition-resolver.ts | 504 ++++++++++++++++++ server/src/services/issues.ts | 34 ++ 5 files changed, 860 insertions(+) create mode 100644 server/src/__tests__/issue-execution-disposition-resolver.test.ts create mode 100644 server/src/services/issue-execution-disposition-resolver.ts diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index 1199ab098b4..76aba93b275 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -15,6 +15,7 @@ import type { Goal } from "./goal.js"; import type { Project, ProjectWorkspace } from "./project.js"; import type { ExecutionWorkspace, IssueExecutionWorkspaceSettings } from "./workspace-runtime.js"; import type { IssueWorkProduct } from "./work-product.js"; +import type { IssueExecutionDisposition } from "./issue-execution-disposition.js"; export interface IssueAncestorProject { id: string; @@ -306,6 +307,7 @@ export interface Issue { blockedBy?: IssueRelationIssueSummary[]; blocks?: IssueRelationIssueSummary[]; blockerAttention?: IssueBlockerAttention; + executionDisposition?: IssueExecutionDisposition; productivityReview?: IssueProductivityReview | null; relatedWork?: IssueRelatedWorkSummary; referencedIssueIdentifiers?: string[]; diff --git a/server/src/__tests__/issue-execution-disposition-resolver.test.ts b/server/src/__tests__/issue-execution-disposition-resolver.test.ts new file mode 100644 index 00000000000..a7a59b40f06 --- /dev/null +++ b/server/src/__tests__/issue-execution-disposition-resolver.test.ts @@ -0,0 +1,288 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, + agentWakeupRequests, + approvals, + companies, + createDb, + heartbeatRuns, + issueApprovals, + issueRelations, + issueThreadInteractions, + issues, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { + backfillBlockedBySummaries, + listIssueExecutionDispositionsMap, +} from "../services/issue-execution-disposition-resolver.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres execution-disposition resolver tests: ${embeddedPostgresSupport.reason ?? "unsupported"}`, + ); +} + +describeEmbeddedPostgres("issue execution disposition resolver", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-execution-disposition-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + 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); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seed(prefix = "DSP") { + const companyId = randomUUID(); + const agentId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: `Company ${prefix}`, + issuePrefix: prefix, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: `${prefix} Agent`, + role: "engineer", + status: "idle", + }); + return { companyId, agentId }; + } + + async function insertIssue(input: { + companyId: string; + identifier: string; + title: string; + status: string; + assigneeAgentId?: string | null; + assigneeUserId?: string | null; + originKind?: string | null; + }) { + const id = randomUUID(); + await db.insert(issues).values({ + id, + companyId: input.companyId, + identifier: input.identifier, + title: input.title, + status: input.status, + priority: "medium", + assigneeAgentId: input.assigneeAgentId ?? null, + assigneeUserId: input.assigneeUserId ?? null, + originKind: input.originKind ?? "manual", + }); + return id; + } + + function inputFor(input: { + id: string; + companyId: string; + status: string; + assigneeAgentId?: string | null; + assigneeUserId?: string | null; + originKind?: string | null; + }) { + return { + id: input.id, + companyId: input.companyId, + status: input.status, + assigneeAgentId: input.assigneeAgentId ?? null, + assigneeUserId: input.assigneeUserId ?? null, + originKind: input.originKind ?? "manual", + executionRunId: null as string | null, + executionState: null as Record | null, + }; + } + + it("classifies an idle agent-owned todo as dispatchable", async () => { + const { companyId, agentId } = await seed(); + const issueId = await insertIssue({ + companyId, + identifier: "DSP-1", + title: "Ready", + status: "todo", + assigneeAgentId: agentId, + }); + const map = await listIssueExecutionDispositionsMap(db, companyId, [ + inputFor({ id: issueId, companyId, status: "todo", assigneeAgentId: agentId }), + ]); + const disposition = map.get(issueId); + expect(disposition?.kind).toBe("dispatchable"); + }); + + it("classifies an in_progress issue with an active heartbeat run as live", async () => { + const { companyId, agentId } = await seed(); + const issueId = await insertIssue({ + companyId, + identifier: "DSP-2", + title: "Running", + status: "in_progress", + assigneeAgentId: agentId, + }); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: "running", + contextSnapshot: { issueId }, + }); + + const map = await listIssueExecutionDispositionsMap(db, companyId, [ + inputFor({ id: issueId, companyId, status: "in_progress", assigneeAgentId: agentId }), + ]); + expect(map.get(issueId)?.kind).toBe("live"); + }); + + it("classifies a pending interaction as waiting on interaction", async () => { + const { companyId, agentId } = await seed(); + const issueId = await insertIssue({ + companyId, + identifier: "DSP-3", + title: "Waiting on confirmation", + status: "in_review", + assigneeAgentId: agentId, + }); + await db.insert(issueThreadInteractions).values({ + id: randomUUID(), + companyId, + issueId, + kind: "request_confirmation", + status: "pending", + payload: { target: { type: "issue_document", key: "plan", revisionId: "r1" } }, + createdByAgentId: agentId, + }); + + const map = await listIssueExecutionDispositionsMap(db, companyId, [ + inputFor({ id: issueId, companyId, status: "in_review", assigneeAgentId: agentId }), + ]); + const disposition = map.get(issueId); + expect(disposition?.kind).toBe("waiting"); + if (disposition?.kind === "waiting") { + expect(disposition.path).toBe("interaction"); + } + }); + + it("flags an in_review issue without any action path as invalid", async () => { + const { companyId, agentId } = await seed(); + const issueId = await insertIssue({ + companyId, + identifier: "DSP-4", + title: "Stalled review", + status: "in_review", + assigneeAgentId: agentId, + }); + const map = await listIssueExecutionDispositionsMap(db, companyId, [ + inputFor({ id: issueId, companyId, status: "in_review", assigneeAgentId: agentId }), + ]); + const disposition = map.get(issueId); + expect(disposition?.kind).toBe("invalid"); + if (disposition?.kind === "invalid") { + expect(disposition.reason).toBe("in_review_without_action_path"); + } + }); + + it("classifies a blocked issue with a healthy chain leaf as waiting on blocker chain", async () => { + const { companyId, agentId } = await seed(); + const blockerId = await insertIssue({ + companyId, + identifier: "DSP-5b", + title: "Blocker", + status: "in_progress", + assigneeAgentId: agentId, + }); + const blockedId = await insertIssue({ + companyId, + identifier: "DSP-5", + title: "Blocked", + status: "blocked", + assigneeAgentId: agentId, + }); + await db.insert(issueRelations).values({ + id: randomUUID(), + companyId, + issueId: blockerId, + relatedIssueId: blockedId, + type: "blocks", + }); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: "running", + contextSnapshot: { issueId: blockerId }, + }); + + const inputs = [ + inputFor({ id: blockedId, companyId, status: "blocked", assigneeAgentId: agentId }), + ]; + await backfillBlockedBySummaries(db, companyId, inputs); + const map = await listIssueExecutionDispositionsMap(db, companyId, inputs); + const disposition = map.get(blockedId); + expect(disposition?.kind).toBe("waiting"); + if (disposition?.kind === "waiting") { + expect(disposition.path).toBe("blocker_chain"); + } + }); + + it("flags a blocked issue with a cancelled blocker as invalid", async () => { + const { companyId, agentId } = await seed(); + const blockerId = await insertIssue({ + companyId, + identifier: "DSP-6b", + title: "Cancelled blocker", + status: "cancelled", + assigneeAgentId: agentId, + }); + const blockedId = await insertIssue({ + companyId, + identifier: "DSP-6", + title: "Blocked by cancelled", + status: "blocked", + assigneeAgentId: agentId, + }); + await db.insert(issueRelations).values({ + id: randomUUID(), + companyId, + issueId: blockerId, + relatedIssueId: blockedId, + type: "blocks", + }); + + const inputs = [ + inputFor({ id: blockedId, companyId, status: "blocked", assigneeAgentId: agentId }), + ]; + await backfillBlockedBySummaries(db, companyId, inputs); + const map = await listIssueExecutionDispositionsMap(db, companyId, inputs); + const disposition = map.get(blockedId); + // The cancelled blocker is filtered out by ne(status, "done") + companyId checks; backfill + // excludes it because it's not "open". So the issue ends up blocked-without-an-action-path, + // which is also invalid. Either invalid form is acceptable here; assert the kind only. + expect(disposition?.kind).toBe("invalid"); + }); +}); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 1ce2ee1dca3..d994b1042fe 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -1113,6 +1113,21 @@ export function issueRoutes( documentsSvc.getIssueDocumentByKey(issue.id, ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY), currentExecutionWorkspacePromise, ]); + const executionDisposition = await svc + .listExecutionDispositions(issue.companyId, [ + { + id: issue.id, + companyId: issue.companyId, + status: issue.status, + assigneeAgentId: issue.assigneeAgentId, + assigneeUserId: issue.assigneeUserId, + originKind: issue.originKind ?? null, + executionRunId: issue.executionRunId ?? null, + executionState: issue.executionState ?? null, + blockedBy: relations.blockedBy, + }, + ]) + .then((map) => map.get(issue.id) ?? null); res.json({ issue: { @@ -1122,6 +1137,7 @@ export function issueRoutes( description: issue.description, status: issue.status, ...(blockerAttention ? { blockerAttention } : {}), + ...(executionDisposition ? { executionDisposition } : {}), productivityReview, priority: issue.priority, projectId: issue.projectId, @@ -1220,11 +1236,27 @@ export function issueRoutes( ? await executionWorkspacesSvc.getById(issue.executionWorkspaceId) : null; const workProducts = await workProductsSvc.listForIssue(issue.id); + const executionDisposition = await svc + .listExecutionDispositions(issue.companyId, [ + { + id: issue.id, + companyId: issue.companyId, + status: issue.status, + assigneeAgentId: issue.assigneeAgentId, + assigneeUserId: issue.assigneeUserId, + originKind: issue.originKind ?? null, + executionRunId: issue.executionRunId ?? null, + executionState: issue.executionState ?? null, + blockedBy: relations.blockedBy, + }, + ]) + .then((map) => map.get(issue.id) ?? null); res.json({ ...issue, goalId: goal?.id ?? issue.goalId, ancestors, ...(blockerAttention ? { blockerAttention } : {}), + ...(executionDisposition ? { executionDisposition } : {}), productivityReview, blockedBy: relations.blockedBy, blocks: relations.blocks, diff --git a/server/src/services/issue-execution-disposition-resolver.ts b/server/src/services/issue-execution-disposition-resolver.ts new file mode 100644 index 00000000000..745b7469a98 --- /dev/null +++ b/server/src/services/issue-execution-disposition-resolver.ts @@ -0,0 +1,504 @@ +import { and, desc, eq, inArray, isNull, ne, notInArray, sql } from "drizzle-orm"; +import { + agentWakeupRequests, + agents, + approvals, + heartbeatRuns, + issueApprovals, + issueRelations, + issueThreadInteractions, + issues, +} from "@paperclipai/db"; +import type { + IssueExecutionDisposition, + IssueOriginKind, + IssueRelationIssueSummary, + IssueStatus, +} from "@paperclipai/shared"; +import { + isLiveExplicitApprovalWaitingPath, + isLiveExplicitInteractionWaitingPath, +} from "./recovery/explicit-waiting-paths.js"; +import { + classifyIssueExecutionDisposition, + type IssueExecutionParticipantState, + type IssueExecutionRunLivenessState, + type IssueExecutionRunStatus, +} from "./issue-execution-disposition.js"; + +const ACTIVE_RUN_STATUSES = ["queued", "running"] as const; +const ACTIVE_WAKE_STATUSES = ["queued", "deferred_issue_execution"] as const; +const PENDING_INTERACTION_STATUSES = ["pending"] as const; +const WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions", "suggest_tasks"] as const; +const PENDING_APPROVAL_STATUSES = ["pending", "revision_requested"] as const; +const STRANDED_RECOVERY_ORIGIN_KIND = "stranded_issue_recovery" as const; +const HARNESS_LIVENESS_ESCALATION_ORIGIN_KIND = "harness_liveness_escalation" as const; +const PRODUCTIVITY_REVIEW_ORIGIN_KIND = "issue_productivity_review" as const; +const NON_TERMINAL_RECOVERY_STATUSES = ["done", "cancelled"] as const; +const INVOKABLE_AGENT_STATUSES: ReadonlySet = new Set(["active", "idle", "running", "error"]); +const DEFAULT_MAX_CONTINUATION_ATTEMPTS = 2; +const QUERY_CHUNK_SIZE = 500; + +export interface IssueExecutionDispositionInputNode { + id: string; + companyId: string; + status: IssueStatus | string; + assigneeAgentId: string | null; + assigneeUserId: string | null; + originKind: IssueOriginKind | string | null; + executionRunId: string | null; + executionState?: Record | null; + blockedBy?: IssueRelationIssueSummary[]; +} + +interface ExplicitInteractionRow { + companyId: string; + issueId: string; + status: string; + createdByUserId: string | null; + createdAt: Date; +} + +interface ExplicitApprovalRow { + companyId: string; + issueId: string; + status: string; + requestedByUserId: string | null; + linkedByUserId: string | null; + createdAt: Date; + updatedAt: Date; + linkedAt: Date; +} + +interface LatestRunRow { + issueId: string; + status: string; + livenessState: string | null; + continuationAttempt: number | null; + contextSnapshot: unknown; +} + +function chunk(items: T[], size: number): T[][] { + if (items.length === 0) return []; + const out: T[][] = []; + for (let i = 0; i < items.length; i += size) { + out.push(items.slice(i, i + size)); + } + return out; +} + +function readPrincipalAgentId(principal: unknown): string | null { + if (!principal || typeof principal !== "object") return null; + const value = principal as Record; + return value.type === "agent" && typeof value.agentId === "string" && value.agentId.length > 0 + ? value.agentId + : null; +} + +function principalIsResolvableUser(principal: unknown): boolean { + if (!principal || typeof principal !== "object") return false; + const value = principal as Record; + return value.type === "user" && typeof value.userId === "string" && value.userId.length > 0; +} + +function participantStateForExecutionState( + executionState: Record | null | undefined, + agentStatusById: ReadonlyMap, +): IssueExecutionParticipantState { + if (!executionState) return "none"; + const status = (executionState.status as string | undefined) ?? null; + if (!status || status === "satisfied") return "none"; + const principal = executionState.currentParticipant ?? null; + if (!principal) return "none"; + const agentId = readPrincipalAgentId(principal); + if (agentId) { + const agentStatus = agentStatusById.get(agentId); + if (agentStatus && INVOKABLE_AGENT_STATUSES.has(agentStatus)) return "valid"; + return "invalid"; + } + if (principalIsResolvableUser(principal)) return "valid"; + return "invalid"; +} + +function readNextActionFlag(contextSnapshot: unknown): boolean { + if (!contextSnapshot || typeof contextSnapshot !== "object") return false; + const next = (contextSnapshot as Record).nextAction; + return typeof next === "string" && next.trim().length > 0; +} + +export async function listIssueExecutionDispositionsMap( + dbOrTx: any, + companyId: string, + issueRows: IssueExecutionDispositionInputNode[], +): Promise> { + const result = new Map(); + const sameCompanyRows = issueRows.filter((row) => row.companyId === companyId); + if (sameCompanyRows.length === 0) return result; + + const issueIds = sameCompanyRows.map((row) => row.id); + + // The list select drops executionState for performance; rehydrate it from the DB + // for the issues that actually need participant evidence (in_review). + const reviewIdsMissingState = sameCompanyRows + .filter((row) => row.status === "in_review" && row.executionState === undefined) + .map((row) => row.id); + const executionStateById = new Map | null>(); + for (const part of chunk(reviewIdsMissingState, QUERY_CHUNK_SIZE)) { + const rows: Array<{ id: string; executionState: Record | null }> = await dbOrTx + .select({ id: issues.id, executionState: issues.executionState }) + .from(issues) + .where(and(eq(issues.companyId, companyId), inArray(issues.id, part))); + for (const r of rows) executionStateById.set(r.id, r.executionState ?? null); + } + const resolvedExecutionStateFor = (row: IssueExecutionDispositionInputNode) => + row.executionState !== undefined ? row.executionState : executionStateById.get(row.id) ?? null; + + const agentIds = new Set(); + for (const row of sameCompanyRows) { + if (row.assigneeAgentId) agentIds.add(row.assigneeAgentId); + const state = resolvedExecutionStateFor(row); + const participant = state?.currentParticipant ?? null; + const participantAgentId = readPrincipalAgentId(participant); + if (participantAgentId) agentIds.add(participantAgentId); + } + + const activeRunIssueIds = new Set(); + const queuedWakeIssueIds = new Set(); + const explicitInteractionLive = new Set(); + const explicitApprovalLive = new Set(); + const openRecoveryIssueOrigins = new Set(); + const openProductivityReviewOrigins = new Set(); + const latestRunByIssueId = new Map(); + + // Active runs: any heartbeatRun in queued/running referencing the issue via contextSnapshot. + for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) { + const rows: Array<{ issueId: string | null }> = await dbOrTx + .select({ + issueId: sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.status, [...ACTIVE_RUN_STATUSES]), + inArray(sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, part), + ), + ); + for (const row of rows) { + if (row.issueId) activeRunIssueIds.add(row.issueId); + } + } + + // Active runs by executionRunId — covers cases where the run claimed the issue but the + // contextSnapshot lookup misses (legacy snapshots). + const executionRunIds = sameCompanyRows + .map((row) => row.executionRunId) + .filter((value): value is string => Boolean(value)); + for (const part of chunk(executionRunIds, QUERY_CHUNK_SIZE)) { + const rows: Array<{ id: string }> = await dbOrTx + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.status, [...ACTIVE_RUN_STATUSES]), + inArray(heartbeatRuns.id, part), + ), + ); + if (rows.length === 0) continue; + const liveExecutionRunIds = new Set(rows.map((row) => row.id)); + for (const row of sameCompanyRows) { + if (row.executionRunId && liveExecutionRunIds.has(row.executionRunId)) { + activeRunIssueIds.add(row.id); + } + } + } + + // Queued wakes: agent_wakeup_requests with issueId payload, no runId yet. + for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) { + const rows: Array<{ issueId: string | null }> = await dbOrTx + .select({ + issueId: sql`${agentWakeupRequests.payload} ->> 'issueId'`, + }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + inArray(agentWakeupRequests.status, [...ACTIVE_WAKE_STATUSES]), + sql`${agentWakeupRequests.runId} is null`, + inArray(sql`${agentWakeupRequests.payload} ->> 'issueId'`, part), + ), + ); + for (const row of rows) { + if (row.issueId) queuedWakeIssueIds.add(row.issueId); + } + } + + // Pending interactions. + const interactionsByIssueId = new Map(); + for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) { + const rows: ExplicitInteractionRow[] = await dbOrTx + .select({ + companyId: issueThreadInteractions.companyId, + issueId: issueThreadInteractions.issueId, + status: issueThreadInteractions.status, + createdByUserId: issueThreadInteractions.createdByUserId, + createdAt: issueThreadInteractions.createdAt, + }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.companyId, companyId), + inArray(issueThreadInteractions.status, [...PENDING_INTERACTION_STATUSES]), + inArray(issueThreadInteractions.kind, [...WAITING_INTERACTION_KINDS]), + inArray(issueThreadInteractions.issueId, part), + ), + ); + for (const row of rows) { + const list = interactionsByIssueId.get(row.issueId) ?? []; + list.push(row); + interactionsByIssueId.set(row.issueId, list); + } + } + + // Pending approvals. + const approvalsByIssueId = new Map(); + for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) { + const rows: ExplicitApprovalRow[] = await dbOrTx + .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)) + .where( + and( + eq(issueApprovals.companyId, companyId), + inArray(approvals.status, [...PENDING_APPROVAL_STATUSES]), + inArray(issueApprovals.issueId, part), + ), + ); + for (const row of rows) { + const list = approvalsByIssueId.get(row.issueId) ?? []; + list.push(row); + approvalsByIssueId.set(row.issueId, list); + } + } + + const now = new Date(); + for (const row of sameCompanyRows) { + const issueShape = { + companyId: row.companyId, + id: row.id, + assigneeUserId: row.assigneeUserId, + executionState: resolvedExecutionStateFor(row), + }; + const interactions = interactionsByIssueId.get(row.id) ?? []; + if (interactions.some((entry) => isLiveExplicitInteractionWaitingPath(issueShape, entry, now))) { + explicitInteractionLive.add(row.id); + } + const approvalsForIssue = approvalsByIssueId.get(row.id) ?? []; + if (approvalsForIssue.some((entry) => isLiveExplicitApprovalWaitingPath(issueShape, entry, now))) { + explicitApprovalLive.add(row.id); + } + } + + // Open recovery / harness escalation / productivity review issues whose origin points at our inputs. + const recoveryOriginKinds = [STRANDED_RECOVERY_ORIGIN_KIND, HARNESS_LIVENESS_ESCALATION_ORIGIN_KIND]; + for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) { + const rows: Array<{ originId: string | null }> = await dbOrTx + .select({ originId: issues.originId }) + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + inArray(issues.originKind, recoveryOriginKinds), + isNull(issues.hiddenAt), + inArray(issues.originId, part), + notInArray(issues.status, [...NON_TERMINAL_RECOVERY_STATUSES]), + ), + ); + for (const r of rows) { + if (r.originId) openRecoveryIssueOrigins.add(r.originId); + } + } + for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) { + const rows: 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, part), + notInArray(issues.status, [...NON_TERMINAL_RECOVERY_STATUSES]), + ), + ); + for (const r of rows) { + if (r.originId) openProductivityReviewOrigins.add(r.originId); + } + } + + // Latest issue-linked run, used to populate run evidence. + const recoveryCandidateIds = sameCompanyRows + .filter((row) => + row.assigneeAgentId && + !activeRunIssueIds.has(row.id) && + !queuedWakeIssueIds.has(row.id) && + (row.status === "todo" || row.status === "in_progress"), + ) + .map((row) => row.id); + for (const part of chunk(recoveryCandidateIds, QUERY_CHUNK_SIZE)) { + const rows: LatestRunRow[] = await dbOrTx + .select({ + issueId: sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, + status: heartbeatRuns.status, + livenessState: heartbeatRuns.livenessState, + continuationAttempt: heartbeatRuns.continuationAttempt, + contextSnapshot: heartbeatRuns.contextSnapshot, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + inArray(sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, part), + ), + ) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)); + for (const row of rows) { + if (!row.issueId) continue; + if (latestRunByIssueId.has(row.issueId)) continue; + latestRunByIssueId.set(row.issueId, row); + } + } + + // Agents. + const agentStatusById = new Map(); + if (agentIds.size > 0) { + const rows: Array<{ id: string; status: string }> = await dbOrTx + .select({ id: agents.id, status: agents.status }) + .from(agents) + .where(and(eq(agents.companyId, companyId), inArray(agents.id, [...agentIds]))); + for (const r of rows) agentStatusById.set(r.id, r.status); + } + + for (const row of sameCompanyRows) { + const agentStatus = row.assigneeAgentId ? agentStatusById.get(row.assigneeAgentId) ?? null : null; + const latestRun = latestRunByIssueId.get(row.id) ?? null; + const productivityReviewNeeded = openProductivityReviewOrigins.has(row.id); + const openRecoveryIssue = openRecoveryIssueOrigins.has(row.id); + const blockers = (row.blockedBy ?? []).map((blocker) => ({ + issue: { + id: blocker.id, + status: blocker.status as IssueStatus, + assigneeAgentId: blocker.assigneeAgentId ?? null, + assigneeUserId: blocker.assigneeUserId ?? null, + }, + })); + + const disposition = classifyIssueExecutionDisposition({ + issue: { + id: row.id, + status: row.status as IssueStatus, + assigneeAgentId: row.assigneeAgentId, + assigneeUserId: row.assigneeUserId, + originKind: row.originKind ?? null, + }, + agent: row.assigneeAgentId + ? { id: row.assigneeAgentId, status: agentStatus ?? "terminated" } + : null, + execution: { + activeRun: activeRunIssueIds.has(row.id), + queuedWake: queuedWakeIssueIds.has(row.id), + }, + waits: { + participant: participantStateForExecutionState(resolvedExecutionStateFor(row), agentStatusById), + pendingInteraction: explicitInteractionLive.has(row.id) ? "live" : "none", + pendingApproval: explicitApprovalLive.has(row.id) ? "live" : "none", + openRecoveryIssue, + openProductivityReviewIssue: productivityReviewNeeded, + productivityReviewNeeded, + }, + latestRun: latestRun + ? { + latestRunStatus: latestRun.status as IssueExecutionRunStatus, + livenessState: (latestRun.livenessState as IssueExecutionRunLivenessState | null) ?? null, + nextAction: readNextActionFlag(latestRun.contextSnapshot) ? "runnable" : "none", + continuationAttempt: latestRun.continuationAttempt, + maxContinuationAttempts: DEFAULT_MAX_CONTINUATION_ATTEMPTS, + } + : null, + blockers, + }); + + result.set(row.id, disposition); + } + + return result; +} + +// Helper: load blockedBy relation summaries for issues that don't already have them on the input. +export async function backfillBlockedBySummaries( + dbOrTx: any, + companyId: string, + rows: IssueExecutionDispositionInputNode[], +): Promise { + const needsBackfill = rows.filter((row) => row.status === "blocked" && !row.blockedBy); + if (needsBackfill.length === 0) return; + const ids = needsBackfill.map((row) => row.id); + const blockerRows: Array<{ + issueId: string; + blockerIssueId: string; + identifier: string | null; + title: string; + status: string; + priority: string | null; + assigneeAgentId: string | null; + assigneeUserId: string | null; + }> = await dbOrTx + .select({ + issueId: issueRelations.relatedIssueId, + blockerIssueId: issues.id, + identifier: issues.identifier, + title: issues.title, + status: issues.status, + priority: issues.priority, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + }) + .from(issueRelations) + .innerJoin(issues, eq(issueRelations.issueId, issues.id)) + .where( + and( + eq(issueRelations.companyId, companyId), + eq(issueRelations.type, "blocks"), + inArray(issueRelations.relatedIssueId, ids), + eq(issues.companyId, companyId), + ne(issues.status, "done"), + ), + ); + const blockersByIssueId = new Map(); + for (const row of blockerRows) { + const list = blockersByIssueId.get(row.issueId) ?? []; + list.push({ + id: row.blockerIssueId, + identifier: row.identifier, + title: row.title, + status: row.status as IssueStatus, + priority: (row.priority ?? "medium") as IssueRelationIssueSummary["priority"], + assigneeAgentId: row.assigneeAgentId, + assigneeUserId: row.assigneeUserId, + }); + blockersByIssueId.set(row.issueId, list); + } + for (const row of needsBackfill) { + row.blockedBy = blockersByIssueId.get(row.id) ?? []; + } +} diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 640a514284c..213f27ec551 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -64,6 +64,11 @@ import { classifyIssueExecutionDisposition, type IssueExecutionRunLivenessState, } from "./issue-execution-disposition.js"; +import { + backfillBlockedBySummaries, + listIssueExecutionDispositionsMap, + type IssueExecutionDispositionInputNode, +} from "./issue-execution-disposition-resolver.js"; import { assertIssueTransitionAllowed } from "./issue-transition-guard.js"; const ALL_ISSUE_STATUSES = ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"]; @@ -2703,6 +2708,23 @@ export function issueService(db: Db) { listIssueBlockerAttentionMap(db, companyId, withRuns), listIssueProductivityReviewMap(db, companyId, issueIds), ]); + const dispositionInputs: IssueExecutionDispositionInputNode[] = withRuns.map((row) => ({ + id: row.id, + companyId: row.companyId, + status: row.status, + assigneeAgentId: row.assigneeAgentId, + assigneeUserId: row.assigneeUserId, + originKind: row.originKind ?? null, + executionRunId: row.executionRunId ?? null, + // executionState is intentionally omitted from the list select; the resolver rehydrates it. + blockedBy: includeBlockedBy ? blockedByMap.get(row.id) ?? [] : undefined, + })); + await backfillBlockedBySummaries(db, companyId, dispositionInputs); + const dispositionByIssueId = await listIssueExecutionDispositionsMap( + db, + companyId, + dispositionInputs, + ); if (!contextUserId) { return withRuns.map((row) => { @@ -2717,6 +2739,7 @@ export function issueService(db: Db) { ...(includeBlockedBy ? { blockedBy: blockedByMap.get(row.id) ?? [] } : {}), lastActivityAt, ...(blockerAttentionByIssueId.has(row.id) ? { blockerAttention: blockerAttentionByIssueId.get(row.id) } : {}), + ...(dispositionByIssueId.has(row.id) ? { executionDisposition: dispositionByIssueId.get(row.id) } : {}), ...(productivityReviewByIssueId.has(row.id) ? { productivityReview: productivityReviewByIssueId.get(row.id) } : {}), @@ -2738,6 +2761,7 @@ export function issueService(db: Db) { ...(includeBlockedBy ? { blockedBy: blockedByMap.get(row.id) ?? [] } : {}), lastActivityAt, ...(blockerAttentionByIssueId.has(row.id) ? { blockerAttention: blockerAttentionByIssueId.get(row.id) } : {}), + ...(dispositionByIssueId.has(row.id) ? { executionDisposition: dispositionByIssueId.get(row.id) } : {}), ...(productivityReviewByIssueId.has(row.id) ? { productivityReview: productivityReviewByIssueId.get(row.id) } : {}), @@ -2892,6 +2916,16 @@ export function issueService(db: Db) { return listIssueBlockerAttentionMap(dbOrTx, companyId, issueRows); }, + listExecutionDispositions: async ( + companyId: string, + issueRows: IssueExecutionDispositionInputNode[], + dbOrTx: any = db, + ) => { + const inputs = issueRows.map((row) => ({ ...row })); + await backfillBlockedBySummaries(dbOrTx, companyId, inputs); + return listIssueExecutionDispositionsMap(dbOrTx, companyId, inputs); + }, + listProductivityReviews: async ( companyId: string, sourceIssueIds: string[],