mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 15:55:46 +01:00
fix(native): park a response_wake with no admitted response instead of self-waking (#20935)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - A native agent run ends with a structured result. The status arbiter
turns that result into an issue status and next actions.
> - A result can report `yielded` with `continuation.kind:
"response_wake"`. This means "wake me when somebody responds".
> - Only a direct Board comment gave this kind a passive wait. Every
other `response_wake` became an immediate `issue_status_changed` wake to
the same agent.
> - The agent woke, saw nothing new, reported `response_wake` again, and
woke again. Nothing bounded this loop. It spent provider budget and its
back-to-back runs caused other failures on warm sandboxes.
> - This pull request parks such a result as a visible wait, and puts
declared continuation wakes under the rewake throttle as a backstop.
> - The benefit is that an agent that waits for input stops and costs
nothing until a real response arrives.
## Linked Issues or Issue Description
**What happened?**
Restart-chaos waves on master `683fd1e9a5` (no faults injected) showed
one task run 30 to 66 times, about 12 s apart. Example: wave
`w-20261111T190352-s5101`, task t02 (W2C), 65 runs in 13 minutes. Each
run reported `yielded` with `continuation.kind: "response_wake"` and the
summary "No changes made. This status-change wake does not trigger phase
2". Each run was woken by `issue_status_changed`, and each finalization
committed `live_continuation_registered` with an `enqueue_continuation`
effect of kind `response_wake`.
Root cause: `arbitrateNativeStatus()`
(`server/src/services/native-runtime/status-arbiter.ts`) only treats
`response_wake` as passive when `boardResponseWaitAuthorized` (from
`readNativeBoardResponseWaitSource`), external chat, or plan wait is
true. Otherwise it falls to the generic yielded branch, and
`status-decision-committer.ts` (`enqueue_continuation`) enqueues an
`issue_status_changed` wake to the same agent. `issue_status_changed` is
not in `THROTTLED_ISSUE_REWAKE_REASONS`, and the native committer writes
an `issue.updated` activity row with the run's id for every decision, so
the rewake throttle saw every native run as progress.
**Expected behavior**
A `response_wake` with no admitted response source parks the task. A
comment, an interaction answer or a status change by somebody else wakes
the agent. No path can loop without a bound.
**Steps to reproduce**
Run `tests/restart-chaos/main.ts --env local --adapters codex
--workloads W2,W2Q,W2C --concurrency 10 --tasks 10 --seed 5101
--lifecycle warm --triggers time`. W2/W2C tasks loop after phase 1.
## What Changed
- `status-arbiter.ts`: a `yielded` / `response_wake` result with no
admitted response source and no blocking remaining work now commits
`response_wake_waiting_for_input`, keeps the task `in_progress`, and has
no effects. Policy version is `phase6-v14`. Board, external chat, plan,
governance, dependency, pause and unfinished-work branches are unchanged
and still come first.
- `native-runtime/native-response-wake-wait.ts` (new) and
`recovery/service.ts`: recovery treats that applied decision as a
durable wait while it is the issue's latest decision, the issue is
`in_progress` with the same agent assignee, and no user comment arrived
after the run started. This applies to the stranded-work sweep and to
the paused-agent passive-wait check.
- `execution-projection.ts`: a committed run with this reason shows the
phase `waiting_for_answer`, label "Waiting for a reply", and a next
action that tells the user to comment or change the status. This shows
only while that decision is still the issue's latest decision and the
issue is `in_progress`; an older run shows "Completed".
- Backstop in `issue-rewake-throttle.ts` and `heartbeat/queue.ts`: an
`issue_status_changed` wake that carries a declared `continuationKind`
(the `enqueue_continuation` effect) is now a throttle candidate. The
native status committer's own `issue.updated` / status-decision activity
no longer counts as run progress or as new input. After two consecutive
successful runs with no issue-visible progress, such a wake waits for
the existing escalating cooldown (2 min, doubling, max 30 min). The
native status dispatcher (`dispatchNativeStatusWakeBatch`) does not drop
a throttled intent: it keeps the intent `queued`, reports the cooldown
deadline to the deferred-receipt worker, and dispatches it again when
the cooldown ends. Status wakes without a declared continuation (blocker
owner, notify owner, retry) are unchanged.
- Restart chaos: new enforced universal check U5 (`universal.ts`). A
task with more than 10 runs fails as `U5:run_budget_exceeded`
(`RESTART_CHAOS_U5_MAX_RUNS`). The README records the calibration.
- `doc/architecture/native-status-arbitration.md`: the decision table
and the wait rules.
## Verification
- RED on master (same reason as the waves):
`heartbeat-process-recovery.test.ts` "parks a native response_wake that
no response admitted…" fails with decision `reasonCode:
"live_continuation_registered"` and effect `{ kind:
"enqueue_continuation", continuationKind: "response_wake" }`.
`heartbeat-issue-rewake-throttle.test.ts` "throttles a
same_agent/response_wake/retry continuation…" fails 3/3 because the wake
is admitted.
- GREEN: the integration test finalizes the run, dispatches native
status wakes and runs the stranded-work sweep three times. It asserts
one run total, one wake request, no recovery action, the task
`in_progress`, and the projection "Waiting for a reply".
- `heartbeat-process-recovery.test.ts` "delays a no-progress native
continuation until the rewake cooldown…" finalizes a real `same_agent`
continuation after a no-progress run, dispatches twice (one throttle
receipt, intent stays `queued`, deadline equals `nextAllowedAt`), then
dispatches after the cooldown and gets the run. It fails when the
dispatcher drops the throttled intent.
- Focused suites pass: `heartbeat-process-recovery`,
`heartbeat-issue-rewake-throttle`, `status-arbiter`,
`native-status-arbiter-corpus`, `runtime-mode`, `execution-projection`,
`chat-run-milestone-work`, `heartbeat-dependency-scheduling`,
`heartbeat-run-summary` (904 tests), and all restart-chaos unit tests
(584 tests). `pnpm --filter @paperclipai/server typecheck` passes.
- Universal audit over the 17 saved waves of 2026-11-11: U5 flags 12
tasks, all looping tasks (12 to 66 runs), and 0 tasks on all-pass waves.
- Chaos rerun of local warm seed 5101 (W2, W2Q, W2C): result is added in
a comment on this PR.
## Risks
- Behavioral shift: a model that reports `response_wake` to mean
"continue now" no longer gets an immediate run. The task waits visibly
for a reply. The runner prompt already describes `response_wake` as a
real response wait, and `same_agent` stays available for server-owned
continuations.
- The throttle now applies to native runs. A chain of declared
continuations with no issue-visible output (no comment, document, work
product or other activity) waits for the cooldown after the second such
run. `same_agent` continuations are server-owned (repository
preparation), so this affects only repeated no-output chains.
- Old saved state: decisions committed under `phase6-v13` keep their
recorded effects. A run that finalizes after deploy uses the new policy.
No migration.
- Budget hard-stop behavior is unchanged.
## Model Used
- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with
tool use (shell, file edits, test runs).
## 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:
1 parent
80b2369f26
commit
edd00526e7
16 files changed
+400
-29
No files matched your search
@@ -355,6 +355,92 @@ describeEmbeddedPostgres("heartbeat issue rewake throttle", () => {
|
||||
expect((await latestWakeRequest(agentId))?.reason).toBe("issue_rewake_throttled");
|
||||
});
|
||||
|
||||
function nativeStatusWake(agentId: string, issueId: string, payload: Record<string, unknown>) {
|
||||
return heartbeat.wakeup(agentId, {
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_status_changed",
|
||||
payload: { issueId, nativeDecisionId: randomUUID(), ...payload },
|
||||
contextSnapshot: { issueId, wakeReason: "issue_status_changed", source: "native_status_decision" },
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: "native-status-committer",
|
||||
});
|
||||
}
|
||||
|
||||
async function recordNativeStatusDecision(companyId: string, issueId: string, runId: string) {
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "system",
|
||||
actorId: "native-status-committer",
|
||||
runId,
|
||||
action: "issue.updated",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
details: { source: "native_status_decision", reasonCode: "live_continuation_registered" },
|
||||
createdAt: new Date(Date.now() - 11_000),
|
||||
});
|
||||
}
|
||||
|
||||
it.each(["same_agent", "response_wake", "retry"])(
|
||||
"throttles a %s continuation the agent declared after consecutive no-progress runs",
|
||||
async (continuationKind) => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
for (const finishedSecondsAgo of [40, 10]) {
|
||||
const runId = await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo });
|
||||
await recordNativeStatusDecision(companyId, issueId, runId);
|
||||
}
|
||||
|
||||
const wake = await nativeStatusWake(agentId, issueId, {
|
||||
continuationKind,
|
||||
continuationSummary: "Nothing changed; wait.",
|
||||
continuationIdempotencyKey: `loop-${randomUUID()}`,
|
||||
});
|
||||
expect(wake).toBeNull();
|
||||
expect(await latestWakeRequest(agentId)).toMatchObject({
|
||||
status: "skipped",
|
||||
reason: "issue_rewake_throttled",
|
||||
payload: expect.objectContaining({
|
||||
continuationKind,
|
||||
heartbeatSkip: expect.objectContaining({ requestedReason: "issue_status_changed", noProgressStreak: 2 }),
|
||||
}),
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it("keeps a declared continuation prompt after a run with issue-visible progress", async () => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 40 });
|
||||
const progressRunId = await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 10 });
|
||||
await recordNativeStatusDecision(companyId, issueId, progressRunId);
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "agent",
|
||||
actorId: agentId,
|
||||
agentId,
|
||||
runId: progressRunId,
|
||||
action: "issue.document_upserted",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
createdAt: new Date(Date.now() - 12_000),
|
||||
});
|
||||
|
||||
const wake = await nativeStatusWake(agentId, issueId, {
|
||||
continuationKind: "same_agent",
|
||||
continuationSummary: "Continue the remaining steps.",
|
||||
continuationIdempotencyKey: "next-step",
|
||||
});
|
||||
expect(wake).not.toBeNull();
|
||||
});
|
||||
|
||||
it("does not throttle status wakes that carry no declared continuation", async () => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 40 });
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 10 });
|
||||
|
||||
const wake = await nativeStatusWake(agentId, issueId, { unblockAction: "Resolve the dependency." });
|
||||
expect(wake).not.toBeNull();
|
||||
});
|
||||
|
||||
it("does not throttle the wake that follows a failed run", async () => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
|
||||
|
||||
@@ -27,6 +27,7 @@ import { createHash, randomUUID } from "node:crypto";
|
||||
import { recordLegacyWorkspaceRestoreFailure, terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { getExecutionBlocker } from "../services/execution-blocker.js";
|
||||
import { executionProjectionForRun } from "../services/execution-projection.js";
|
||||
import { JsonRpcCallError } from "@paperclipai/plugin-sdk";
|
||||
import { leaseFaultError } from "../services/environment-lease-fault-injection.js";
|
||||
import { taskInterruptionPresentation } from "../services/task-interruption-presentation.js";
|
||||
@@ -14953,7 +14954,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
async function seedNativePassiveBoardResponse(
|
||||
continuationKind:
|
||||
"response_wake" | "same_agent" | "retry" = "response_wake",
|
||||
options: { blockingWork?: boolean; repairUsed?: boolean } = {},
|
||||
options: { blockingWork?: boolean; repairUsed?: boolean; assignmentWake?: boolean } = {},
|
||||
) {
|
||||
const fixture = await seedStrandedIssueFixture({
|
||||
status: "in_progress",
|
||||
@@ -14983,7 +14984,12 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
});
|
||||
await db
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
.set(options.assignmentWake ? {
|
||||
status: "completed",
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: "issue-assignment",
|
||||
payload: { issueId },
|
||||
} : {
|
||||
source: "automation",
|
||||
reason: "issue_commented",
|
||||
status: "completed",
|
||||
@@ -15127,6 +15133,74 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("parks a native response_wake that no response admitted instead of waking the same agent again", async () => {
|
||||
const fixture = await seedNativePassiveBoardResponse("response_wake", { assignmentWake: true });
|
||||
expect(await readNativeBoardResponseWaitSource(db, fixture)).toBeNull();
|
||||
await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
|
||||
const heartbeat = heartbeatService(db);
|
||||
for (let sweep = 0; sweep < 3; sweep += 1) {
|
||||
await heartbeat.dispatchPendingNativeStatusWakeups({ companyId: fixture.companyId });
|
||||
const recovery = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(recovery.continuationRequeued + recovery.escalated).toBe(0);
|
||||
}
|
||||
expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, fixture.runId))).toEqual([
|
||||
expect.objectContaining({ reasonCode: "response_wake_waiting_for_input", toStatus: "in_progress", applicationState: "applied" }),
|
||||
]);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, fixture.companyId))).toHaveLength(1);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, fixture.companyId))).toHaveLength(1);
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))).toHaveLength(0);
|
||||
expect(await db.select().from(issues).where(eq(issues.id, fixture.issueId))).toEqual([
|
||||
expect.objectContaining({ status: "in_progress", assigneeAgentId: fixture.agentId }),
|
||||
]);
|
||||
expect(await executionProjectionForRun(db, fixture.companyId, fixture.runId))
|
||||
.toMatchObject({ phase: "waiting_for_answer", label: "Waiting for a reply" });
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
await db.update(issues).set({ status: "done" }).where(eq(issues.id, fixture.issueId));
|
||||
expect(await executionProjectionForRun(db, fixture.companyId, fixture.runId))
|
||||
.toMatchObject({ phase: "completed", label: "Completed" });
|
||||
});
|
||||
|
||||
it("delays a no-progress native continuation until the rewake cooldown instead of dropping it", async () => {
|
||||
const fixture = await seedNativePassiveBoardResponse("same_agent", { assignmentWake: true });
|
||||
const finishedAt = new Date(Date.now() - 30_000);
|
||||
const priorRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: priorRunId, companyId: fixture.companyId, agentId: fixture.agentId,
|
||||
invocationSource: "assignment", status: "succeeded", startedAt: new Date(finishedAt.getTime() - 5_000), finishedAt,
|
||||
contextSnapshot: { issueId: fixture.issueId } });
|
||||
await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
|
||||
const [intent] = await db.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, fixture.companyId), eq(agentWakeupRequests.requestedByActorId, "native-status-committer")));
|
||||
expect(intent?.payload).toMatchObject({ continuationKind: "same_agent" });
|
||||
const heartbeat = heartbeatService(db);
|
||||
const deadlines: number[] = [];
|
||||
const dispatch = () => heartbeat.dispatchPendingNativeStatusWakeups({
|
||||
companyId: fixture.companyId, requestId: intent!.id, onDeferredDeadline: (at) => deadlines.push(at),
|
||||
});
|
||||
const throttled = () => db.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, fixture.companyId), eq(agentWakeupRequests.reason, "issue_rewake_throttled")));
|
||||
const runCount = async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, fixture.companyId))).length;
|
||||
|
||||
await dispatch();
|
||||
await dispatch();
|
||||
const [receipt] = await throttled();
|
||||
const nextAllowedAt = Date.parse(String((receipt?.payload as { heartbeatSkip?: { nextAllowedAt?: string } } | null)?.heartbeatSkip?.nextAllowedAt));
|
||||
expect(deadlines).toEqual([nextAllowedAt, nextAllowedAt]);
|
||||
expect(await throttled()).toHaveLength(1);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, intent!.id)))
|
||||
.toEqual([expect.objectContaining({ status: "queued", runId: null })]);
|
||||
expect(await runCount()).toBe(2);
|
||||
|
||||
const past = new Date(Date.now() - 30 * 60_000);
|
||||
await db.update(heartbeatRuns).set({ finishedAt: past }).where(eq(heartbeatRuns.companyId, fixture.companyId));
|
||||
await db.update(agentWakeupRequests).set({
|
||||
payload: sql`jsonb_set(${agentWakeupRequests.payload}, '{heartbeatSkip,nextAllowedAt}', to_jsonb(${past.toISOString()}::text))`,
|
||||
}).where(eq(agentWakeupRequests.id, receipt!.id));
|
||||
await dispatch();
|
||||
expect(await runCount()).toBe(3);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, intent!.id)))
|
||||
.toEqual([expect.objectContaining({ status: "coalesced", runId: expect.any(String) })]);
|
||||
});
|
||||
|
||||
it("rejects an unfinished native response wait while the finish report is still correctable", async () => {
|
||||
const fixture = await seedNativePassiveBoardResponse("response_wake", { blockingWork: true });
|
||||
const [stored] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, fixture.runId));
|
||||
@@ -15202,7 +15276,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
await commitNativeStatusDecision({ db, companyId: fixture.companyId, issueId: fixture.issueId, runId: fixture.runId,
|
||||
assessmentId: coordinator!.assessmentId!, priorStatus: issue!.status, priorStatusVersion: issue!.statusVersion,
|
||||
priorDecisionId: issue!.lastStatusDecisionId,
|
||||
decision: { policyVersion: "phase6-v13", statusAction: "in_progress", toStatus: "in_progress",
|
||||
decision: { policyVersion: "phase6-v14", statusAction: "in_progress", toStatus: "in_progress",
|
||||
reasonCode: "board_response_waiting", unblockDescriptor: null, effects: [] },
|
||||
requireBoardResponseWaitSource: source!.source, requireBoardResponseWaitOrigin: origin!,
|
||||
});
|
||||
@@ -15826,7 +15900,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
priorStatusVersion: issue!.statusVersion,
|
||||
priorDecisionId: issue!.lastStatusDecisionId,
|
||||
decision: {
|
||||
policyVersion: "phase6-v13",
|
||||
policyVersion: "phase6-v14",
|
||||
statusAction: reasonCode === "prior_status_preserved_no_live_path" ? "preserve" : "in_progress",
|
||||
toStatus: "in_progress",
|
||||
reasonCode,
|
||||
@@ -17035,7 +17109,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
await db.update(issueQuestionResponseDeliveries).set({ payloadSha256: "changed" }).where(eq(issueQuestionResponseDeliveries.id, f.deliveryId));
|
||||
await expect(commitNativeStatusDecision({ db, companyId: f.companyId, issueId: f.issueId, runId: f.runId, assessmentId: coordinator!.assessmentId!,
|
||||
priorStatus: issue!.status, priorStatusVersion: issue!.statusVersion, priorDecisionId: issue!.lastStatusDecisionId,
|
||||
decision: { policyVersion: "phase6-v13", statusAction: "in_progress", toStatus: "in_progress", reasonCode: "native_plan_accepted_waiting_for_continuation", unblockDescriptor: null, effects: [] },
|
||||
decision: { policyVersion: "phase6-v14", statusAction: "in_progress", toStatus: "in_progress", reasonCode: "native_plan_accepted_waiting_for_continuation", unblockDescriptor: null, effects: [] },
|
||||
requirePlanWaitSource: f.proof.source })).rejects.toBeInstanceOf(NativeStatusRaceError);
|
||||
expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId))).toHaveLength(0);
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.runId))).toHaveLength(0);
|
||||
|
||||
@@ -21,6 +21,7 @@ import { readTaskDrain } from "./task-admission.js";
|
||||
import { getStartupRecoveryState } from "../startup-recovery-state.js";
|
||||
import { hasPendingRunCleanup } from "./execution-blocker.js";
|
||||
import { isAgentAwaitingSetup } from "../modules/agent-lifecycle/index.js";
|
||||
import { NATIVE_RESPONSE_WAKE_WAITING_REASON } from "./native-runtime/status-arbiter.js";
|
||||
const MAX_PREDECESSOR_CLEANUP_CHECKS = 16;
|
||||
const text = (v: unknown) => (typeof v === "string" ? v : null);
|
||||
const executionRunColumns = {
|
||||
@@ -48,6 +49,9 @@ const executionRunColumns = {
|
||||
'failureRetriesBeforeAiConnectionWait', ${heartbeatRuns.contextSnapshot}->'failureRetriesBeforeAiConnectionWait',
|
||||
'failureRetriesBeforeWorkspaceWait', ${heartbeatRuns.contextSnapshot}->'failureRetriesBeforeWorkspaceWait',
|
||||
'executionRetryAccounting', ${heartbeatRuns.contextSnapshot}->'executionRetryAccounting')`,
|
||||
resultJson: sql<Record<string, unknown>>`jsonb_build_object(
|
||||
'finalizationPhase', ${heartbeatRuns.resultJson}->'finalizationPhase',
|
||||
'finalizationReasonCode', ${heartbeatRuns.resultJson}->'finalizationReasonCode')`,
|
||||
};
|
||||
type Run = Pick<typeof heartbeatRuns.$inferSelect, keyof typeof executionRunColumns>;
|
||||
type Coordinator = typeof nativeRunFinalizations.$inferSelect;
|
||||
@@ -240,6 +244,18 @@ export async function executionProjectionsForRuns(
|
||||
}
|
||||
}
|
||||
const coordinatorByRun = new Map(coordinators.map((row) => [row.runId, row]));
|
||||
const responseWaitRuns = runs.filter(run => run.status === "succeeded"
|
||||
&& run.resultJson?.finalizationReasonCode === NATIVE_RESPONSE_WAKE_WAITING_REASON && issueIdOf(run));
|
||||
const responseWaitIssues = responseWaitRuns.length ? await read(async () => db
|
||||
.select({ id: issues.id, status: issues.status, lastStatusDecisionId: issues.lastStatusDecisionId })
|
||||
.from(issues)
|
||||
.where(and(eq(issues.companyId, companyId), inArray(issues.id, [...new Set(responseWaitRuns.map(run => issueIdOf(run)!))])))) : [];
|
||||
const responseWaitIssueById = new Map(responseWaitIssues.map(row => [row.id, row]));
|
||||
const currentResponseWait = (run: Run) => {
|
||||
const issue = responseWaitIssueById.get(issueIdOf(run) ?? "");
|
||||
const decisionId = coordinatorByRun.get(run.id)?.decisionId;
|
||||
return Boolean(issue && decisionId && issue.status === "in_progress" && issue.lastStatusDecisionId === decisionId);
|
||||
};
|
||||
for (const run of runs) {
|
||||
const issueId = run.nativeIssueId ?? text(run.contextSnapshot?.issueId);
|
||||
const matching = recovery.filter(
|
||||
@@ -261,6 +277,7 @@ export async function executionProjectionsForRuns(
|
||||
waitingByRun.get(run.id),
|
||||
settlingPredecessorKeys.has(run.id),
|
||||
setupByAgent.get(run.agentId),
|
||||
currentResponseWait(run),
|
||||
),
|
||||
);
|
||||
}
|
||||
@@ -287,6 +304,7 @@ export function projectExecution(
|
||||
waitingForProvider?: WaitingExternalProvider,
|
||||
waitingForPredecessor = false,
|
||||
setupAgent?: { lifecycleState: string; lifecycleHolds: readonly string[]; retryRequired?: boolean },
|
||||
responseWaitCurrent = false,
|
||||
): ExecutionProjection {
|
||||
const detail = coordinator?.failureDetail ?? {};
|
||||
const successorRunId = text(detail.successorRunId);
|
||||
@@ -466,6 +484,11 @@ export function projectExecution(
|
||||
? "Waiting for access"
|
||||
: "Waiting for answer",
|
||||
);
|
||||
if (responseWaitCurrent && run.resultJson?.finalizationPhase === "committed" &&
|
||||
run.resultJson.finalizationReasonCode === NATIVE_RESPONSE_WAKE_WAITING_REASON) {
|
||||
projection.nextAction = "The agent finished its turn and is waiting for a reply. Comment on the task or change its status to continue.";
|
||||
return set("waiting_for_answer", "Waiting for a reply");
|
||||
}
|
||||
return set("completed", "Completed");
|
||||
}
|
||||
if (["failed", "cancelled", "timed_out", "interrupted"].includes(run.status))
|
||||
|
||||
@@ -153,6 +153,7 @@ import {
|
||||
ISSUE_REWAKE_RUN_SAMPLE_LIMIT,
|
||||
evaluateIssueRewakeThrottle,
|
||||
isThrottleCandidateIssueRewake,
|
||||
NATIVE_STATUS_COMMITTER_ACTOR_ID,
|
||||
} from "../issue-rewake-throttle.js";
|
||||
import { logActivity, publishActivity, type ActivityPublication } from "../activity-log.js";
|
||||
import {
|
||||
@@ -4388,9 +4389,11 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
forceFreshSession:
|
||||
enrichedContextSnapshot.forceFreshSession === true,
|
||||
hasExplicitResume: Boolean(explicitResumeSession),
|
||||
declaredContinuationKind: readNonEmptyString(parseObject(payload).continuationKind),
|
||||
})
|
||||
) {
|
||||
const throttleNow = new Date();
|
||||
const notNativeStatusCommitterActivity = sql`not (${activityLog.actorType} = 'system' and ${activityLog.actorId} = ${NATIVE_STATUS_COMMITTER_ACTOR_ID})`;
|
||||
const recentTerminalRuns = await tx
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
@@ -4430,6 +4433,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
activityLog.action,
|
||||
ISSUE_PROGRESS_ACTIVITY_ACTIONS,
|
||||
),
|
||||
notNativeStatusCommitterActivity,
|
||||
),
|
||||
);
|
||||
const lastRunFinishedAt =
|
||||
@@ -4445,6 +4449,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
eq(activityLog.entityId, issue.id),
|
||||
gt(activityLog.createdAt, lastRunFinishedAt),
|
||||
sql`coalesce(${activityLog.details}->>'origin', '') <> 'fast_response'`,
|
||||
notNativeStatusCommitterActivity,
|
||||
inArray(
|
||||
activityLog.action,
|
||||
ISSUE_NEW_INPUT_ACTIVITY_ACTIONS,
|
||||
@@ -5070,6 +5075,12 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
return (input: NativeStatusWakeDispatchInput = {}) => dispatchNativeStatusWakeBatch(input, deliveredByIssueScope);
|
||||
}
|
||||
|
||||
function throttledDispatchUntil(dispatch: typeof agentWakeupRequests.$inferSelect): number | null {
|
||||
if (dispatch.status !== "skipped" || dispatch.reason !== "issue_rewake_throttled") return null;
|
||||
const at = Date.parse(String(parseObject(parseObject(dispatch.payload).heartbeatSkip).nextAllowedAt ?? ""));
|
||||
return Number.isFinite(at) ? at : null;
|
||||
}
|
||||
|
||||
function dispatchPendingNativeStatusWakeups(input: NativeStatusWakeDispatchInput = {}) {
|
||||
return createNativeStatusWakeBatch()(input);
|
||||
}
|
||||
@@ -5124,7 +5135,14 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (existingDispatch) {
|
||||
const throttledUntil = existingDispatch ? throttledDispatchUntil(existingDispatch) : null;
|
||||
if (throttledUntil !== null && throttledUntil > now.getTime()) {
|
||||
input.onDeferredDeadline?.(throttledUntil);
|
||||
deferred += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (existingDispatch && throttledUntil === null) {
|
||||
const recoveredStatus = existingDispatch.runId
|
||||
? "coalesced"
|
||||
: existingDispatch.status === "deferred_issue_execution"
|
||||
@@ -5300,6 +5318,17 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
const throttledDeliveryUntil = !wakeRun && delivered ? throttledDispatchUntil(delivered) : null;
|
||||
if (throttledDeliveryUntil !== null) {
|
||||
await db
|
||||
.update(agentWakeupRequests)
|
||||
.set({ status: "queued", claimedAt: null, finishedAt: null, error: null, updatedAt: new Date() })
|
||||
.where(and(eq(agentWakeupRequests.id, candidate.id), isNull(agentWakeupRequests.runId)));
|
||||
input.onDeferredDeadline?.(throttledDeliveryUntil);
|
||||
deferred += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
// The committer intent is the durable outbox entry. When admission is
|
||||
// blocked, enqueueWakeup creates a separate deferred dispatch receipt;
|
||||
// leave the original intent coalesced so the release drain cannot
|
||||
|
||||
@@ -103,8 +103,11 @@ export interface IssueRewakeCandidateInput {
|
||||
requestedByActorType?: "user" | "agent" | "system" | null;
|
||||
forceFreshSession: boolean;
|
||||
hasExplicitResume: boolean;
|
||||
declaredContinuationKind?: string | null;
|
||||
}
|
||||
|
||||
export const NATIVE_STATUS_COMMITTER_ACTOR_ID = "native-status-committer";
|
||||
|
||||
/**
|
||||
* Whether a wake is even a candidate for throttling. Wakes that carry new
|
||||
* information or an explicit operator escalation always pass.
|
||||
@@ -116,6 +119,7 @@ export function isThrottleCandidateIssueRewake(input: IssueRewakeCandidateInput)
|
||||
if (input.hasExplicitResume && input.requestedByActorType !== "agent") return false;
|
||||
if (input.wakeCommentId) return input.requestedByActorType === "agent";
|
||||
if (input.reason === null) return true;
|
||||
if (input.reason === "issue_status_changed" && input.declaredContinuationKind) return true;
|
||||
return THROTTLED_ISSUE_REWAKE_REASONS.has(input.reason);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
import {
|
||||
heartbeatRuns,
|
||||
issueComments,
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
statusDecisions,
|
||||
type Db,
|
||||
} from "@paperclipai/db";
|
||||
import { NATIVE_RESPONSE_WAKE_WAITING_REASON } from "./status-arbiter.js";
|
||||
|
||||
type Binding = {
|
||||
companyId: string;
|
||||
issueId: string;
|
||||
runId: string;
|
||||
agentId: string;
|
||||
};
|
||||
|
||||
export async function hasCommittedNativeResponseWakeWait(
|
||||
db: Db,
|
||||
binding: Binding,
|
||||
): Promise<boolean> {
|
||||
const [wait] = await db
|
||||
.select({ runStartedAt: heartbeatRuns.startedAt })
|
||||
.from(nativeRunFinalizations)
|
||||
.innerJoin(
|
||||
statusDecisions,
|
||||
and(
|
||||
eq(statusDecisions.id, nativeRunFinalizations.decisionId),
|
||||
eq(statusDecisions.companyId, binding.companyId),
|
||||
eq(statusDecisions.issueId, binding.issueId),
|
||||
eq(statusDecisions.runId, binding.runId),
|
||||
eq(statusDecisions.applicationState, "applied"),
|
||||
eq(statusDecisions.reasonCode, NATIVE_RESPONSE_WAKE_WAITING_REASON),
|
||||
),
|
||||
)
|
||||
.innerJoin(
|
||||
heartbeatRuns,
|
||||
and(
|
||||
eq(heartbeatRuns.id, binding.runId),
|
||||
eq(heartbeatRuns.companyId, binding.companyId),
|
||||
eq(heartbeatRuns.agentId, binding.agentId),
|
||||
eq(heartbeatRuns.nativeIssueId, binding.issueId),
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
eq(heartbeatRuns.status, "succeeded"),
|
||||
),
|
||||
)
|
||||
.innerJoin(
|
||||
issues,
|
||||
and(
|
||||
eq(issues.id, binding.issueId),
|
||||
eq(issues.companyId, binding.companyId),
|
||||
eq(issues.assigneeAgentId, binding.agentId),
|
||||
eq(issues.status, "in_progress"),
|
||||
eq(issues.lastStatusDecisionId, statusDecisions.id),
|
||||
isNull(issues.assigneeUserId),
|
||||
isNull(issues.hiddenAt),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(nativeRunFinalizations.companyId, binding.companyId),
|
||||
eq(nativeRunFinalizations.issueId, binding.issueId),
|
||||
eq(nativeRunFinalizations.runId, binding.runId),
|
||||
eq(nativeRunFinalizations.phase, "committed"),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
if (!wait?.runStartedAt) return false;
|
||||
const [unanswered] = await db
|
||||
.select({ id: issueComments.id })
|
||||
.from(issueComments)
|
||||
.where(
|
||||
and(
|
||||
eq(issueComments.companyId, binding.companyId),
|
||||
eq(issueComments.issueId, binding.issueId),
|
||||
eq(issueComments.authorType, "user"),
|
||||
isNull(issueComments.createdByRunId),
|
||||
isNull(issueComments.deletedAt),
|
||||
sql`${issueComments.createdAt} > ${wait.runStartedAt.toISOString()}::timestamptz`,
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
return !unanswered;
|
||||
}
|
||||
@@ -168,8 +168,9 @@ describe("native status authority", () => {
|
||||
effects: [],
|
||||
});
|
||||
expect(arbitrate({ assessment: passive })).toMatchObject({
|
||||
reasonCode: "live_continuation_registered",
|
||||
effects: [expect.objectContaining({ kind: "enqueue_continuation" })],
|
||||
reasonCode: "response_wake_waiting_for_input",
|
||||
toStatus: "in_progress",
|
||||
effects: [],
|
||||
});
|
||||
expect(
|
||||
arbitrate({ assessment: passive, boardResponseWaitOrigin: true }),
|
||||
@@ -385,13 +386,9 @@ describe("native status authority", () => {
|
||||
}),
|
||||
).toEqual(
|
||||
expect.objectContaining({
|
||||
reasonCode: "live_continuation_registered",
|
||||
effects: [
|
||||
expect.objectContaining({
|
||||
kind: "enqueue_continuation",
|
||||
continuationKind: "response_wake",
|
||||
}),
|
||||
],
|
||||
statusAction: "in_progress",
|
||||
reasonCode: "response_wake_waiting_for_input",
|
||||
effects: [],
|
||||
}),
|
||||
);
|
||||
expect(arbitrate({ terminalState: "cancelled" })).toEqual(
|
||||
@@ -601,7 +598,7 @@ describe("native status authority", () => {
|
||||
expect.objectContaining({
|
||||
statusAction: "blocked",
|
||||
toStatus: "blocked",
|
||||
policyVersion: "phase6-v13",
|
||||
policyVersion: "phase6-v14",
|
||||
reasonCode: "current_track_blocker_waiting",
|
||||
unblockDescriptor: {
|
||||
owner: "board",
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
import type { NativeEvidenceAssessment } from "./evidence-classifier.js";
|
||||
import { NATIVE_MODEL_REJECTION_MESSAGE, NATIVE_PROVIDER_CAPACITY_MAX_RETRIES, NATIVE_PROVIDER_OVERLOADED_CODE, NATIVE_PROVIDER_OVERLOADED_MESSAGE } from "./native-provider-failure.js";
|
||||
|
||||
export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v13";
|
||||
export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v14";
|
||||
|
||||
/** The automatic continuation after a failed turn is used once per chain. */
|
||||
export const NATIVE_RUN_FAILED_RETRY_CAUSE = "native_run_failed";
|
||||
export const NATIVE_RESPONSE_WAKE_WAITING_REASON = "response_wake_waiting_for_input";
|
||||
export const NATIVE_RUN_FAILED_CONTINUATION_EXHAUSTED_ACTION =
|
||||
"The agent's turn failed again after Paperclip continued it automatically. The saved work is kept. Check the failed run, then Retry the task.";
|
||||
|
||||
@@ -748,6 +749,20 @@ export function arbitrateNativeStatus(input: {
|
||||
effects: [],
|
||||
};
|
||||
}
|
||||
if (
|
||||
input.assessment.reportedDisposition === "yielded" &&
|
||||
input.assessment.continuation?.kind === "response_wake" &&
|
||||
!unfinishedResponseWait
|
||||
) {
|
||||
return {
|
||||
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
|
||||
statusAction: "in_progress",
|
||||
toStatus: "in_progress",
|
||||
reasonCode: NATIVE_RESPONSE_WAKE_WAITING_REASON,
|
||||
unblockDescriptor: null,
|
||||
effects: [],
|
||||
};
|
||||
}
|
||||
if (
|
||||
input.assessment.reportedDisposition === "yielded" &&
|
||||
input.assessment.continuation &&
|
||||
|
||||
@@ -25,6 +25,7 @@ import { onboardingFirstTaskAwaitingUserSql } from "./onboarding-first-task-wait
|
||||
import { isAiAuthenticationBlocked } from "../ai-auth-failure.js";
|
||||
import { RESTRICTED_WORKSPACE_SOURCE_REASON } from "../workspace-source-restriction.js";
|
||||
import { hasCommittedNativePlanWait } from "../native-runtime/native-plan-wait.js";
|
||||
import { hasCommittedNativeResponseWakeWait } from "../native-runtime/native-response-wake-wait.js";
|
||||
import { listExternalReviewWaits } from "../native-runtime/external-review-wait.js";
|
||||
import { isNativeWorkspaceExportRepairCause } from "@paperclipai/shared";
|
||||
import { pendingExternalConversationSettlement, settleSlackConversation } from "../slack-conversation-lifecycle.js";
|
||||
@@ -1265,6 +1266,7 @@ export function recoveryService(
|
||||
agentId: latestRun.agentId,
|
||||
};
|
||||
if (await hasCommittedNativePlanWait(db, binding)) return true;
|
||||
if (issue.status === "in_progress" && await hasCommittedNativeResponseWakeWait(db, binding)) return true;
|
||||
const [receipt] = await db
|
||||
.select({
|
||||
run: heartbeatRuns,
|
||||
@@ -1467,6 +1469,8 @@ export function recoveryService(
|
||||
(await listExternalReviewWaits(db, issue.companyId, [issue.id])).length > 0) return true;
|
||||
if (issue.status === "in_progress" && latestRun?.status === "succeeded" && latestRun.agentId === issue.assigneeAgentId &&
|
||||
await hasCommittedNativePlanWait(db, { companyId: issue.companyId, issueId: issue.id, runId: latestRun.id, agentId: latestRun.agentId })) return true;
|
||||
if (issue.status === "in_progress" && latestRun?.status === "succeeded" && latestRun.agentId === issue.assigneeAgentId &&
|
||||
await hasCommittedNativeResponseWakeWait(db, { companyId: issue.companyId, issueId: issue.id, runId: latestRun.id, agentId: latestRun.agentId })) return true;
|
||||
if (
|
||||
issue.status === "in_progress" &&
|
||||
latestRun?.status === "succeeded" &&
|
||||
|
||||
Reference in new issue
Block a user