mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 15:55:46 +01:00
fix(dispatch): let the answer to the agent's own question continue the task it blocked on (#20931)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - An agent can ask the person a question (`request_human_input`) and wait for the answer > - A later comment from the person moves an open question into history (#15392). The agent still needs the answer, so it reports a blocker and the task becomes Blocked > - The person then answers. Dispatch cancels the answer's continuation with `issue_not_in_progress`, because only a dependency-blocked task may take an answer while Blocked > - The task rests Blocked with a failed run and a Retry. The answer is lost until a person clicks Retry > - This pull request lets the answer continue the task when the same agent blocked it to wait for that answer, proven from durable records > - The benefit is that the person's answer always reaches the agent that asked, without a Retry ## Linked Issues or Issue Description Refs #15392, Refs #20872 **What happened?** Restart-chaos waves with no fault injected fail every W2Q task on every provider (Claude, OpenCode, Codex) with `R3:retry_needed`. For Claude task t03 on master7e8901aa60: run 5aeb9244 asks a question and the task goes `in_review`. A board comment ("I will answer shortly; do not write the brief yet") wakes run d728886f. That comment makes the question historical, so `paperclip_finish` with a `response_wake` wait is rejected ("without a pending wait condition"). The agent then calls `paperclip_block`; the status decision `current_track_blocker_waiting` sets the task Blocked. The board answers. Continuation run adbda5de is cancelled: "resolved-interaction continuation issue is no longer in_progress (current status: blocked)". The task rests blocked/failed with Retry. **What did you expect to happen?** The answer continues the task. The agent writes the brief with the answer and finishes. **Steps to reproduce** 1. An agent asks a question on its task. 2. The person posts a comment before answering. 3. The agent reports a blocker while it waits for the answer. 4. The person answers the question. 5. The answer's continuation run is cancelled with `issue_not_in_progress`. ## What Changed - Status committer (`status-decision-committer.ts`): when it applies an agent-reported, board-owned blocker (`current_track_blocker_waiting` or `task_wide_blocker_bound`), it records the agent's own pending, undismissed ordinary questions on the task as `awaitingInteractionIds` in the decision JSON, in the same transaction under the issue lock (`questionsAwaitedByAgentBlock`). - New `agentBlockingQuestionState` (`server/src/services/blocked-question-answer.ts`). It reads only server-owned rows. It returns the verified `lastStatusDecisionId` and `statusVersion` when: the answered interaction was created by this agent on this task and resolved by a person; the task is Blocked and still assigned to the agent; the task's `lastStatusDecisionId` is an applied Blocked decision with one of the two reason codes, made by a run of the same agent, whose `awaitingInteractionIds` contains this interaction; and the task's `statusVersion` still equals that decision's `projectedStatusVersion` (nothing changed the status since). - `decideQueuedRunStaleness` (`server/src/modules/run-dispatch/domain/policy.ts`): the Blocked exemption for an answer continuation also accepts this fact (`isAgentBlockingQuestionAnswer`). The postgres adapter loads it under the issue lock. - `heartbeat.ts` resolved-interaction checkout: for that answer, `blocked` is an accepted expected status, and the verified state is passed to `issues.checkout` as `blockedState`. - `issues.checkout`: with `blockedState`, the guarded update accepts a Blocked row only when `lastStatusDecisionId` and `statusVersion` still match, so a person who re-blocks the task in between keeps the new block. When the expected statuses include `blocked`, a checkout from Blocked also clears `unblockDescriptor`, `blockedTransitionAt` and `blockedOwnerNotifiedAt`, as an ordinary status change from Blocked does. - `doc/execution-semantics.md` describes the rule. ## Verification - New integration test `server/src/__tests__/heartbeat-blocked-question-answer.test.ts` (embedded Postgres, real `commitNativeStatusDecision`): the agent's blocker decision makes the task Blocked, the person answers, the answer is delivered through `heartbeat.wakeup` with the same context shape as staging. - Red on master (product hunks removed, test kept): `delivers the answer when the agent blocked the task to wait for it` fails with `status: cancelled, errorCode: issue_not_in_progress`, the same code as the chaos run. - Green: 5/5 pass. The answer run succeeds, the adapter runs, and the blocker fields are cleared. Negative cases still cancel with `issue_not_in_progress`: a person re-blocked the task after the agent, the question did not exist when the agent blocked, and the task is done. A checkout with a stale verified state conflicts and keeps the person's new blocker. - `policy.test.ts`: new case for the fact (continue when set, cancel when unset, reassignment, backlog/done/cancelled). - `cd server && pnpm exec vitest run src/modules/run-dispatch src/__tests__/heartbeat-dependency-scheduling.test.ts`: 208 passed. - `cd server && pnpm exec vitest run src/__tests__/issues-service.test.ts src/__tests__/heartbeat-stale-queue-invalidation.test.ts src/services/explicit-native-continuation.test.ts`: 656 passed. - `cd server && pnpm exec tsc --noEmit -p tsconfig.json`: passes. - Chaos, local, seed 5101, `--workloads W2Q --faults K0 --adapters codex,claude --lifecycle warm`, on722303e4a2(first head): wave w-20261111T204405-s5101, 6/6 PASS (3 Codex, 3 Claude). Every task went in_review -> blocked -> in_progress -> done without Retry. Before: the same local seed on master (w-20261111T190352-s5101) failed 3/3 Codex W2Q cells with `R3:retry_needed`, and Daytona Claude wave w-20261111T200553-s6101 failed 2/3 W2Q cells the same way. ## Risks - Low to medium. Only a resolved interaction continuation on a Blocked task changes behavior, and only when the durable linkage holds. If a board-owned blocker about something else and the agent's own question are open together, the answer resumes the agent; it then reports its blocker again. The blocker report carries no interaction reference, so the server cannot tell them apart; cancelling the answer instead loses it. A person's Blocked status, a block from before the question, dependency blocks, reassignment, done and cancelled keep the old cancel. - State saved by an older release: no schema change and no migration. A Blocked decision saved before this change has no `awaitingInteractionIds`, so its answer keeps the old behavior (cancel, Retry works). - The checkout change only applies when a caller passes `blocked` as an expected status. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), with tool use and code execution in an agent harness. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
b01959cb15
commit
228b0169be
9 files changed
+327
-2
No files matched your search
@@ -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<typeof import("../adapters/index.ts")>("../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<typeof createDb>;
|
||||
let heartbeat!: ReturnType<typeof heartbeatService>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<void> {
|
||||
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);
|
||||
});
|
||||
@@ -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")
|
||||
|
||||
@@ -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 };
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<Db, "select">, input: {
|
||||
companyId: string; issueId: string; agentId: string | null | undefined; reasonCode: string | null;
|
||||
toStatus: string; unblockOwner: unknown;
|
||||
}): Promise<string[]> {
|
||||
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<Db, "select">, input: {
|
||||
companyId: string; issueId: string; agentId: string; context: unknown;
|
||||
}): Promise<AgentBlockingQuestionState | null> {
|
||||
const interactionId = queuedInteractionId(input.context);
|
||||
if (!interactionId) return null;
|
||||
const interactionStatus = (input.context as Record<string, unknown>).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;
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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,
|
||||
),
|
||||
|
||||
@@ -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 }
|
||||
: {}),
|
||||
|
||||
Reference in new issue
Block a user