Q60: step limits end sessions as checkpoints; only sessions without progress spend the budget
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
8775c34650
commit
88463e4396
6 files changed
+166
-35
No files matched your search
@@ -11,6 +11,7 @@ import { fitContextBudget } from "./context-budget";
|
||||
import { throwIfRunStopped, untilStopped } from "./run-control";
|
||||
|
||||
const MAX_ITERATIONS = 12;
|
||||
const STEP_LIMIT_TEXT = "I used this session's steps and will continue in the next one.";
|
||||
|
||||
export interface AgentRunLoopInput {
|
||||
orgId: string;
|
||||
@@ -35,6 +36,8 @@ export interface AgentRunLoopResult {
|
||||
iterations: number;
|
||||
toolCallCount: number;
|
||||
lastRequestId: string;
|
||||
/** The session was still working when it reached the step limit; the task continues next session. */
|
||||
hitStepLimit: boolean;
|
||||
}
|
||||
|
||||
export interface AgentRunLoopDeps {
|
||||
@@ -60,6 +63,7 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn
|
||||
const messages: AgentTurnMessage[] = [{ role: "user", text: input.initialUserTurn }];
|
||||
let toolCallCount = 0;
|
||||
let lastRequestId = "";
|
||||
let lastText = "";
|
||||
|
||||
for (let iteration = 1; iteration <= MAX_ITERATIONS; iteration += 1) {
|
||||
const compaction = compactToolHistory(messages);
|
||||
@@ -96,9 +100,10 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn
|
||||
messages.push({ role: "assistant", text: turn.text, toolCalls: [] }, { role: "user", text: lateMessages });
|
||||
continue;
|
||||
}
|
||||
return { finalText: turn.text, iterations: iteration, toolCallCount, lastRequestId };
|
||||
return { finalText: turn.text, iterations: iteration, toolCallCount, lastRequestId, hitStepLimit: false };
|
||||
}
|
||||
|
||||
lastText = turn.text.trim() || lastText;
|
||||
messages.push({ role: "assistant", text: turn.text, toolCalls: turn.toolCalls });
|
||||
const results: Array<{ toolCallId: string; toolName: string; output: string }> = [];
|
||||
for (const call of turn.toolCalls) {
|
||||
@@ -136,7 +141,9 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn
|
||||
deps.onToolResults?.();
|
||||
}
|
||||
|
||||
throw new Error(`Agent run did not converge within ${MAX_ITERATIONS} iterations.`);
|
||||
// Still working at the limit: end as a checkpoint; the next session picks up from the run journal.
|
||||
console.info(`[run-loop] reached the ${MAX_ITERATIONS}-step limit with work in progress; ending the session as a checkpoint`);
|
||||
return { finalText: lastText || STEP_LIMIT_TEXT, iterations: MAX_ITERATIONS, toolCallCount, lastRequestId, hitStepLimit: true };
|
||||
}
|
||||
|
||||
function shouldAnnounceTaskAssignment(call: { name: string; arguments: Record<string, unknown> }, toolCalls: Array<{ name: string; arguments: Record<string, unknown> }>): boolean {
|
||||
|
||||
@@ -5,7 +5,10 @@ import type { ProviderCooldowns } from "./provider-cooldowns";
|
||||
import { assertExecutionApproval, createIsolationContext, loadWorkspace, requireAgentProvider, requireAssignedAgent, requireTask } from "./task-run-preparation";
|
||||
import { AuditService } from "./audit-service";
|
||||
|
||||
/** Sessions that made no progress before a task stalls. */
|
||||
const TASK_RETRY_BUDGET = 3;
|
||||
/** Sessions of any kind before a task stalls, so endless busywork still stops. */
|
||||
const TASK_SESSION_CEILING = 12;
|
||||
|
||||
/** Enqueues follow-up runs for open tasks each worker tick. */
|
||||
export class HeartbeatScheduler {
|
||||
@@ -19,22 +22,7 @@ export class HeartbeatScheduler {
|
||||
private readonly notifyDataChanged: () => void,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* Heartbeat: keeps every open assigned task of a heartbeat-enabled agent
|
||||
* moving. A task is enqueued whenever it has no run in flight and its
|
||||
* status is still open — a completed run that left the task open gets a
|
||||
* follow-up session (the prompt carries the prior-run count), because a
|
||||
* task is finished when it reaches done, not when a run happens to end.
|
||||
* The retry budget bounds this: at most TASK_RETRY_BUDGET non-transient
|
||||
* runs per task, then it parks for a human or manager to intervene.
|
||||
*
|
||||
* Three rules keep the loop self-healing when a provider stops serving:
|
||||
* tasks on a provider over its monthly window are parked (not enqueued into
|
||||
* certain failure), tasks on a provider inside a transient-failure cooldown
|
||||
* wait it out, and transient failures (usage/rate limits, overload) never
|
||||
* count toward the retry budget. Raising a limit — or the provider's own
|
||||
* window resetting — therefore resumes work without a restart.
|
||||
*/
|
||||
/** Queues the next session of open work; stalls a task after sessions without progress or too many in all; transient provider failures never count. */
|
||||
enqueueHeartbeatRuns(): string[] {
|
||||
const window = currentProviderUsageWindow();
|
||||
const nowIso = new Date(this.now()).toISOString();
|
||||
@@ -62,19 +50,26 @@ export class HeartbeatScheduler {
|
||||
`).all(window.start, nowIso) as Array<{ task_id: string; org_id: string; provider_id: string; agent_id: string; interval_sec: number }>;
|
||||
this.observeQuotaBlocks(window.start);
|
||||
const cooling = new Set(this.cooldowns.cooledProviderIds());
|
||||
const attemptMessages = this.db.prepare(`
|
||||
select r.error_message from task_runs r join tasks t on t.id = r.task_id
|
||||
where r.task_id = ? and r.state not in ('superseded', 'stopped') and r.rowid > t.run_budget_floor
|
||||
const sessionsOf = this.db.prepare(`
|
||||
select r.error_message,
|
||||
exists (select 1 from run_actions a where a.run_id = r.id and a.result_excerpt not like 'ERROR%') as progressed
|
||||
from task_runs r join tasks t on t.id = r.task_id
|
||||
where r.task_id = ? and r.state not in ('superseded', 'stopped', 'queued', 'running') and r.rowid > t.run_budget_floor
|
||||
`);
|
||||
const created: string[] = [];
|
||||
const pacedAgents = new Set<string>();
|
||||
for (const row of rows) {
|
||||
if (cooling.has(row.provider_id)) continue;
|
||||
if (row.interval_sec > 0 && pacedAgents.has(row.agent_id)) continue;
|
||||
const attempts = (attemptMessages.all(row.task_id) as Array<{ error_message: string | null }>)
|
||||
.filter((run) => !isTransientProviderFailure(run.error_message)).length;
|
||||
if (attempts >= TASK_RETRY_BUDGET) {
|
||||
this.markStalled(row.org_id, row.task_id);
|
||||
const sessions = (sessionsOf.all(row.task_id) as Array<{ error_message: string | null; progressed: number }>)
|
||||
.filter((run) => !isTransientProviderFailure(run.error_message));
|
||||
const stuck = sessions.filter((run) => run.progressed === 0).length;
|
||||
if (stuck >= TASK_RETRY_BUDGET) {
|
||||
this.markStalled(row.org_id, row.task_id, `${TASK_RETRY_BUDGET} sessions made no progress`);
|
||||
continue;
|
||||
}
|
||||
if (sessions.length >= TASK_SESSION_CEILING) {
|
||||
this.markStalled(row.org_id, row.task_id, `${TASK_SESSION_CEILING} sessions ran without reaching done`);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
@@ -88,8 +83,8 @@ export class HeartbeatScheduler {
|
||||
}
|
||||
|
||||
/** Records once, on the task itself, that it spent its run budget without reaching done. */
|
||||
private markStalled(orgId: string, taskId: string): void {
|
||||
const reason = `Stalled: ${TASK_RETRY_BUDGET} sessions ran without reaching done. Edit, reassign or reopen the task to give it a fresh budget.`;
|
||||
private markStalled(orgId: string, taskId: string, why: string): void {
|
||||
const reason = `Stalled: ${why}. Edit, reassign or reopen the task to give it a fresh budget.`;
|
||||
const marked = this.db.prepare("update tasks set stalled_reason = ? where id = ? and stalled_reason is null").run(reason, taskId);
|
||||
if (marked.changes === 0) return;
|
||||
const task = this.db.prepare("select title from tasks where id = ?").get(taskId) as { title: string };
|
||||
|
||||
@@ -246,7 +246,8 @@ export class TaskExecutionService {
|
||||
// in plain text instead of calling send_message, deliver that text into
|
||||
// the source thread BEFORE the run may complete — a delivery failure
|
||||
// fails the run rather than closing it on silence.
|
||||
if (chatContext) {
|
||||
// A checkpoint is unfinished work: no stand-in reply, and a chat wake stays open for its next session.
|
||||
if (chatContext && !result.hitStepLimit) {
|
||||
const delivered = deliverChatReplyIfMissing(this.db, this.chat, {
|
||||
orgId,
|
||||
agentId: agent.id,
|
||||
@@ -257,7 +258,7 @@ export class TaskExecutionService {
|
||||
});
|
||||
if (delivered) this.notifyDataChanged();
|
||||
}
|
||||
this.completeRun(orgId, runId, result.lastRequestId);
|
||||
this.completeRun(orgId, runId, result.lastRequestId, result.hitStepLimit);
|
||||
this.cooldowns.clear(provider);
|
||||
if (task.origin === null) recordRunMemory(this.db, { orgId, runId, projectId: task.project_id, taskTitle: task.title, outcome: "completed", text: result.finalText });
|
||||
this.audit.record({
|
||||
@@ -274,6 +275,7 @@ export class TaskExecutionService {
|
||||
providerRequestId: result.lastRequestId,
|
||||
iterations: result.iterations,
|
||||
toolCalls: result.toolCallCount,
|
||||
hitStepLimit: result.hitStepLimit,
|
||||
executionContextId: isolation.executionContextId,
|
||||
serviceIdentityId: isolation.serviceIdentityId,
|
||||
skillRuntimeSnapshotId: skillRuntimeSnapshot?.id ?? null,
|
||||
@@ -398,13 +400,13 @@ export class TaskExecutionService {
|
||||
return row?.language === "zh" ? "zh" : "en";
|
||||
}
|
||||
|
||||
private completeRun(orgId: string, runId: string, providerRequestId: string): void {
|
||||
private completeRun(orgId: string, runId: string, providerRequestId: string, checkpoint: boolean): void {
|
||||
this.db.prepare(`
|
||||
update task_runs
|
||||
set state = ?, provider_request_id = ?, completed_at = ?, error_message = null
|
||||
where id = ? and org_id = ?
|
||||
`).run(TASK_RUN_STATE.COMPLETED, providerRequestId, new Date().toISOString(), runId, orgId);
|
||||
this.closeChatWakeTask(orgId, runId);
|
||||
if (!checkpoint) this.closeChatWakeTask(orgId, runId);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -414,9 +416,9 @@ export class TaskExecutionService {
|
||||
* enqueued again) and pile zombie "Reply to ..." rows onto the board.
|
||||
*/
|
||||
private closeChatWakeTask(orgId: string, runId: string): void {
|
||||
const run = this.db.prepare("select task_id, trigger from task_runs where id = ? and org_id = ?")
|
||||
.get(runId, orgId) as { task_id: string; trigger: string } | undefined;
|
||||
if (!run || run.trigger !== "chat_mention") return;
|
||||
const run = this.db.prepare("select r.task_id, t.origin from task_runs r join tasks t on t.id = r.task_id where r.id = ? and r.org_id = ?")
|
||||
.get(runId, orgId) as { task_id: string; origin: string | null } | undefined;
|
||||
if (!run || run.origin !== "chat_wake") return;
|
||||
const task = this.db.prepare("select status from tasks where id = ? and org_id = ?")
|
||||
.get(run.task_id, orgId) as { status: string } | undefined;
|
||||
if (!task || (task.status !== TASK_STATUS.BACKLOG && task.status !== TASK_STATUS.IN_PROGRESS)) return;
|
||||
|
||||
Reference in new issue
Block a user