diff --git a/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts b/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts new file mode 100644 index 0000000000..637512f399 --- /dev/null +++ b/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts @@ -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("../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; + let heartbeat!: ReturnType; + let tempDb: Awaited> | 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 | null)?.heartbeatSkip as + | Record + | undefined; + expect(heartbeatSkip?.noProgressStreak).toBe(2); + expect(typeof heartbeatSkip?.nextAllowedAt).toBe("string"); + + const runCount = await db + .select({ count: sql`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"); + }); +}); diff --git a/server/src/__tests__/issue-rewake-throttle.test.ts b/server/src/__tests__/issue-rewake-throttle.test.ts new file mode 100644 index 0000000000..4c711431ba --- /dev/null +++ b/server/src/__tests__/issue-rewake-throttle.test.ts @@ -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 }); + }); +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 2a7095194f..2c792bef19 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -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(); diff --git a/server/src/services/issue-rewake-throttle.ts b/server/src/services/issue-rewake-throttle.ts new file mode 100644 index 0000000000..cd9022f263 --- /dev/null +++ b/server/src/services/issue-rewake-throttle.ts @@ -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 = 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; + /** 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 }; +}