Q68: sessions are bounded in time, tool calls per turn and goal audits
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
c38b424f4b
commit
4325dfa874
6 files changed
+116
-20
No files matched your search
@@ -6,7 +6,7 @@ import type { AgentToolContext, AgentToolDeps } from "./agent-tools";
|
||||
import type { JsonRecord } from "./provider-http";
|
||||
import type { ProviderService } from "./provider-service";
|
||||
import type { AgentTurnMessage, ToolCall, ToolDefinition } from "./provider-tool-turn";
|
||||
import { throwIfRunStopped } from "./run-control";
|
||||
import { MAX_TOOL_CALLS_PER_TURN, throwIfRunStopped, untilStopped } from "./run-control";
|
||||
|
||||
const AUDIT_MAX_ITERATIONS = 6;
|
||||
const STATUS_TASK_LINES = 8;
|
||||
@@ -134,19 +134,24 @@ async function auditGoalCompletion(
|
||||
|
||||
for (let iteration = 1; iteration <= AUDIT_MAX_ITERATIONS; iteration += 1) {
|
||||
fitContextBudget(instructions, messages, auditTools, args.inputBudgetTokens, iteration);
|
||||
const turn = await args.providers.runToolTurn({
|
||||
throwIfRunStopped();
|
||||
const turn = await untilStopped(args.providers.runToolTurn({
|
||||
orgId: args.ctx.orgId,
|
||||
providerId: args.providerId,
|
||||
workspaceId: args.workspaceId,
|
||||
instructions,
|
||||
messages,
|
||||
tools: auditTools,
|
||||
});
|
||||
}));
|
||||
if (turn.toolCalls.length === 0) return parseVerdict(turn.text);
|
||||
messages.push({ role: "assistant", text: turn.text, toolCalls: turn.toolCalls });
|
||||
const results: Array<{ toolCallId: string; toolName: string; output: string }> = [];
|
||||
for (const toolCall of turn.toolCalls) {
|
||||
results.push({ toolCallId: toolCall.id, toolName: toolCall.name, output: await executeWorkspaceTool(auditCtx, toolCall) });
|
||||
for (const [index, toolCall] of turn.toolCalls.entries()) {
|
||||
throwIfRunStopped();
|
||||
const output = index < MAX_TOOL_CALLS_PER_TURN
|
||||
? await untilStopped(executeWorkspaceTool(auditCtx, toolCall))
|
||||
: `ERROR: not run: this turn asked for ${turn.toolCalls.length} tool calls; the limit is ${MAX_TOOL_CALLS_PER_TURN}.`;
|
||||
results.push({ toolCallId: toolCall.id, toolName: toolCall.name, output });
|
||||
}
|
||||
messages.push({ role: "tool", results });
|
||||
}
|
||||
|
||||
@@ -8,9 +8,11 @@ import type { AgentTurnMessage, ToolCall, ToolDefinition } from "./provider-tool
|
||||
import { isComposioTool, executeComposioTool } from "./composio/composio-tool-bridge";
|
||||
import { capToolOutput, compactToolHistory } from "./run-context-compaction";
|
||||
import { fitContextBudget } from "./context-budget";
|
||||
import { throwIfRunStopped, untilStopped } from "./run-control";
|
||||
import { MAX_TOOL_CALLS_PER_TURN, throwIfRunStopped, untilStopped } from "./run-control";
|
||||
|
||||
const MAX_ITERATIONS = 12;
|
||||
/** Each goal audit runs its own model loop, so a session gets only a couple. */
|
||||
const MAX_GOAL_AUDITS_PER_SESSION = 2;
|
||||
const STEP_LIMIT_TEXT = "I used this session's steps and will continue in the next one.";
|
||||
|
||||
export interface AgentRunLoopInput {
|
||||
@@ -52,18 +54,14 @@ export function runTools(externalTools: ToolDefinition[]): ToolDefinition[] {
|
||||
return [...AGENT_TOOLS, ...WORKSPACE_TOOLS, ...WEB_TOOLS, ...externalTools];
|
||||
}
|
||||
|
||||
/**
|
||||
* Drives a real tool-use loop against the live provider: prompt the model with
|
||||
* the agent tools, execute every tool call it makes, feed results back, and
|
||||
* repeat until the model returns a turn with no tool calls (it is done) or the
|
||||
* iteration cap is hit (which throws rather than silently truncating).
|
||||
*/
|
||||
/** Runs the model with the agent tools until it stops calling them; step, time, tool-call and audit limits end it as a checkpoint or refuse the excess. */
|
||||
export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopInput): Promise<AgentRunLoopResult> {
|
||||
const allTools = runTools(input.externalTools ?? []);
|
||||
const messages: AgentTurnMessage[] = [{ role: "user", text: input.initialUserTurn }];
|
||||
let toolCallCount = 0;
|
||||
let lastRequestId = "";
|
||||
let lastText = "";
|
||||
let goalAudits = 0;
|
||||
|
||||
for (let iteration = 1; iteration <= MAX_ITERATIONS; iteration += 1) {
|
||||
const compaction = compactToolHistory(messages);
|
||||
@@ -106,8 +104,17 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn
|
||||
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) {
|
||||
for (const [index, call] of turn.toolCalls.entries()) {
|
||||
throwIfRunStopped();
|
||||
if (index >= MAX_TOOL_CALLS_PER_TURN) {
|
||||
results.push({ toolCallId: call.id, toolName: call.name, output: `ERROR: not run: this turn asked for ${turn.toolCalls.length} tool calls; the limit is ${MAX_TOOL_CALLS_PER_TURN}. Repeat the ones you still need.` });
|
||||
continue;
|
||||
}
|
||||
if (call.name === "complete_goal" && goalAudits >= MAX_GOAL_AUDITS_PER_SESSION) {
|
||||
results.push({ toolCallId: call.id, toolName: call.name, output: `ERROR: this session already ran ${MAX_GOAL_AUDITS_PER_SESSION} goal audits. Close the gaps they named, then try again next session.` });
|
||||
continue;
|
||||
}
|
||||
if (call.name === "complete_goal") goalAudits += 1;
|
||||
toolCallCount += 1;
|
||||
let output: string;
|
||||
if (isComposioTool(call.name)) {
|
||||
|
||||
@@ -2,20 +2,30 @@ import { AsyncLocalStorage } from "node:async_hooks";
|
||||
|
||||
/** The error message a run ends with when the operator stops it. */
|
||||
export const RUN_STOPPED_MESSAGE = "Stopped by the operator.";
|
||||
/** The reason a session hits its wall-clock budget; it ends as a checkpoint, not a failure. */
|
||||
export const SESSION_TIME_LIMIT_MESSAGE = "The session reached its time limit.";
|
||||
export const SESSION_TIME_LIMIT_MS = 30 * 60_000;
|
||||
/** Tool calls one model turn may run, in a session or a nested audit. */
|
||||
export const MAX_TOOL_CALLS_PER_TURN = 16;
|
||||
|
||||
const runScope = new AsyncLocalStorage<{ runId: string; signal: AbortSignal }>();
|
||||
const controllers = new Map<string, AbortController>();
|
||||
const settling = new Map<string, Promise<unknown>>();
|
||||
|
||||
/** Runs a session under a stop signal that every provider request and tool inside it observes. */
|
||||
export async function withRunControl<T>(runId: string, operation: () => Promise<T>): Promise<T> {
|
||||
/** Runs a session under a stop signal that every provider request and tool inside it observes; the signal also fires at the session's deadline. */
|
||||
export async function withRunControl<T>(runId: string, operation: () => Promise<T>, timeLimitMs: number = SESSION_TIME_LIMIT_MS): Promise<T> {
|
||||
const controller = new AbortController();
|
||||
controllers.set(runId, controller);
|
||||
const deadline = setTimeout(() => {
|
||||
console.info(`[run-control] run ${runId} reached its ${timeLimitMs / 60_000}-minute limit`);
|
||||
controller.abort(new Error(SESSION_TIME_LIMIT_MESSAGE));
|
||||
}, timeLimitMs);
|
||||
const running = runScope.run({ runId, signal: controller.signal }, operation);
|
||||
settling.set(runId, running.catch(() => undefined));
|
||||
try {
|
||||
return await running;
|
||||
} finally {
|
||||
clearTimeout(deadline);
|
||||
controllers.delete(runId);
|
||||
settling.delete(runId);
|
||||
}
|
||||
|
||||
@@ -13,7 +13,7 @@ import { runAgentLoop, runTools } from "./agent-run-loop";
|
||||
import type { RunModel } from "./skill-runtime";
|
||||
import { combineMidRunPolls, createGoalChangePoller } from "./goal-change-poller";
|
||||
import { recordRunAction, runJournalLines } from "./run-journal";
|
||||
import { runStopReason, withRunControl } from "./run-control";
|
||||
import { runStopReason, SESSION_TIME_LIMIT_MESSAGE, SESSION_TIME_LIMIT_MS, withRunControl } from "./run-control";
|
||||
import type { AgentToolContext } from "./agent-tools";
|
||||
import {
|
||||
buildAgentInstructions,
|
||||
@@ -77,6 +77,8 @@ export class TaskExecutionService {
|
||||
private readonly notifyDataChanged: () => void = () => undefined,
|
||||
/** Injectable clock so tests can walk providers out of cooldown. */
|
||||
now: () => number = () => Date.now(),
|
||||
/** Wall-clock budget of one session. */
|
||||
private readonly sessionTimeLimitMs: number = SESSION_TIME_LIMIT_MS,
|
||||
) {
|
||||
this.audit = new AuditService(db);
|
||||
this.organization = new OrganizationService(db);
|
||||
@@ -99,7 +101,7 @@ export class TaskExecutionService {
|
||||
const workspace = loadWorkspace(this.db, task.workspace_id);
|
||||
const isolation = createIsolationContext(this.db, input.orgId, task, agent, provider, workspace);
|
||||
const runId = this.createRun(input.orgId, task.id, agent.id, provider.id, isolation);
|
||||
return withRunControl(runId, () => this.executeAgentRun({ orgId: input.orgId, runId, task, agent, provider, workspace, isolation }));
|
||||
return withRunControl(runId, () => this.executeAgentRun({ orgId: input.orgId, runId, task, agent, provider, workspace, isolation }), this.sessionTimeLimitMs);
|
||||
}
|
||||
|
||||
/** Executes a claimed run after re-checking that the task still matches it; a mismatch supersedes the run. */
|
||||
@@ -157,7 +159,7 @@ export class TaskExecutionService {
|
||||
if (task && task.origin === null) recordRunMemory(this.db, { orgId, runId, projectId: task.project_id, taskTitle: task.title, outcome: "failed", text: message });
|
||||
throw error;
|
||||
}
|
||||
return withRunControl(runId, () => this.executeAgentRun(prepared));
|
||||
return withRunControl(runId, () => this.executeAgentRun(prepared), this.sessionTimeLimitMs);
|
||||
}
|
||||
|
||||
private async executeAgentRun(args: {
|
||||
@@ -286,6 +288,12 @@ export class TaskExecutionService {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
// A stop wins over whatever error the interrupted step raised.
|
||||
const stopReason = runStopReason();
|
||||
// Out of time is not a failure: the session ends as a checkpoint and the next one continues from the journal.
|
||||
if (stopReason === SESSION_TIME_LIMIT_MESSAGE) {
|
||||
this.completeRun(orgId, runId, providerRequestId, true);
|
||||
this.audit.record({ orgId, entityType: "task_run", entityId: runId, actorType: "system", actorId: null, action: "completed", payload: { taskId: task.id, agentId: agent.id, timeLimit: true } });
|
||||
return runId;
|
||||
}
|
||||
if (stopReason) {
|
||||
this.markStopped(orgId, runId, providerRequestId, stopReason, { taskId: task.id, agentId: agent.id, providerId: provider.id });
|
||||
throw new Error(stopReason);
|
||||
@@ -400,10 +408,10 @@ export class TaskExecutionService {
|
||||
return row?.language === "zh" ? "zh" : "en";
|
||||
}
|
||||
|
||||
private completeRun(orgId: string, runId: string, providerRequestId: string, checkpoint: boolean): void {
|
||||
private completeRun(orgId: string, runId: string, providerRequestId: string | null, checkpoint: boolean): void {
|
||||
this.db.prepare(`
|
||||
update task_runs
|
||||
set state = ?, provider_request_id = ?, completed_at = ?, error_message = null
|
||||
set state = ?, provider_request_id = coalesce(?, provider_request_id), completed_at = ?, error_message = null
|
||||
where id = ? and org_id = ?
|
||||
`).run(TASK_RUN_STATE.COMPLETED, providerRequestId, new Date().toISOString(), runId, orgId);
|
||||
if (!checkpoint) this.closeChatWakeTask(orgId, runId);
|
||||
|
||||
Reference in new issue
Block a user