diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e12aa309131..e8938daa137 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4112,6 +4112,24 @@ An absent, changed, pending, or agent-resolved response grants no such exception That bounded response remains a continuation after a workspace or AI-account wait, so final dispatch still verifies it. Finishing a dependency during preparation does not turn the admitted response into an attempt to check out the task. + +A person's answer to the agent's own question also continues a task that the same +agent blocked while it waited for that answer. The agent can report such a wait as +a blocker, for example after a later comment from the person moved the question +into history. When the status committer applies an agent-reported, board-owned +blocker (`current_track_blocker_waiting` or `task_wide_blocker_bound`), it records +in the same transaction the agent's own pending, undismissed ordinary questions on +the task as `awaitingInteractionIds` in the decision. Dispatch accepts the Blocked +task only when the answered interaction is in that list, was created by the +assignee and resolved by a person, the decision is still the task's +`lastStatusDecisionId`, and the task's `statusVersion` still equals the decision's +`projectedStatusVersion`. The checkout repeats the decision and version comparison +in its guarded update, so a person who re-blocks the task in between keeps the new +block. The continuation then checks the task out to `in_progress` and clears its +blocker fields. If the blocker was about something else, the agent receives the +answer and reports its blocker again. A task blocked by a person, blocked before +the question existed, blocked by a decision saved before this rule, reassigned, +done or cancelled still cancels the answer's continuation, and Retry remains. Task interruption notices ignore Retry receipts for a superseded run. A later successful conversation must not inherit the old failure banner. This read-only projection neither retires Retry authority nor acknowledges independent saved input. diff --git a/server/src/__tests__/heartbeat-blocked-question-answer.test.ts b/server/src/__tests__/heartbeat-blocked-question-answer.test.ts new file mode 100644 index 00000000000..33aa15a8325 --- /dev/null +++ b/server/src/__tests__/heartbeat-blocked-question-answer.test.ts @@ -0,0 +1,197 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, + authUsers, + companies, + companyMemberships, + completionContracts, + createDb, + heartbeatRuns, + issueThreadInteractions, + issues, + nativeRunFinalizations, + nativeRunResults, + statusDecisions, + workAssessments, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { heartbeatService } from "../services/heartbeat.ts"; +import { issueService } from "../services/issues.ts"; +import { commitNativeStatusDecision } from "../services/native-runtime/status-decision-committer.js"; +import { NATIVE_STATUS_ARBITER_POLICY_VERSION } from "../services/native-runtime/status-arbiter.js"; + +const mockAdapterExecute = vi.hoisted(() => vi.fn(async () => ({ + exitCode: 0, signal: null, timedOut: false, errorMessage: null, + summary: "Wrote the brief with the answer.", provider: "test", model: "test-model", +}))); + +vi.mock("../adapters/index.ts", async () => { + const actual = await vi.importActual("../adapters/index.ts"); + return { ...actual, getServerAdapter: vi.fn(() => ({ supportsLocalAgentJwt: false, execute: mockAdapterExecute })) }; +}); + +const support = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = support.supported ? describe : describe.skip; + +describeEmbeddedPostgres("an answer to the agent's own question continues the task it blocked on", () => { + let db!: ReturnType; + let heartbeat!: ReturnType; + let tempDb: Awaited> | null = null; + const userId = "blocked-question-user"; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-blocked-question-answer-"); + db = createDb(tempDb.connectionString); + await db.insert(authUsers).values({ id: userId, name: "Board", email: "board@example.test", + createdAt: new Date(), updatedAt: new Date() }); + heartbeat = heartbeatService(db); + }, 30_000); + + afterAll(async () => { + await heartbeat?.drainActiveRunExecutions(); + await db?.$client.end(); + await tempDb?.cleanup(); + }); + + async function seed(options: { askAfterBlock?: boolean } = {}) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Question fixture", + issuePrefix: `Q${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: userId }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, + status: "active", membershipRole: "owner" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Writer", role: "engineer", status: "active", + adapterType: "codex_local", adapterConfig: {}, permissions: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Write the brief", status: "in_review", + assigneeAgentId: agentId, responsibleUserId: userId }); + if (options.askAfterBlock) await agentReportsBlocker({ companyId, agentId, issueId }); + const askingRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: askingRunId, companyId, agentId, status: "succeeded", + runtimeMode: "native", nativeIssueId: issueId, invocationSource: "assignment", triggerDetail: "system", + contextSnapshot: { issueId }, finishedAt: new Date() }); + const [question] = await db.insert(issueThreadInteractions).values({ companyId, issueId, + createdByAgentId: agentId, sourceRunId: askingRunId, kind: "ask_user_questions", status: "pending", + continuationPolicy: "wake_assignee", title: "Pick a codename", + payload: { version: 1, questions: [{ id: "codename", prompt: "Which codename?", selectionMode: "single", + options: [{ id: "alpha", label: "ALPHA" }, { id: "bravo", label: "BRAVO" }] }] }, + }).returning(); + return { companyId, agentId, issueId, question: question! }; + } + + async function agentReportsBlocker(input: { companyId: string; agentId: string; issueId: string }): Promise { + const runId = randomUUID(), contractId = randomUUID(), resultId = randomUUID(), assessmentId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId: input.companyId, issueId: input.issueId, + revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: {}, canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId: input.companyId, agentId: input.agentId, + status: "running", runtimeMode: "native", nativeIssueId: input.issueId, invocationSource: "automation", + triggerDetail: "system", completionContractId: contractId, contextSnapshot: { issueId: input.issueId } }); + await db.insert(nativeRunResults).values({ id: resultId, companyId: input.companyId, issueId: input.issueId, + runId, completionContractId: contractId, serverFingerprint: resultId, schemaStatus: "accepted", + resultJson: {}, canonicalSha256: resultId }); + const [snapshot] = await db.select().from(issues).where(eq(issues.id, input.issueId)); + await db.insert(workAssessments).values({ id: assessmentId, companyId: input.companyId, issueId: input.issueId, + runId, contractId, resultId, triggerKind: "native_result", triggerActorCompanyId: input.companyId, + priorIssueStatus: snapshot!.status, priorStatusVersion: snapshot!.statusVersion, + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, assessmentJson: {}, inputDigest: assessmentId }); + await db.insert(nativeRunFinalizations).values({ runId, companyId: input.companyId, issueId: input.issueId, + phase: "arbitrating", resultId, assessmentId }); + await commitNativeStatusDecision({ + db, companyId: input.companyId, issueId: input.issueId, runId, assessmentId, + priorStatus: snapshot!.status, priorStatusVersion: Number(snapshot!.statusVersion), + priorDecisionId: snapshot!.lastStatusDecisionId, + decision: { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "blocked", toStatus: "blocked", + reasonCode: "current_track_blocker_waiting", + unblockDescriptor: { owner: "board", action: "Answer the pending codename question." }, + effects: [{ kind: "release_checkout" }] }, + }); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + } + + async function answer(question: typeof issueThreadInteractions.$inferSelect) { + await db.update(issueThreadInteractions).set({ status: "answered", resolvedByUserId: userId, resolvedAt: new Date(), + result: { version: 1, answers: [{ questionId: "codename", optionIds: ["bravo"] }] } }) + .where(eq(issueThreadInteractions.id, question.id)); + } + + async function deliverAnswer(input: { agentId: string; issueId: string; questionId: string }) { + const context = { issueId: input.issueId, mutation: "interaction", interactionId: input.questionId, + interactionKind: "ask_user_questions", interactionStatus: "answered", + interactionResolvedAt: new Date().toISOString(), wakeReason: "issue_commented" }; + const run = await heartbeat.wakeup(input.agentId, { source: "automation", triggerDetail: "system", + reason: "issue_commented", payload: context, contextSnapshot: context, + requestedByActorType: "user", requestedByActorId: userId }); + await heartbeat.drainActiveRunExecutions(); + return run ? (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]! : null; + } + + it("delivers the answer when the agent blocked the task to wait for it", async () => { + mockAdapterExecute.mockClear(); + const { agentId, issueId, question, companyId } = await seed(); + await agentReportsBlocker({ companyId, agentId, issueId }); + const [blocked] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(blocked!.status).toBe("blocked"); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.id, blocked!.lastStatusDecisionId!)); + expect(decision!.decisionJson).toMatchObject({ awaitingInteractionIds: [question.id] }); + await answer(question); + const run = await deliverAnswer({ agentId, issueId, questionId: question.id }); + expect({ status: run?.status, errorCode: run?.errorCode ?? null, error: run?.error ?? null }) + .toEqual({ status: "succeeded", errorCode: null, error: null }); + expect(run?.startedAt).not.toBeNull(); + expect(run?.contextSnapshot).toMatchObject({ interactionId: question.id, interactionStatus: "answered" }); + expect(mockAdapterExecute).toHaveBeenCalled(); + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(issue).toMatchObject({ blockedTransitionAt: null, unblockDescriptor: null }); + }, 60_000); + + it("keeps cancelling the answer when a person blocked the task after the agent", async () => { + mockAdapterExecute.mockClear(); + const { agentId, issueId, question, companyId } = await seed(); + await agentReportsBlocker({ companyId, agentId, issueId }); + await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, issueId)); + await db.update(issues).set({ status: "blocked", blockedTransitionAt: new Date() }).where(eq(issues.id, issueId)); + await answer(question); + const run = await deliverAnswer({ agentId, issueId, questionId: question.id }); + expect(run).toMatchObject({ status: "cancelled", errorCode: "issue_not_in_progress" }); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + }, 60_000); + + it("keeps cancelling the answer when the question did not exist when the agent blocked", async () => { + mockAdapterExecute.mockClear(); + const { agentId, issueId, question } = await seed({ askAfterBlock: true }); + await answer(question); + const run = await deliverAnswer({ agentId, issueId, questionId: question.id }); + expect(run).toMatchObject({ status: "cancelled", errorCode: "issue_not_in_progress" }); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + }, 60_000); + + it("keeps cancelling the answer once the task is done", async () => { + mockAdapterExecute.mockClear(); + const { agentId, issueId, question, companyId } = await seed(); + await agentReportsBlocker({ companyId, agentId, issueId }); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId)); + await answer(question); + const run = await deliverAnswer({ agentId, issueId, questionId: question.id }); + if (run) expect(run).toMatchObject({ status: "cancelled" }); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + }, 60_000); + + it("does not check out a task whose verified block changed before checkout", async () => { + const { agentId, issueId, question, companyId } = await seed(); + await agentReportsBlocker({ companyId, agentId, issueId }); + await answer(question); + const [verified] = await db.select().from(issues).where(eq(issues.id, issueId)); + const blockedState = { lastStatusDecisionId: verified!.lastStatusDecisionId!, statusVersion: Number(verified!.statusVersion) }; + await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, issueId)); + await db.update(issues).set({ status: "blocked", unblockDescriptor: { owner: "board", action: "Person's new blocker" } }) + .where(eq(issues.id, issueId)); + await expect(issueService(db).checkout(issueId, agentId, ["in_progress", "in_review", "blocked"], null, { blockedState })) + .rejects.toThrow(); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]) + .toMatchObject({ status: "blocked", unblockDescriptor: { owner: "board", action: "Person's new blocker" } }); + }, 60_000); +}); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 84dbbff7769..e5247fe4ed5 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -44,6 +44,7 @@ import { lockRunBudgetRecoveryOwnerInTransaction, recheckRunBudgetReservationInT import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; import { isUnstartedTaskQuestionAnswer } from "../../../services/unstarted-task-question-answer.js"; import { isVerifiedDependencyInteractionResponse } from "../../../services/queued-interaction-response.js"; +import { agentBlockingQuestionState } from "../../../services/blocked-question-answer.js"; import { isHeartbeatWakeOnDemandEnabled } from "../../../services/heartbeat-policy.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; @@ -706,6 +707,9 @@ export function createPostgresRunDispatchAdapter( && context.dependencyBlockedInteraction === true && await isVerifiedDependencyInteractionResponse(dbOrTx, { companyId: input.companyId, issueId, agentId: input.agentId, context })), + isAgentBlockingQuestionAnswer: Boolean(isResolvedInteractionContinuation && issue?.status === "blocked" + && await agentBlockingQuestionState(dbOrTx, { companyId: input.companyId, issueId, + agentId: input.agentId, context })), isUnstartedTaskQuestionAnswer: Boolean(isResolvedInteractionContinuation && issue?.status === "todo" && await isUnstartedTaskQuestionAnswer(dbOrTx, { companyId: input.companyId, issueId, interactionId: context.interactionId })), isConnectionContinuation: (isResolvedInteractionContinuation && context.interactionKind === "connection_intent") diff --git a/server/src/modules/run-dispatch/domain/policy.test.ts b/server/src/modules/run-dispatch/domain/policy.test.ts index a97b3d294fd..d0d50fb357b 100644 --- a/server/src/modules/run-dispatch/domain/policy.test.ts +++ b/server/src/modules/run-dispatch/domain/policy.test.ts @@ -267,6 +267,19 @@ describe("decideScheduledRetryGate", () => { }); describe("decideQueuedRunStaleness", () => { + it("continues a task its agent blocked to wait for the answered question", () => { + const answer = { ...baseStalenessFacts(), isResolvedInteractionContinuation: true, + issueStatus: "blocked", isAgentBlockingQuestionAnswer: true }; + expect(decideQueuedRunStaleness(answer, NOW)).toEqual({ stale: false }); + expect(decideQueuedRunStaleness({ ...answer, isAgentBlockingQuestionAnswer: false }, NOW)) + .toMatchObject({ stale: true, errorCode: "issue_not_in_progress" }); + expect(decideQueuedRunStaleness({ ...answer, issueAssigneeAgentId: "other-agent" }, NOW)) + .toMatchObject({ stale: true, errorCode: "issue_assignee_changed" }); + for (const issueStatus of ["backlog", "done", "cancelled"]) { + expect(decideQueuedRunStaleness({ ...answer, issueStatus }, NOW)).toMatchObject({ stale: true }); + } + }); + it("keeps a verified response bounded to a blocked task and its current assignee", () => { const response = { ...baseStalenessFacts(), isResolvedInteractionContinuation: true, issueStatus: "blocked", isDependencyBlockedInteractionResponse: true }; diff --git a/server/src/modules/run-dispatch/domain/policy.ts b/server/src/modules/run-dispatch/domain/policy.ts index 655025d3b81..d1aa1426806 100644 --- a/server/src/modules/run-dispatch/domain/policy.ts +++ b/server/src/modules/run-dispatch/domain/policy.ts @@ -149,6 +149,7 @@ export type QueuedRunFacts = { isUnstartedTaskQuestionAnswer?: boolean; /** The stored human response was verified again for bounded dependency interaction. */ isDependencyBlockedInteractionResponse?: boolean; + isAgentBlockingQuestionAnswer?: boolean; /** A connection resolution or tool refresh can resume an agent waiting in review. */ isConnectionContinuation?: boolean; isInteractionWake: boolean; @@ -516,7 +517,8 @@ export function decideQueuedRunStaleness( if (facts.isResolvedInteractionContinuation || facts.isConnectionContinuation) { // A todo task that has never started has no parked execution to protect. const startsUnstartedTask = facts.issueStatus === "todo" && facts.isUnstartedTaskQuestionAnswer === true; - const answersBlockedTask = facts.issueStatus === "blocked" && facts.isDependencyBlockedInteractionResponse === true; + const answersBlockedTask = facts.issueStatus === "blocked" && + (facts.isDependencyBlockedInteractionResponse === true || facts.isAgentBlockingQuestionAnswer === true); const earlyStatus = decideIssueStatus({ status: facts.issueStatus, requiresInProgress: facts.issueStatus !== "in_review" && !startsUnstartedTask && !answersBlockedTask, diff --git a/server/src/services/blocked-question-answer.ts b/server/src/services/blocked-question-answer.ts new file mode 100644 index 00000000000..38355ea86eb --- /dev/null +++ b/server/src/services/blocked-question-answer.ts @@ -0,0 +1,68 @@ +import { and, eq, inArray, isNotNull, isNull, sql } from "drizzle-orm"; +import { heartbeatRuns, issueThreadInteractions, issues, statusDecisions, type Db } from "@paperclipai/db"; +import { queuedInteractionId } from "./queued-interaction-response.js"; +import { ordinaryQuestionCondition } from "./issue-question-context.js"; + +const AGENT_REPORTED_BLOCKER_REASON_CODES = ["current_track_blocker_waiting", "task_wide_blocker_bound"]; + +export type AgentBlockingQuestionState = { lastStatusDecisionId: string; statusVersion: number }; + +export async function questionsAwaitedByAgentBlock(db: Pick, input: { + companyId: string; issueId: string; agentId: string | null | undefined; reasonCode: string | null; + toStatus: string; unblockOwner: unknown; +}): Promise { + if (!input.agentId || input.toStatus !== "blocked" || input.unblockOwner !== "board" + || !AGENT_REPORTED_BLOCKER_REASON_CODES.includes(input.reasonCode ?? "")) return []; + const rows = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + eq(issueThreadInteractions.createdByAgentId, input.agentId), + eq(issueThreadInteractions.status, "pending"), + isNull(issueThreadInteractions.dismissedAt), + ordinaryQuestionCondition(), + )); + return rows.map((row) => row.id).sort(); +} + +export async function agentBlockingQuestionState(db: Pick, input: { + companyId: string; issueId: string; agentId: string; context: unknown; +}): Promise { + const interactionId = queuedInteractionId(input.context); + if (!interactionId) return null; + const interactionStatus = (input.context as Record).interactionStatus; + const [row] = await db + .select({ lastStatusDecisionId: statusDecisions.id, statusVersion: issues.statusVersion }) + .from(issueThreadInteractions) + .innerJoin(issues, and( + eq(issues.id, issueThreadInteractions.issueId), + eq(issues.companyId, issueThreadInteractions.companyId), + )) + .innerJoin(statusDecisions, and( + eq(statusDecisions.id, issues.lastStatusDecisionId), + eq(statusDecisions.companyId, issues.companyId), + eq(statusDecisions.issueId, issues.id), + )) + .innerJoin(heartbeatRuns, and( + eq(heartbeatRuns.id, statusDecisions.runId), + eq(heartbeatRuns.companyId, statusDecisions.companyId), + )) + .where(and( + eq(issueThreadInteractions.id, interactionId), + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + eq(issueThreadInteractions.createdByAgentId, input.agentId), + eq(issueThreadInteractions.status, String(interactionStatus)), + isNotNull(issueThreadInteractions.resolvedByUserId), + isNotNull(issueThreadInteractions.resolvedAt), + eq(issues.status, "blocked"), + eq(issues.assigneeAgentId, input.agentId), + eq(statusDecisions.applicationState, "applied"), + eq(statusDecisions.toStatus, "blocked"), + inArray(statusDecisions.reasonCode, AGENT_REPORTED_BLOCKER_REASON_CODES), + sql`coalesce(${statusDecisions.decisionJson}->'awaitingInteractionIds', '[]'::jsonb) ? ${interactionId}`, + sql`${statusDecisions.decisionJson}->>'projectedStatusVersion' = ${issues.statusVersion}::text`, + eq(heartbeatRuns.agentId, input.agentId), + )) + .limit(1); + return row ? { lastStatusDecisionId: row.lastStatusDecisionId, statusVersion: Number(row.statusVersion) } : null; +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7846c8d53f8..0ad306b27a5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -7,6 +7,7 @@ import { isProcessRunAdmissionClosed } from "./heartbeat/run-admission.js"; import type { StopInvocationOptions } from "./heartbeat/run-control.js"; import { isContextCompactionRun } from "./context-compaction.js"; import { isUnstartedTaskQuestionAnswer } from "./unstarted-task-question-answer.js"; +import { agentBlockingQuestionState } from "./blocked-question-answer.js"; import { isVerifiedDependencyInteractionResponse } from "./queued-interaction-response.js"; import { registerDeferredReceiptWork } from "./heartbeat/deferred-receipt-work.js"; import { registerProviderLimitResumeWork, type PrepareChatRetry } from "./native-runtime/provider-limit-resume.js"; @@ -2304,11 +2305,16 @@ export function heartbeatService( await timePreparation("issue_checkout", async () => { const startsUnstartedTask = issueStatus === "todo" && await isUnstartedTaskQuestionAnswer(db, { companyId: agent.companyId, issueId, interactionId: context.interactionId }); + const blockedState = issueStatus === "blocked" + ? await agentBlockingQuestionState(db, { companyId: agent.companyId, issueId, agentId: agent.id, context }) + : null; await issuesSvc.checkout( issueId, agent.id, - [...resolvedInteractionCheckoutExpectedStatuses(), ...(startsUnstartedTask ? ["todo"] : [])], + [...resolvedInteractionCheckoutExpectedStatuses(), ...(startsUnstartedTask ? ["todo"] : []), + ...(blockedState ? ["blocked"] : [])], run.id, + blockedState ? { blockedState } : undefined, ); }); context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = true; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index f34437d8be0..d8e3ecdfff6 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -12550,6 +12550,7 @@ export function issueService(db: Db) { agentId: string, expectedStatuses: string[], checkoutRunId: string | null, + options: { blockedState?: { lastStatusDecisionId: string; statusVersion: number } } = {}, ) => { const issueCompany = await db .select({ companyId: issues.companyId }) @@ -12651,11 +12652,21 @@ export function issueService(db: Db) { status: "in_progress", startedAt: now, updatedAt: now, + ...(expectedStatuses.includes("blocked") ? { + unblockDescriptor: sql`case when ${issues.status} = 'blocked' then null else ${issues.unblockDescriptor} end`, + blockedTransitionAt: sql`case when ${issues.status} = 'blocked' then null else ${issues.blockedTransitionAt} end`, + blockedOwnerNotifiedAt: sql`case when ${issues.status} = 'blocked' then null else ${issues.blockedOwnerNotifiedAt} end`, + } : {}), }) .where( and( eq(issues.id, id), inArray(issues.status, expectedStatuses), + options.blockedState ? or( + ne(issues.status, "blocked"), + and(eq(issues.lastStatusDecisionId, options.blockedState.lastStatusDecisionId), + eq(issues.statusVersion, options.blockedState.statusVersion)), + ) : undefined, or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition), executionLockCondition, ), diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 00143623712..92964d162a5 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -26,6 +26,7 @@ import { normalizePrpResultSignals, type PrpStructuredRunResult } from "../../ve import { readNativeMonitorDelivery, readNativePreservedMonitorWait, readNativeSupersededMonitorWait, type NativeSupersededMonitorWait, type NativeMonitorDelivery, type NativePreservedMonitorWait } from "./native-monitor-delivery.js"; import { readNativePlanWait, type NativePlanWaitSource } from "./native-plan-wait.js"; import { activeIssueInteractionCondition } from "../issue-question-context.js"; +import { questionsAwaitedByAgentBlock } from "../blocked-question-answer.js"; import { isConversation } from "../agent-conversations.js"; import { hasPendingNativeChildCompletion, type NativeChildCompletionRecipient } from "./native-child-completion-delivery.js"; import { and, asc, desc, eq, inArray, notInArray, or, sql } from "drizzle-orm"; @@ -2159,12 +2160,17 @@ export async function commitNativeStatusDecision(input: { }) .where(eq(nativeRunFinalizations.runId, input.runId)); }; + const awaitingInteractionIds = await questionsAwaitedByAgentBlock(tx as unknown as Db, { + companyId: input.companyId, issueId: input.issueId, agentId: originRun?.agentId, reasonCode, + toStatus: input.decision.toStatus, unblockOwner: input.decision.unblockDescriptor?.owner, + }); const decisionJson = { statusAction: input.decision.statusAction, toStatus: input.decision.toStatus, reasonCode, unblockDescriptor: input.decision.unblockDescriptor, effects: input.decision.effects, + ...(awaitingInteractionIds.length ? { awaitingInteractionIds } : {}), ...(externalChatReviewPresentation ? { externalChatReviewPresentation } : {}),