fix(heartbeat): throttle redundant issue re-wakes (#9470)
## Thinking Path > - Paperclip is the open source control plane people use to coordinate AI agents and their work > - Heartbeat admission decides when an agent should start another adapter session for an issue > - After process-loss recovery, assignment pollers and reconcilers can repeatedly request another wake while the issue remains `in_progress` > - When the preceding runs succeeded without issue-visible progress, those event-free wakes provide no new information but still pay the full cost of an adapter session > - Existing liveness evidence is too broad for this case because workspace tool calls can make a run look active without moving the issue > - This pull request adds an issue-scoped admission throttle for consecutive no-progress re-wakes while preserving every wake that carries new information or recovery intent > - The benefit is bounded recovery cost without delaying comments, operator actions, failures, or other meaningful events ## Linked Issues or Issue Description No public GitHub issue exists for this bug. **What happened?** After a process died, external wake drivers could re-wake the same agent for the same `in_progress` issue every few seconds. Each succeeded run that produced no issue-visible progress could be followed by another full adapter session despite no new issue input. In the observed recovery smoke, one recovery consumed 25 sessions and 2.4× the direct-run cost. **Expected behavior** Repeated event-free re-wakes should back off after consecutive successful runs produce no issue-visible progress. Any new information, explicit operator intent, or failed-run recovery should continue immediately. **Steps to reproduce** 1. Start an issue heartbeat and simulate process loss while the issue remains `in_progress`. 2. Allow assignment/reconciliation drivers to request repeated event-free wakes for the same agent and issue. 3. Complete each follow-up run successfully without adding a comment, issue mutation, document, work product, interaction, or continuation. 4. Observe repeated adapter sessions starting every few seconds without new issue input. **Environment** - Version: reproduced on `master` before this change - Deployment: local development, built from source - Adapter scope: core bug; not adapter-specific - Database: reproduced and tested with embedded Postgres ## What Changed - Add a pure issue re-wake throttle that detects consecutive succeeded runs without issue-visible progress and applies a 120-second exponential cooldown capped at 30 minutes. - Gate event-free `enqueueWakeup` requests and return the explicit skip reason `issue_rewake_throttled` while the cooldown is active. - Always bypass throttling for comment wakes, new issue activity, explicit resumes, `forceFreshSession`, event-shaped reasons, and post-failure recovery. - Add focused pure unit coverage and database-backed heartbeat admission coverage for throttle and bypass behavior. ## Verification - `cd server && pnpm vitest run src/__tests__/issue-rewake-throttle.test.ts` — 12 passed. - `cd server && pnpm vitest run src/__tests__/heartbeat-issue-rewake-throttle.test.ts` — 6 passed with embedded Postgres. - `cd server && pnpm run typecheck` — passed. - Neighbor suites previously verified: `heartbeat-dependency-scheduling`, `heartbeat-process-recovery`, `run-continuations`, `heartbeat-issue-liveness-escalation`, `recovery-stale-issue-lock-sweep`, and `heartbeat-comment-wake-batching` — 131 tests passed. ## Risks - A progress classifier that is too narrow could defer a legitimate event-free poll; the cooldown is bounded and new issue activity bypasses it immediately. - A progress classifier that is too broad could allow the original heartbeat storm; tests intentionally distinguish issue-visible mutations from workspace-only activity. - Low compatibility risk: no schema, API contract, or migration changes. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex coding agent. The runtime does not expose the exact underlying model ID or context-window size; reasoning, terminal tool use, code inspection, GitHub CLI access, and test execution were enabled. ## 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 public PR branch name describes the change 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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
9e7e84e3fe
commit
5dff52631d
|
|
@ -0,0 +1,329 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { and, desc, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agentRuntimeState,
|
||||
agentWakeupRequests,
|
||||
agents,
|
||||
companies,
|
||||
companySkills,
|
||||
createDb,
|
||||
environmentLeases,
|
||||
environments,
|
||||
executionWorkspaces,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueComments,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import { heartbeatService } from "../services/heartbeat.ts";
|
||||
import { runningProcesses } from "../adapters/index.ts";
|
||||
|
||||
const mockAdapterExecute = vi.hoisted(() =>
|
||||
vi.fn(async () => ({
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
errorMessage: null,
|
||||
summary: "Issue rewake throttle test run.",
|
||||
provider: "test",
|
||||
model: "test-model",
|
||||
})),
|
||||
);
|
||||
|
||||
vi.mock("../adapters/index.ts", async () => {
|
||||
const actual = await vi.importActual<typeof import("../adapters/index.ts")>("../adapters/index.ts");
|
||||
return {
|
||||
...actual,
|
||||
getServerAdapter: vi.fn(() => ({
|
||||
supportsLocalAgentJwt: false,
|
||||
execute: mockAdapterExecute,
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
if (!embeddedPostgresSupport.supported) {
|
||||
console.warn(
|
||||
`Skipping embedded Postgres issue rewake throttle tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
|
||||
);
|
||||
}
|
||||
|
||||
describeEmbeddedPostgres("heartbeat issue rewake throttle", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let heartbeat!: ReturnType<typeof heartbeatService>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-issue-rewake-throttle-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
heartbeat = heartbeatService(db);
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
runningProcesses.clear();
|
||||
for (let attempt = 0; attempt < 100; attempt += 1) {
|
||||
const runs = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns);
|
||||
if (!runs.some((run) => run.status === "queued" || run.status === "running")) break;
|
||||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||||
}
|
||||
await db.delete(environmentLeases);
|
||||
await db.delete(issueComments);
|
||||
await db.delete(issues);
|
||||
await db.delete(heartbeatRunEvents);
|
||||
await db.delete(activityLog);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agentWakeupRequests);
|
||||
await db.delete(agentRuntimeState);
|
||||
await db.delete(agents);
|
||||
await db.delete(environments);
|
||||
await db.delete(executionWorkspaces);
|
||||
await db.delete(companySkills);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
async function seedCompanyAgentIssue() {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
defaultResponsibleUserId: "responsible-user",
|
||||
});
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "CodexCoder",
|
||||
role: "engineer",
|
||||
status: "active",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {
|
||||
heartbeat: {
|
||||
wakeOnDemand: true,
|
||||
maxConcurrentRuns: 1,
|
||||
},
|
||||
},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Interrupted import mission",
|
||||
status: "in_progress",
|
||||
priority: "medium",
|
||||
assigneeAgentId: agentId,
|
||||
responsibleUserId: "responsible-user",
|
||||
});
|
||||
|
||||
return { companyId, agentId, issueId };
|
||||
}
|
||||
|
||||
async function seedTerminalRun(input: {
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
issueId: string;
|
||||
status?: string;
|
||||
finishedSecondsAgo: number;
|
||||
startedSecondsAgo?: number;
|
||||
}) {
|
||||
const runId = randomUUID();
|
||||
const finishedAt = new Date(Date.now() - input.finishedSecondsAgo * 1000);
|
||||
const startedAt = input.startedSecondsAgo === undefined
|
||||
? new Date(finishedAt.getTime() - 5_000)
|
||||
: new Date(Date.now() - input.startedSecondsAgo * 1000);
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId: input.companyId,
|
||||
agentId: input.agentId,
|
||||
invocationSource: "assignment",
|
||||
status: input.status ?? "succeeded",
|
||||
responsibleUserId: "responsible-user",
|
||||
createdAt: startedAt,
|
||||
startedAt,
|
||||
finishedAt,
|
||||
contextSnapshot: { issueId: input.issueId, wakeReason: "issue_assigned" },
|
||||
});
|
||||
return runId;
|
||||
}
|
||||
|
||||
function assignmentWake(agentId: string, issueId: string) {
|
||||
return heartbeat.wakeup(agentId, {
|
||||
source: "assignment",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_assigned",
|
||||
payload: { issueId },
|
||||
contextSnapshot: { issueId, wakeReason: "issue_assigned" },
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: "test",
|
||||
});
|
||||
}
|
||||
|
||||
async function latestWakeRequest(agentId: string) {
|
||||
return db
|
||||
.select({
|
||||
status: agentWakeupRequests.status,
|
||||
reason: agentWakeupRequests.reason,
|
||||
payload: agentWakeupRequests.payload,
|
||||
})
|
||||
.from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.agentId, agentId))
|
||||
.orderBy(desc(agentWakeupRequests.requestedAt))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
it("skips event-free re-wakes after consecutive no-progress runs and admits them again on new input", async () => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 40 });
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 10 });
|
||||
|
||||
const throttledWake = await assignmentWake(agentId, issueId);
|
||||
expect(throttledWake).toBeNull();
|
||||
|
||||
const skipped = await latestWakeRequest(agentId);
|
||||
expect(skipped?.status).toBe("skipped");
|
||||
expect(skipped?.reason).toBe("issue_rewake_throttled");
|
||||
const heartbeatSkip = (skipped?.payload as Record<string, unknown> | null)?.heartbeatSkip as
|
||||
| Record<string, unknown>
|
||||
| undefined;
|
||||
expect(heartbeatSkip?.noProgressStreak).toBe(2);
|
||||
expect(typeof heartbeatSkip?.nextAllowedAt).toBe("string");
|
||||
|
||||
const runCount = await db
|
||||
.select({ count: sql<number>`count(*)::int` })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.companyId, companyId))
|
||||
.then((rows) => rows[0]?.count ?? 0);
|
||||
expect(runCount).toBe(2);
|
||||
|
||||
// A board comment on the issue is new input: the next event-free wake is
|
||||
// admitted even though the streak has not been broken by a run.
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: "board-user",
|
||||
action: "issue.comment_added",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
});
|
||||
|
||||
const admittedWake = await assignmentWake(agentId, issueId);
|
||||
expect(admittedWake).not.toBeNull();
|
||||
});
|
||||
|
||||
it("does not throttle comment-driven wakes even during a no-progress streak", async () => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 40 });
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 10 });
|
||||
|
||||
const commentWake = await heartbeat.wakeup(agentId, {
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_commented",
|
||||
payload: { issueId, commentId: randomUUID() },
|
||||
contextSnapshot: { issueId, wakeReason: "issue_commented" },
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: "test",
|
||||
});
|
||||
expect(commentWake).not.toBeNull();
|
||||
});
|
||||
|
||||
it("does not throttle the wake that follows a failed run", async () => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 70 });
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 40 });
|
||||
await seedTerminalRun({ companyId, agentId, issueId, status: "failed", finishedSecondsAgo: 10 });
|
||||
|
||||
const recoveryWake = await assignmentWake(agentId, issueId);
|
||||
expect(recoveryWake).not.toBeNull();
|
||||
});
|
||||
|
||||
it("does not throttle when a recent run produced 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 db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "agent",
|
||||
actorId: agentId,
|
||||
agentId,
|
||||
runId: progressRunId,
|
||||
action: "issue.comment_added",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
createdAt: new Date(Date.now() - 11_000),
|
||||
});
|
||||
|
||||
const wake = await assignmentWake(agentId, issueId);
|
||||
expect(wake).not.toBeNull();
|
||||
});
|
||||
|
||||
it("does not count progress on another issue toward the current issue", async () => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
const otherIssueId = randomUUID();
|
||||
await db.insert(issues).values({
|
||||
id: otherIssueId,
|
||||
companyId,
|
||||
title: "Related follow-up",
|
||||
status: "in_progress",
|
||||
priority: "medium",
|
||||
assigneeAgentId: agentId,
|
||||
responsibleUserId: "responsible-user",
|
||||
});
|
||||
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 40 });
|
||||
const progressRunId = await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 10 });
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "agent",
|
||||
actorId: agentId,
|
||||
agentId,
|
||||
runId: progressRunId,
|
||||
action: "issue.comment_added",
|
||||
entityType: "issue",
|
||||
entityId: otherIssueId,
|
||||
createdAt: new Date(Date.now() - 11_000),
|
||||
});
|
||||
|
||||
const wake = await assignmentWake(agentId, issueId);
|
||||
expect(wake).toBeNull();
|
||||
expect((await latestWakeRequest(agentId))?.reason).toBe("issue_rewake_throttled");
|
||||
});
|
||||
|
||||
it("counts a long-running session that finished inside the lookback window", async () => {
|
||||
const { companyId, agentId, issueId } = await seedCompanyAgentIssue();
|
||||
|
||||
await seedTerminalRun({
|
||||
companyId,
|
||||
agentId,
|
||||
issueId,
|
||||
finishedSecondsAgo: 40,
|
||||
startedSecondsAgo: 7 * 60 * 60,
|
||||
});
|
||||
await seedTerminalRun({ companyId, agentId, issueId, finishedSecondsAgo: 10 });
|
||||
|
||||
const wake = await assignmentWake(agentId, issueId);
|
||||
expect(wake).toBeNull();
|
||||
expect((await latestWakeRequest(agentId))?.reason).toBe("issue_rewake_throttled");
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,184 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
ISSUE_REWAKE_BASE_COOLDOWN_MS,
|
||||
ISSUE_REWAKE_MAX_COOLDOWN_MS,
|
||||
ISSUE_REWAKE_NO_PROGRESS_THRESHOLD,
|
||||
computeIssueRewakeCooldownMs,
|
||||
evaluateIssueRewakeThrottle,
|
||||
isThrottleCandidateIssueRewake,
|
||||
} from "../services/issue-rewake-throttle.ts";
|
||||
|
||||
const NOW = new Date("2026-07-12T18:14:00.000Z");
|
||||
|
||||
function runSample(input: {
|
||||
id: string;
|
||||
status?: string;
|
||||
finishedSecondsAgo: number;
|
||||
}) {
|
||||
return {
|
||||
id: input.id,
|
||||
status: input.status ?? "succeeded",
|
||||
finishedAt: new Date(NOW.getTime() - input.finishedSecondsAgo * 1000),
|
||||
};
|
||||
}
|
||||
|
||||
describe("isThrottleCandidateIssueRewake", () => {
|
||||
const base = {
|
||||
reason: "issue_assigned",
|
||||
wakeCommentId: null,
|
||||
forceFreshSession: false,
|
||||
hasExplicitResume: false,
|
||||
};
|
||||
|
||||
it("throttles state-poll reasons and reason-less invokes", () => {
|
||||
expect(isThrottleCandidateIssueRewake(base)).toBe(true);
|
||||
expect(isThrottleCandidateIssueRewake({ ...base, reason: null })).toBe(true);
|
||||
expect(isThrottleCandidateIssueRewake({ ...base, reason: "issue_continuation_needed" })).toBe(true);
|
||||
expect(isThrottleCandidateIssueRewake({ ...base, reason: "issue_assignment_recovery" })).toBe(true);
|
||||
expect(isThrottleCandidateIssueRewake({ ...base, reason: "issue_graph_liveness_backstop" })).toBe(true);
|
||||
});
|
||||
|
||||
it("never throttles wakes that carry new information or an explicit escalation", () => {
|
||||
expect(isThrottleCandidateIssueRewake({ ...base, wakeCommentId: "comment-1" })).toBe(false);
|
||||
expect(isThrottleCandidateIssueRewake({ ...base, forceFreshSession: true })).toBe(false);
|
||||
expect(isThrottleCandidateIssueRewake({ ...base, hasExplicitResume: true })).toBe(false);
|
||||
});
|
||||
|
||||
it("passes event-shaped wake reasons through", () => {
|
||||
for (const reason of [
|
||||
"issue_commented",
|
||||
"issue_comment_mentioned",
|
||||
"issue_blockers_resolved",
|
||||
"issue_children_completed",
|
||||
"issue_monitor_due",
|
||||
"process_lost_retry",
|
||||
"run_liveness_continuation",
|
||||
]) {
|
||||
expect(isThrottleCandidateIssueRewake({ ...base, reason })).toBe(false);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("computeIssueRewakeCooldownMs", () => {
|
||||
it("starts at the base cooldown and doubles per extra no-progress run, capped", () => {
|
||||
expect(computeIssueRewakeCooldownMs(ISSUE_REWAKE_NO_PROGRESS_THRESHOLD)).toBe(ISSUE_REWAKE_BASE_COOLDOWN_MS);
|
||||
expect(computeIssueRewakeCooldownMs(ISSUE_REWAKE_NO_PROGRESS_THRESHOLD + 1)).toBe(ISSUE_REWAKE_BASE_COOLDOWN_MS * 2);
|
||||
expect(computeIssueRewakeCooldownMs(ISSUE_REWAKE_NO_PROGRESS_THRESHOLD + 3)).toBe(ISSUE_REWAKE_BASE_COOLDOWN_MS * 8);
|
||||
expect(computeIssueRewakeCooldownMs(100)).toBe(ISSUE_REWAKE_MAX_COOLDOWN_MS);
|
||||
});
|
||||
});
|
||||
|
||||
describe("evaluateIssueRewakeThrottle", () => {
|
||||
it("allows when there is no run history", () => {
|
||||
expect(
|
||||
evaluateIssueRewakeThrottle({
|
||||
now: NOW,
|
||||
recentTerminalRuns: [],
|
||||
runIdsWithIssueProgress: new Set(),
|
||||
hasNewIssueInputSinceLastRun: false,
|
||||
}),
|
||||
).toEqual({ blocked: false, noProgressStreak: 0 });
|
||||
});
|
||||
|
||||
it("allows below the no-progress threshold", () => {
|
||||
const decision = evaluateIssueRewakeThrottle({
|
||||
now: NOW,
|
||||
recentTerminalRuns: [runSample({ id: "r1", finishedSecondsAgo: 10 })],
|
||||
runIdsWithIssueProgress: new Set(),
|
||||
hasNewIssueInputSinceLastRun: false,
|
||||
});
|
||||
expect(decision).toEqual({ blocked: false, noProgressStreak: 1 });
|
||||
});
|
||||
|
||||
it("blocks inside the cooldown once the streak reaches the threshold", () => {
|
||||
const decision = evaluateIssueRewakeThrottle({
|
||||
now: NOW,
|
||||
recentTerminalRuns: [
|
||||
runSample({ id: "r2", finishedSecondsAgo: 10 }),
|
||||
runSample({ id: "r1", finishedSecondsAgo: 40 }),
|
||||
],
|
||||
runIdsWithIssueProgress: new Set(),
|
||||
hasNewIssueInputSinceLastRun: false,
|
||||
});
|
||||
expect(decision.blocked).toBe(true);
|
||||
if (decision.blocked) {
|
||||
expect(decision.noProgressStreak).toBe(2);
|
||||
expect(decision.cooldownMs).toBe(ISSUE_REWAKE_BASE_COOLDOWN_MS);
|
||||
expect(decision.nextAllowedAt.getTime()).toBe(
|
||||
NOW.getTime() - 10_000 + ISSUE_REWAKE_BASE_COOLDOWN_MS,
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
it("allows again after the cooldown elapses", () => {
|
||||
const decision = evaluateIssueRewakeThrottle({
|
||||
now: NOW,
|
||||
recentTerminalRuns: [
|
||||
runSample({ id: "r2", finishedSecondsAgo: ISSUE_REWAKE_BASE_COOLDOWN_MS / 1000 + 1 }),
|
||||
runSample({ id: "r1", finishedSecondsAgo: ISSUE_REWAKE_BASE_COOLDOWN_MS / 1000 + 30 }),
|
||||
],
|
||||
runIdsWithIssueProgress: new Set(),
|
||||
hasNewIssueInputSinceLastRun: false,
|
||||
});
|
||||
expect(decision).toEqual({ blocked: false, noProgressStreak: 2 });
|
||||
});
|
||||
|
||||
it("escalates the cooldown as the streak grows", () => {
|
||||
const decision = evaluateIssueRewakeThrottle({
|
||||
now: NOW,
|
||||
recentTerminalRuns: [
|
||||
runSample({ id: "r4", finishedSecondsAgo: 10 }),
|
||||
runSample({ id: "r3", finishedSecondsAgo: 30 }),
|
||||
runSample({ id: "r2", finishedSecondsAgo: 60 }),
|
||||
runSample({ id: "r1", finishedSecondsAgo: 90 }),
|
||||
],
|
||||
runIdsWithIssueProgress: new Set(),
|
||||
hasNewIssueInputSinceLastRun: false,
|
||||
});
|
||||
expect(decision.blocked).toBe(true);
|
||||
if (decision.blocked) {
|
||||
expect(decision.noProgressStreak).toBe(4);
|
||||
expect(decision.cooldownMs).toBe(ISSUE_REWAKE_BASE_COOLDOWN_MS * 4);
|
||||
}
|
||||
});
|
||||
|
||||
it("resets at the most recent run with issue-visible progress", () => {
|
||||
const decision = evaluateIssueRewakeThrottle({
|
||||
now: NOW,
|
||||
recentTerminalRuns: [
|
||||
runSample({ id: "r3", finishedSecondsAgo: 10 }),
|
||||
runSample({ id: "r2", finishedSecondsAgo: 40 }),
|
||||
runSample({ id: "r1", finishedSecondsAgo: 70 }),
|
||||
],
|
||||
runIdsWithIssueProgress: new Set(["r2"]),
|
||||
hasNewIssueInputSinceLastRun: false,
|
||||
});
|
||||
expect(decision).toEqual({ blocked: false, noProgressStreak: 1 });
|
||||
});
|
||||
|
||||
it("does not delay recovery after a failed run", () => {
|
||||
const decision = evaluateIssueRewakeThrottle({
|
||||
now: NOW,
|
||||
recentTerminalRuns: [
|
||||
runSample({ id: "r2", status: "failed", finishedSecondsAgo: 10 }),
|
||||
runSample({ id: "r1", finishedSecondsAgo: 40 }),
|
||||
],
|
||||
runIdsWithIssueProgress: new Set(),
|
||||
hasNewIssueInputSinceLastRun: false,
|
||||
});
|
||||
expect(decision).toEqual({ blocked: false, noProgressStreak: 0 });
|
||||
});
|
||||
|
||||
it("allows when new issue input landed after the last run", () => {
|
||||
const decision = evaluateIssueRewakeThrottle({
|
||||
now: NOW,
|
||||
recentTerminalRuns: [
|
||||
runSample({ id: "r2", finishedSecondsAgo: 10 }),
|
||||
runSample({ id: "r1", finishedSecondsAgo: 40 }),
|
||||
],
|
||||
runIdsWithIssueProgress: new Set(),
|
||||
hasNewIssueInputSinceLastRun: true,
|
||||
});
|
||||
expect(decision).toEqual({ blocked: false, noProgressStreak: 0 });
|
||||
});
|
||||
});
|
||||
|
|
@ -99,6 +99,14 @@ import {
|
|||
classifyRunLiveness,
|
||||
type RunLivenessClassificationInput,
|
||||
} from "./run-liveness.js";
|
||||
import {
|
||||
ISSUE_NEW_INPUT_ACTIVITY_ACTIONS,
|
||||
ISSUE_PROGRESS_ACTIVITY_ACTIONS,
|
||||
ISSUE_REWAKE_LOOKBACK_MS,
|
||||
ISSUE_REWAKE_RUN_SAMPLE_LIMIT,
|
||||
evaluateIssueRewakeThrottle,
|
||||
isThrottleCandidateIssueRewake,
|
||||
} from "./issue-rewake-throttle.js";
|
||||
import { logActivity, publishPluginDomainEvent, type LogActivityInput } from "./activity-log.js";
|
||||
import {
|
||||
buildWorkspaceReadyComment,
|
||||
|
|
@ -15149,6 +15157,113 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
}
|
||||
}
|
||||
|
||||
// PAP-13775: no live run holds the lock, so this wake would start a
|
||||
// fresh adapter session. If this agent's recent runs on this issue
|
||||
// keep succeeding without any issue-visible progress and the wake
|
||||
// carries no new information, hold it back for an escalating cooldown
|
||||
// so external pollers/reconcilers can't storm full-price sessions.
|
||||
// Server-side recovery retries insert runs directly and never reach
|
||||
// this gate.
|
||||
if (
|
||||
isThrottleCandidateIssueRewake({
|
||||
reason,
|
||||
wakeCommentId: wakeCommentId ?? null,
|
||||
forceFreshSession: enrichedContextSnapshot.forceFreshSession === true,
|
||||
hasExplicitResume: Boolean(explicitResumeSession),
|
||||
})
|
||||
) {
|
||||
const throttleNow = new Date();
|
||||
const recentTerminalRuns = await tx
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
status: heartbeatRuns.status,
|
||||
finishedAt: heartbeatRuns.finishedAt,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.companyId, agent.companyId),
|
||||
eq(heartbeatRuns.agentId, agentId),
|
||||
sql`${heartbeatRuns.finishedAt} is not null`,
|
||||
gte(heartbeatRuns.finishedAt, new Date(throttleNow.getTime() - ISSUE_REWAKE_LOOKBACK_MS)),
|
||||
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issue.id}`,
|
||||
),
|
||||
)
|
||||
.orderBy(desc(heartbeatRuns.finishedAt))
|
||||
.limit(ISSUE_REWAKE_RUN_SAMPLE_LIMIT);
|
||||
|
||||
if (recentTerminalRuns.length > 0) {
|
||||
const sampleRunIds = recentTerminalRuns.map((sampleRun) => sampleRun.id);
|
||||
const progressRows = await tx
|
||||
.select({ runId: activityLog.runId })
|
||||
.from(activityLog)
|
||||
.where(
|
||||
and(
|
||||
eq(activityLog.companyId, agent.companyId),
|
||||
eq(activityLog.entityType, "issue"),
|
||||
eq(activityLog.entityId, issue.id),
|
||||
inArray(activityLog.runId, sampleRunIds),
|
||||
inArray(activityLog.action, ISSUE_PROGRESS_ACTIVITY_ACTIONS),
|
||||
),
|
||||
);
|
||||
const lastRunFinishedAt = recentTerminalRuns[0]?.finishedAt ?? null;
|
||||
const newInputRows = lastRunFinishedAt
|
||||
? await tx
|
||||
.select({ id: activityLog.id })
|
||||
.from(activityLog)
|
||||
.where(
|
||||
and(
|
||||
eq(activityLog.companyId, agent.companyId),
|
||||
eq(activityLog.entityType, "issue"),
|
||||
eq(activityLog.entityId, issue.id),
|
||||
gt(activityLog.createdAt, lastRunFinishedAt),
|
||||
inArray(activityLog.action, ISSUE_NEW_INPUT_ACTIVITY_ACTIONS),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
: [];
|
||||
|
||||
const throttleDecision = evaluateIssueRewakeThrottle({
|
||||
now: throttleNow,
|
||||
recentTerminalRuns,
|
||||
runIdsWithIssueProgress: new Set(
|
||||
progressRows
|
||||
.map((row) => row.runId)
|
||||
.filter((runId): runId is string => Boolean(runId)),
|
||||
),
|
||||
hasNewIssueInputSinceLastRun: newInputRows.length > 0,
|
||||
});
|
||||
|
||||
if (throttleDecision.blocked) {
|
||||
await tx.insert(agentWakeupRequests).values({
|
||||
companyId: agent.companyId,
|
||||
agentId,
|
||||
source,
|
||||
triggerDetail,
|
||||
reason: "issue_rewake_throttled",
|
||||
payload: {
|
||||
...(payload ?? {}),
|
||||
issueId,
|
||||
heartbeatSkip: {
|
||||
reason: "issue_rewake_throttled",
|
||||
requestedReason: reason,
|
||||
noProgressStreak: throttleDecision.noProgressStreak,
|
||||
cooldownMs: throttleDecision.cooldownMs,
|
||||
lastRunFinishedAt: throttleDecision.lastRunFinishedAt.toISOString(),
|
||||
nextAllowedAt: throttleDecision.nextAllowedAt.toISOString(),
|
||||
},
|
||||
},
|
||||
status: "skipped",
|
||||
requestedByActorType: opts.requestedByActorType ?? null,
|
||||
requestedByActorId: opts.requestedByActorId ?? null,
|
||||
idempotencyKey: opts.idempotencyKey ?? null,
|
||||
finishedAt: throttleNow,
|
||||
});
|
||||
return { kind: "skipped" as const };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const dailyCapBlock = await getHeartbeatDailyCapBlock(agent, policy, {}, tx);
|
||||
if (dailyCapBlock) {
|
||||
const now = new Date();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,177 @@
|
|||
/**
|
||||
* PAP-13775: throttle no-information issue re-wakes.
|
||||
*
|
||||
* After a process death (or any stall), external drivers — assignment pollers,
|
||||
* stranded-issue reconcilers, on-demand invokes — can re-wake the same agent
|
||||
* for the same issue every few seconds for as long as the issue stays
|
||||
* `in_progress`. When each of those runs ends without changing any
|
||||
* issue-visible state, every wake pays a full adapter session for zero new
|
||||
* information (the Phase 4 interruption-recovery smoke paid 25 sessions and
|
||||
* 2.4x cost for one recovery this way).
|
||||
*
|
||||
* This module decides when such a wake should be skipped: once an issue has
|
||||
* accumulated a streak of consecutive succeeded-but-no-issue-progress runs by
|
||||
* the same agent, further event-free wakes are held back for an escalating
|
||||
* cooldown anchored to the last run's finish time. Any genuinely new input —
|
||||
* a comment wake, fresh issue activity, an explicit resume, forceFreshSession,
|
||||
* or an event-carrying wake reason — bypasses the throttle entirely.
|
||||
*
|
||||
* Server-side recovery retries (process-loss retries, missing-comment
|
||||
* follow-ups) insert their runs directly and never pass through this gate, so
|
||||
* crash recovery stays immediate; only repeated no-op re-invocations slow
|
||||
* down.
|
||||
*/
|
||||
|
||||
/** Consecutive no-progress runs required before the cooldown engages. */
|
||||
export const ISSUE_REWAKE_NO_PROGRESS_THRESHOLD = 2;
|
||||
|
||||
/** Cooldown after the threshold streak; doubles per additional no-progress run. */
|
||||
export const ISSUE_REWAKE_BASE_COOLDOWN_MS = 120_000;
|
||||
|
||||
/** Upper bound for the escalating cooldown. */
|
||||
export const ISSUE_REWAKE_MAX_COOLDOWN_MS = 30 * 60_000;
|
||||
|
||||
/** Only runs newer than this feed the streak; older history is ignored. */
|
||||
export const ISSUE_REWAKE_LOOKBACK_MS = 6 * 60 * 60_000;
|
||||
|
||||
/** How many recent terminal runs to sample when computing the streak. */
|
||||
export const ISSUE_REWAKE_RUN_SAMPLE_LIMIT = 8;
|
||||
|
||||
/**
|
||||
* Wake reasons that assert issue state rather than deliver a new event.
|
||||
* These (plus reason-less on-demand invokes) are the only wakes the throttle
|
||||
* applies to; every event-shaped reason (comments, mentions, blockers
|
||||
* resolved, interactions, approvals, monitors, reviews, …) passes through.
|
||||
*/
|
||||
export const THROTTLED_ISSUE_REWAKE_REASONS: ReadonlySet<string> = new Set([
|
||||
"issue_assigned",
|
||||
"issue_continuation_needed",
|
||||
"issue_assignment_recovery",
|
||||
"issue_graph_liveness_backstop",
|
||||
]);
|
||||
|
||||
/**
|
||||
* Activity actions that count as issue-visible progress when attributed to a
|
||||
* run. Deliberately narrower than run-liveness "concrete action evidence":
|
||||
* tool calls inside the workspace do not move the issue, so they do not reset
|
||||
* the streak — a run must leave a comment, mutation, document, work product,
|
||||
* interaction, or scheduled continuation behind.
|
||||
*/
|
||||
export const ISSUE_PROGRESS_ACTIVITY_ACTIONS: string[] = [
|
||||
"issue.updated",
|
||||
"issue.comment_added",
|
||||
"issue.created",
|
||||
"issue.child_created",
|
||||
"issue.assigned",
|
||||
"issue.released",
|
||||
"issue.blockers_updated",
|
||||
"issue.document_upserted",
|
||||
"issue.document_updated",
|
||||
"issue.document_deleted",
|
||||
"issue.document_restored",
|
||||
"issue.document_annotation_comment_added",
|
||||
"issue.document_annotation_thread_created",
|
||||
"issue.document_annotation_thread_resolved",
|
||||
"issue.work_product_created",
|
||||
"issue.work_product_updated",
|
||||
"issue.work_product_deleted",
|
||||
"issue.attachment_added",
|
||||
"issue.attachment_removed",
|
||||
"issue.thread_interaction_created",
|
||||
"issue.monitor_scheduled",
|
||||
"issue.approval_linked",
|
||||
];
|
||||
|
||||
/**
|
||||
* Activity on the issue that counts as new external input since the last run
|
||||
* finished — anything a waiting agent should be woken for, including board
|
||||
* responses to interactions.
|
||||
*/
|
||||
export const ISSUE_NEW_INPUT_ACTIVITY_ACTIONS: string[] = [
|
||||
...ISSUE_PROGRESS_ACTIVITY_ACTIONS,
|
||||
"issue.thread_interaction_accepted",
|
||||
"issue.thread_interaction_answered",
|
||||
"issue.thread_interaction_item_verdicts_submitted",
|
||||
"issue.blockers_resolved_wake_emitted",
|
||||
];
|
||||
|
||||
export interface IssueRewakeCandidateInput {
|
||||
reason: string | null;
|
||||
wakeCommentId: string | null;
|
||||
forceFreshSession: boolean;
|
||||
hasExplicitResume: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a wake is even a candidate for throttling. Wakes that carry new
|
||||
* information or an explicit operator escalation always pass.
|
||||
*/
|
||||
export function isThrottleCandidateIssueRewake(input: IssueRewakeCandidateInput): boolean {
|
||||
if (input.forceFreshSession) return false;
|
||||
if (input.wakeCommentId) return false;
|
||||
if (input.hasExplicitResume) return false;
|
||||
if (input.reason === null) return true;
|
||||
return THROTTLED_ISSUE_REWAKE_REASONS.has(input.reason);
|
||||
}
|
||||
|
||||
export interface RecentIssueRunSample {
|
||||
id: string;
|
||||
status: string;
|
||||
finishedAt: Date | null;
|
||||
}
|
||||
|
||||
export interface IssueRewakeThrottleInput {
|
||||
now: Date;
|
||||
/** Terminal runs for the same (agent, issue), newest finish first. */
|
||||
recentTerminalRuns: RecentIssueRunSample[];
|
||||
/** Runs among the sample that produced issue-visible progress. */
|
||||
runIdsWithIssueProgress: ReadonlySet<string>;
|
||||
/** New issue input landed after the newest run finished. */
|
||||
hasNewIssueInputSinceLastRun: boolean;
|
||||
}
|
||||
|
||||
export type IssueRewakeThrottleDecision =
|
||||
| { blocked: false; noProgressStreak: number }
|
||||
| {
|
||||
blocked: true;
|
||||
noProgressStreak: number;
|
||||
cooldownMs: number;
|
||||
lastRunFinishedAt: Date;
|
||||
nextAllowedAt: Date;
|
||||
};
|
||||
|
||||
export function computeIssueRewakeCooldownMs(noProgressStreak: number): number {
|
||||
const doublings = Math.max(0, noProgressStreak - ISSUE_REWAKE_NO_PROGRESS_THRESHOLD);
|
||||
// Guard the exponent so an absurd streak can't overflow into Infinity.
|
||||
const factor = 2 ** Math.min(doublings, 16);
|
||||
return Math.min(ISSUE_REWAKE_BASE_COOLDOWN_MS * factor, ISSUE_REWAKE_MAX_COOLDOWN_MS);
|
||||
}
|
||||
|
||||
export function evaluateIssueRewakeThrottle(input: IssueRewakeThrottleInput): IssueRewakeThrottleDecision {
|
||||
const runs = input.recentTerminalRuns;
|
||||
if (runs.length === 0) return { blocked: false, noProgressStreak: 0 };
|
||||
if (input.hasNewIssueInputSinceLastRun) return { blocked: false, noProgressStreak: 0 };
|
||||
|
||||
let noProgressStreak = 0;
|
||||
for (const run of runs) {
|
||||
// A failed/cancelled/interrupted run breaks the streak: its follow-up is
|
||||
// recovery, not a redundant re-poll, and must not be delayed.
|
||||
if (run.status !== "succeeded" || !run.finishedAt) break;
|
||||
if (input.runIdsWithIssueProgress.has(run.id)) break;
|
||||
noProgressStreak += 1;
|
||||
}
|
||||
|
||||
if (noProgressStreak < ISSUE_REWAKE_NO_PROGRESS_THRESHOLD) {
|
||||
return { blocked: false, noProgressStreak };
|
||||
}
|
||||
|
||||
const lastRunFinishedAt = runs[0]?.finishedAt;
|
||||
if (!lastRunFinishedAt) return { blocked: false, noProgressStreak };
|
||||
|
||||
const cooldownMs = computeIssueRewakeCooldownMs(noProgressStreak);
|
||||
const nextAllowedAt = new Date(lastRunFinishedAt.getTime() + cooldownMs);
|
||||
if (input.now.getTime() < nextAllowedAt.getTime()) {
|
||||
return { blocked: true, noProgressStreak, cooldownMs, lastRunFinishedAt, nextAllowedAt };
|
||||
}
|
||||
return { blocked: false, noProgressStreak };
|
||||
}
|
||||
Loading…
Reference in New Issue