From edd00526e7659fda60716331fc978f74154c1d34 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:40:44 -0600 Subject: [PATCH] fix(native): park a response_wake with no admitted response instead of self-waking (#20935) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- doc/architecture/native-status-arbitration.md | 22 ++++- .../heartbeat-issue-rewake-throttle.test.ts | 86 +++++++++++++++++++ .../heartbeat-process-recovery.test.ts | 84 ++++++++++++++++-- server/src/services/execution-projection.ts | 23 +++++ server/src/services/heartbeat/queue.ts | 31 ++++++- server/src/services/issue-rewake-throttle.ts | 4 + .../native-response-wake-wait.ts | 85 ++++++++++++++++++ .../native-runtime/status-arbiter.test.ts | 17 ++-- .../services/native-runtime/status-arbiter.ts | 17 +++- server/src/services/recovery/service.ts | 4 + tests/restart-chaos/README.md | 14 ++- tests/restart-chaos/ledger.ts | 2 +- tests/restart-chaos/types.ts | 2 +- tests/restart-chaos/universal-audit.ts | 4 +- tests/restart-chaos/universal.test.ts | 17 +++- tests/restart-chaos/universal.ts | 17 +++- 16 files changed, 400 insertions(+), 29 deletions(-) create mode 100644 server/src/services/native-runtime/native-response-wake-wait.ts diff --git a/doc/architecture/native-status-arbitration.md b/doc/architecture/native-status-arbitration.md index 7fa6d5782ed..91cd4d16621 100644 --- a/doc/architecture/native-status-arbitration.md +++ b/doc/architecture/native-status-arbitration.md @@ -141,7 +141,8 @@ conditions before model disposition: | Runner reports `needs_review` without a decision, or an incomplete completion claim | Keep work with the agent | No automatic human approval; at most one corrective continuation, then a visible recovery action | | Runner reports a task-wide blocker | `blocked` | Persist blocker owner and unblock action | | Runner reports a current-track blocker | `in_progress` | Enqueue another productive track | -| Runner reports `yielded` with a valid continuation | `in_progress` | Enqueue the declared continuation | +| Runner reports `yielded` / `response_wake` with no admitted response source | `in_progress` | `response_wake_waiting_for_input`: no wake; the task shows "Waiting for a reply" until a comment, answer or status change wakes it | +| Runner reports `yielded` with another valid continuation | `in_progress` | Enqueue the declared continuation | | Ordinary task reports `response_wake` and blocking remaining work, without a real wait | `in_progress` | Use the bounded incomplete-work continuation, then a visible recovery error if repair was already used | | Completion evidence is incomplete and continuation is forbidden | Preserve | Record a finalization error and named next action | | Completion evidence is otherwise incomplete | `in_progress` | Enqueue a bounded, idempotent continuation | @@ -171,6 +172,25 @@ old response. Recovery no longer treats an old committed passive wait as healthy when its report admits blocking work and its Board source is still current. Existing recovery gates continue to protect pauses, budgets, and ownership. +A `response_wake` asks for the next response; it is never a request to run +again now. Only a directly admitted Board comment, an authorized external chat, +an accepted plan or a governance gate gives it a specific wait record. Every +other `response_wake` parks the task as `response_wake_waiting_for_input`. +Recovery treats that applied decision as a durable wait while it is the issue's +latest decision and no user comment arrived after the run started. A comment, +an interaction answer or a status change by someone else wakes the agent through +its normal path. + +Declared continuations that do enqueue a wake (`issue_status_changed` with a +`continuationKind`) are throttle candidates in the issue rewake throttle. After +two consecutive successful runs with no issue-visible progress, the wake waits +for the escalating cooldown. The native status dispatcher keeps the intent +queued and dispatches it again at the cooldown deadline; it is never dropped. +The native status committer's own `issue.updated` record does not count as +progress or new input. The "Waiting for a reply" execution label shows only +while the parked decision is still the issue's latest decision and the issue +is `in_progress`. + ## Transactional decision commit `commitNativeStatusDecision()` applies the decision against authoritative issue diff --git a/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts b/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts index f7a7ccd0bc5..2d34f9a201c 100644 --- a/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts +++ b/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts @@ -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) { + 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(); diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index d3ae160fe9a..edf299aa4ed 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -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); diff --git a/server/src/services/execution-projection.ts b/server/src/services/execution-projection.ts index 7140a84c6ec..1bfe7e139b9 100644 --- a/server/src/services/execution-projection.ts +++ b/server/src/services/execution-projection.ts @@ -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>`jsonb_build_object( + 'finalizationPhase', ${heartbeatRuns.resultJson}->'finalizationPhase', + 'finalizationReasonCode', ${heartbeatRuns.resultJson}->'finalizationReasonCode')`, }; type Run = Pick; 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)) diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 18f9c8b4b28..509b8c6bed7 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -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 diff --git a/server/src/services/issue-rewake-throttle.ts b/server/src/services/issue-rewake-throttle.ts index 52cf267710e..9b47dc7baeb 100644 --- a/server/src/services/issue-rewake-throttle.ts +++ b/server/src/services/issue-rewake-throttle.ts @@ -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); } diff --git a/server/src/services/native-runtime/native-response-wake-wait.ts b/server/src/services/native-runtime/native-response-wake-wait.ts new file mode 100644 index 00000000000..af81d2d46f8 --- /dev/null +++ b/server/src/services/native-runtime/native-response-wake-wait.ts @@ -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 { + 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; +} diff --git a/server/src/services/native-runtime/status-arbiter.test.ts b/server/src/services/native-runtime/status-arbiter.test.ts index 0eb079e508d..0e57f8f1b7b 100644 --- a/server/src/services/native-runtime/status-arbiter.test.ts +++ b/server/src/services/native-runtime/status-arbiter.test.ts @@ -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", diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index a7fb4c0baa1..594f7338b21 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -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 && diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index b4700a75722..8fe9f93a46f 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -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" && diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 178c74b5ecb..6e0da141e08 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -1148,9 +1148,9 @@ that lock the run row still fail under the same contention on master. The conten (`contention_produced`) fired only on tasks whose runs failed before any usage was reported; a run that finished produced 208 log events and 20 receipts, well above the 40 and 6 thresholds. -## Universal checks (U1–U4) +## Universal checks (U1–U5) -Every task of every wave is also graded on four scenario-independent rules (`universal.ts`), whatever +Every task of every wave is also graded on five scenario-independent rules (`universal.ts`), whatever its workload. They catch a stuck or misleading task that no scenario anticipated. | Rule | Fails when | Bound (env override) | @@ -1159,8 +1159,9 @@ its workload. They catch a stuck or misleading task that no scenario anticipated | U2 retry loop | more than N consecutive failed runs with the same status and code while the issue status stays the same | 3 (`RESTART_CHAOS_U2_SAME_FAILURE_RUNS`) | | U3 diagnosable | a failed run has no code or a generic one (`adapter_failed`, `unknown`, `error`, ...) and no recorded `failureCause` (a bounded code such as `40001`, `exit_code_137` or `provider_http_429`; a bare `Error` or `unknown_error` does not count), its stored error is empty or names only a bare cause (`last cause: Error`), or a recovery action's cause or next action is empty or bare | none | | U4 stale notice | a Retry or failed notice stays on a task after its latest run succeeded, longer than the bound or at rest | 60 s (`RESTART_CHAOS_U4_STALE_SECONDS`) | +| U5 run budget | one task created more runs than the bound | 10 (`RESTART_CHAOS_U5_MAX_RUNS`) | -U1, U2 and U4 fail the task (signature `U:...`). U3 is a warning: it is reported in each grade +U1, U2, U4 and U5 fail the task (signature `U:...`). U3 is a warning: it is reported in each grade (`grade.universal`), the ledger (`universal`, `universalWarnings`) and the wave's `SUMMARY.md`, and never fails a task. U3 reads run error text and recovery-action causes from the wave database at wave end, because the board's run list hides most error text. @@ -1179,6 +1180,13 @@ node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/universal-audit.ts ; universalWarnings?: string[]; /** DF1: faulted saves, the disk-full rest with a Retry offered, and whether that Retry worked. */ diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index 5c5f3333dab..016b119c853 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -478,7 +478,7 @@ export interface TaskGrade { crossingFaults: string[]; rules: Record; T: OracleResult; - /** Universal checks U1–U4 (universal.ts); absent on grades written before them. */ + /** Universal checks U1–U5 (universal.ts); absent on grades written before them. */ universal?: import("./universal.js").UniversalGrade; pass: boolean; signature: string | null; diff --git a/tests/restart-chaos/universal-audit.ts b/tests/restart-chaos/universal-audit.ts index 4f9520c5adb..872716efdd3 100644 --- a/tests/restart-chaos/universal-audit.ts +++ b/tests/restart-chaos/universal-audit.ts @@ -1,5 +1,5 @@ /** - * Run the universal checks U1–U4 over saved wave evidence, to measure false + * Run the universal checks U1–U5 over saved wave evidence, to measure false * positives before a rule is enforced. * * node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/universal-audit.ts [--json] [...] @@ -29,7 +29,7 @@ function percentiles(values: number[]) { return { n: sorted.length, p50: at(0.5), p95: at(0.95), p99: at(0.99), max: sorted[sorted.length - 1] ?? 0 }; } -const RULES: UniversalId[] = ["U1", "U2", "U3", "U4"]; +const RULES: UniversalId[] = ["U1", "U2", "U3", "U4", "U5"]; export function auditWaves(roots: readonly string[], bounds: UniversalBounds = DEFAULT_UNIVERSAL_BOUNDS): AuditReport { const report: AuditReport = { waves: 0, tasks: 0, knownGoodTasks: 0, knownGoodWaves: 0, diff --git a/tests/restart-chaos/universal.test.ts b/tests/restart-chaos/universal.test.ts index 87c84587ba3..84ad5eb3093 100644 --- a/tests/restart-chaos/universal.test.ts +++ b/tests/restart-chaos/universal.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from "vitest"; import { gradeTask } from "./grader.js"; import type { OracleArtifacts, RunSample, TaskRecord, TaskSample } from "./types.js"; -import { bareCauseMessage, DEFAULT_UNIVERSAL_BOUNDS as BOUNDS, diagnosableFailureCause, gradeU1, gradeU2, gradeU3, gradeU4, gradeUniversal, universalBounds } from "./universal.js"; +import { bareCauseMessage, DEFAULT_UNIVERSAL_BOUNDS as BOUNDS, diagnosableFailureCause, gradeU1, gradeU2, gradeU3, gradeU4, gradeU5, gradeUniversal, universalBounds } from "./universal.js"; import { workloadDefinitions } from "./workloads.js"; const T0 = Date.parse("2026-11-09T12:00:00Z"); @@ -177,6 +177,21 @@ describe("U4 stale notices", () => { }); }); +describe("U5 run budget", () => { + it("fails a task that keeps starting runs (the response_wake loop shape)", () => { + const runs = Array.from({ length: 65 }, (_, index) => run(`r${index}`, "succeeded", index * 12 * SEC, index * 12 * SEC + 10 * SEC)); + const samples = [sample(800 * SEC, "in_progress", runs)]; + expect(gradeU5(record(samples, { restAt: null, finalStatus: null, timedOut: true }), BOUNDS)) + .toMatchObject({ verdict: "fail", detail: "run_budget_exceeded 65 runs > 10", metrics: { runs: 65 } }); + }); + it("passes a task within the budget and reads the bound from the environment", () => { + const runs = Array.from({ length: 5 }, (_, index) => run(`r${index}`, "succeeded", index * 20 * SEC, index * 20 * SEC + 10 * SEC)); + expect(gradeU5(record([sample(120 * SEC, "done", runs)]), BOUNDS).verdict).toBe("pass"); + expect(universalBounds({ RESTART_CHAOS_U5_MAX_RUNS: "4" }).maxRuns).toBe(4); + expect(gradeU5(record([sample(120 * SEC, "done", runs)]), { ...BOUNDS, maxRuns: 4 }).verdict).toBe("fail"); + }); +}); + describe("universal checks in the task grade", () => { const artifacts: OracleArtifacts = { issueStatus: "done", documents: { brief: "CHAOS-abcd1234\n- one\n- two\n- three\n" }, agentComments: [], files: {}, commands: {}, workspacePath: null }; diff --git a/tests/restart-chaos/universal.ts b/tests/restart-chaos/universal.ts index d48f44a78e9..8306aa081c1 100644 --- a/tests/restart-chaos/universal.ts +++ b/tests/restart-chaos/universal.ts @@ -1,5 +1,5 @@ /** - * Universal checks U1–U4: scenario-independent rules graded on every task of + * Universal checks U1–U5: scenario-independent rules graded on every task of * every wave, whatever its workload. Pure functions over observer samples (no * I/O), so they also run over saved evidence (`universal-audit.ts`). * @@ -14,7 +14,7 @@ */ import type { RunSample, TaskRecord, TaskSample } from "./types.js"; -export type UniversalId = "U1" | "U2" | "U3" | "U4"; +export type UniversalId = "U1" | "U2" | "U3" | "U4" | "U5"; export interface UniversalResult { verdict: "pass" | "fail" | "n/a"; detail: string; enforced: boolean; /** Measurements behind the verdict (milliseconds or counts), for bound tuning. */ metrics?: Record } @@ -31,6 +31,7 @@ export interface UniversalBounds { sameFailureRuns: number; /** U4: longest a failure or Retry notice may remain after a later run succeeded. */ staleNoticeMs: number; + maxRuns: number; } /** @@ -45,10 +46,11 @@ export const DEFAULT_UNIVERSAL_BOUNDS: UniversalBounds = { silentRunMs: 3 * 60_000, sameFailureRuns: 3, staleNoticeMs: 60_000, + maxRuns: 10, }; /** Rules that fail a task. The rest are warnings until the audit shows no false positive on known-good waves. */ -export const ENFORCED_UNIVERSAL: ReadonlySet = new Set(["U1", "U2", "U4"]); +export const ENFORCED_UNIVERSAL: ReadonlySet = new Set(["U1", "U2", "U4", "U5"]); export function universalBounds(env: NodeJS.ProcessEnv = process.env): UniversalBounds { const seconds = (name: string, fallback: number) => { @@ -64,6 +66,7 @@ export function universalBounds(env: NodeJS.ProcessEnv = process.env): Universal silentRunMs: DEFAULT_UNIVERSAL_BOUNDS.silentRunMs, sameFailureRuns: Math.round(seconds("RESTART_CHAOS_U2_SAME_FAILURE_RUNS", DEFAULT_UNIVERSAL_BOUNDS.sameFailureRuns)), staleNoticeMs: seconds("RESTART_CHAOS_U4_STALE_SECONDS", DEFAULT_UNIVERSAL_BOUNDS.staleNoticeMs / 1000) * 1000, + maxRuns: Math.round(seconds("RESTART_CHAOS_U5_MAX_RUNS", DEFAULT_UNIVERSAL_BOUNDS.maxRuns)), }; } @@ -293,6 +296,13 @@ export function gradeU4(record: TaskRecord, bounds: UniversalBounds): Omit { + const runs = finalRuns(record).size; + return runs > bounds.maxRuns + ? { verdict: "fail", metrics: { runs }, detail: `run_budget_exceeded ${runs} runs > ${bounds.maxRuns}` } + : { verdict: "pass", metrics: { runs }, detail: `${runs} runs` }; +} + export function gradeUniversal(record: TaskRecord, options: { bounds?: UniversalBounds; db?: UniversalDbEvidence; enforced?: ReadonlySet } = {}): UniversalGrade { const bounds = options.bounds ?? universalBounds(); @@ -302,6 +312,7 @@ export function gradeUniversal(record: TaskRecord, options: { bounds?: Universal U2: gradeU2(record, bounds), U3: gradeU3(record, options.db), U4: gradeU4(record, bounds), + U5: gradeU5(record, bounds), }; return Object.fromEntries(Object.entries(results).map(([id, result]) => [id, { ...result, enforced: enforced.has(id as UniversalId) }])) as UniversalGrade; }