Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/atomic-file-writes.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@pymodel/pythinker-code": patch
---

Write files atomically so an interrupted write leaves the previous content intact.
5 changes: 5 additions & 0 deletions .changeset/cancel-prompt-while-starting.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@pymodel/pythinker-code": patch
---

Allow cancelling a prompt while it is still starting.
5 changes: 5 additions & 0 deletions .changeset/dsml-parser-chunk-invariance.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@pymodel/pythinker-code": patch
---

Fix DSML and Hermes tool calls in streamed responses being dropped, split, or mistaken for quoted documentation depending on how the response was chunked.
5 changes: 5 additions & 0 deletions .changeset/sensitive-file-symlink-alias.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@pymodel/pythinker-code": patch
---

Block file reads and writes that reach a sensitive file through a symlink alias.
5 changes: 5 additions & 0 deletions .changeset/subagent-usage-per-run.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@pymodel/pythinker-code": patch
---

Report each subagent run's own token usage instead of the agent's lifetime total.
5 changes: 5 additions & 0 deletions .changeset/tool-cancel-holds-file-lease.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@pymodel/pythinker-code": patch
---

Keep a cancelled tool that ignores the stop signal from overlapping with the next tool on the same file.
5 changes: 5 additions & 0 deletions .changeset/web-fetch-streaming-limit.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@pymodel/pythinker-code": patch
---

Stop downloading a web page as soon as it exceeds the size limit instead of buffering it first.
29 changes: 26 additions & 3 deletions packages/agent-core-v2/src/agent/prompt/promptService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,7 @@ interface Record extends PromptSnapshot {
readonly launchedDeferred: Deferred<Turn | undefined>;
readonly completionDeferred: Deferred<PromptCompletion>;
handle: PromptHandle;
cancelReason?: Error;
}

function bundledSkillBlockCount(message: ContextMessage): number {
Expand Down Expand Up @@ -227,6 +228,7 @@ export const promptLaunchingKey = defineState<boolean>('prompt.launching', () =>
export class AgentPromptService implements IAgentPromptService {
declare readonly _serviceBrand: undefined;
private active: (Record & { turn: Turn }) | undefined;
private launchingRecord: Record | undefined;
private readonly pending: Record[] = [];
private readonly steered = new Map<string, Record[]>();
private readonly reservedPromptIds = new Set<string>();
Expand Down Expand Up @@ -455,6 +457,7 @@ export class AgentPromptService implements IAgentPromptService {

abort(promptId: string, reason: Error = userCancellationReason()): boolean {
if (this.active?.id === promptId) { this.loop.cancel(this.active.turn.id, reason); return true; }
if (this.launchingRecord?.id === promptId) { this.launchingRecord.cancelReason ??= reason; return true; }
const index = this.pending.findIndex((item) => item.id === promptId);
if (index < 0) throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, `prompt ${promptId} not found`);
const [item] = this.pending.splice(index, 1) as [Record];
Expand All @@ -466,6 +469,11 @@ export class AgentPromptService implements IAgentPromptService {

async drain(reason: Error = userCancellationReason()): Promise<void> {
for (const item of this.pending.slice()) this.abort(item.id, reason);
const launching = this.launchingRecord;
if (launching !== undefined) {
this.abort(launching.id, reason);
await launching.launchedDeferred.promise;
}
if (this.active !== undefined) this.abort(this.active.id, reason);
}

Expand All @@ -488,47 +496,62 @@ export class AgentPromptService implements IAgentPromptService {

clear(): void {
for (const item of this.pending.slice()) this.abort(item.id);
if (this.launchingRecord !== undefined) this.abort(this.launchingRecord.id);
if (this.active !== undefined) this.abort(this.active.id);
this.context.clear();
}

private async startNext(): Promise<void> {
if (this.active !== undefined || this.launching || this.steering > 0) return;
const item = this.pending.shift(); if (item === undefined) return;
if (this.fullCompaction.compacting !== null && this.loop.status().state !== 'running') { this.pending.unshift(item); return; }
this.launching = true;
this.launchingRecord = item;
try {
if (this.fullCompaction.compacting !== null && this.loop.status().state !== 'running') { this.pending.unshift(item); return; }
const { message, captions } = this.extractCompressionCaptions(item.message);
await this.materializeDaemonRefs(message);
if (this.settleCancelledLaunch(item)) return;
if (await this.blockedByHook(message, false)) {
this.appendPrompt(message, captions); item.state = 'blocked'; item.launchedDeferred.resolve(undefined);
item.completionDeferred.resolve({ promptId: item.id, result: undefined, state: 'blocked' });
this.publishCompleted(item.id, 'blocked'); return;
}
const turn = (await this.loop.enqueue(
if (this.settleCancelledLaunch(item)) return;
const receipt = this.loop.enqueue(
new PromptStepRequest(
message,
captions,
this.reminder(),
item.maxOutputSize,
item.infiniteRetry,
),
).assigned).turn;
);
const turn = (await receipt.assigned).turn;
if (turn === undefined) { this.pending.unshift(item); return; }
item.state = 'running'; item.launchedDeferred.resolve(turn); this.active = Object.assign(item, { turn });
this.publishStarted(item);
if (item.cancelReason !== undefined) this.loop.cancel(turn.id, item.cancelReason);
void turn.result.then((result) => this.settle(item, result));
} catch {
item.state = 'failed';
item.launchedDeferred.resolve(undefined);
item.completionDeferred.resolve({ promptId: item.id, result: undefined, state: 'failed' });
this.publishCompleted(item.id, 'failed');
} finally {
this.launchingRecord = undefined;
this.launching = false;
if (this.active === undefined) void this.startNext();
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
}

private settleCancelledLaunch(item: Record): boolean {
if (item.cancelReason === undefined) return false;
item.state = 'cancelled'; item.launchedDeferred.resolve(undefined);
item.completionDeferred.resolve({ promptId: item.id, result: undefined, state: 'cancelled' });
this.publishAborted(item.id);
return true;
}

private settle(item: Record, result: TurnResult): void {
if (this.active?.id !== item.id) return;
this.active = undefined;
Expand Down
114 changes: 79 additions & 35 deletions packages/agent-core-v2/src/agent/toolExecutor/toolExecutorService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ import {
type UnavailableToolDescriber,
} from './toolExecutor';
import { ToolCallStarted, ToolProgress, ToolResultEvent } from './toolExecutorEvents';
import { ToolScheduler } from './toolScheduler';
import { ToolScheduler, type OutstandingEffect } from './toolScheduler';

const ABORT_GRACE_MS = 2_000;
const TOOL_OUTPUT_EMPTY = 'Tool output is empty.';
Expand All @@ -71,12 +71,15 @@ export interface ToolExecutionTask {
export interface ToolExecutionRunResult {
readonly result: ToolResult;
readonly outcome: ToolExecutionOutcome;
readonly cancelled?: boolean;
readonly effectsSettled?: Promise<void>;
}

interface TimedToolResult {
readonly index: number;
readonly result: ToolResult;
readonly outcome: ToolExecutionOutcome;
readonly cancelled: boolean;
readonly durationMs: number;
}

Expand Down Expand Up @@ -111,6 +114,7 @@ export const toolExecutorDupTypeTurnIdKey = defineState<number | undefined>(
export class AgentToolExecutorService implements IAgentToolExecutorService {
declare readonly _serviceBrand: undefined;

private readonly outstandingEffects = new Set<OutstandingEffect>();
private readonly beforeExecuteEmitter = new BeforeToolExecuteEmitter();
readonly onBeforeExecuteTool: Event<BeforeToolExecuteEvent> = this.beforeExecuteEmitter.event;
private readonly willExecuteEmitter = new AsyncEmitter<WillExecuteToolEvent>();
Expand Down Expand Up @@ -302,7 +306,13 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
);

this.dispatchToolResult(call, finalized, options);
this.trackToolCall(call, finalized, timedResult.durationMs, options);
this.trackToolCall(
call,
finalized,
timedResult.durationMs,
options,
timedResult.cancelled || timedResult.outcome === 'aborted',
);

return {
toolCallId: call.toolCall.id,
Expand All @@ -316,8 +326,9 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
result: ToolResult,
durationMs: number,
options: ToolExecutorExecuteOptions,
cancelled: boolean,
): void {
const outcome = toolTelemetryOutcome(result);
const outcome = toolTelemetryOutcome(result, cancelled);
const toolCallId = call.toolCall.id;
const dupType = this.toolCallDupTypes.get(toolCallId) ?? 'normal';
this.toolCallDupTypes.delete(toolCallId);
Expand Down Expand Up @@ -385,6 +396,10 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
return settleError(call.args, call.output, 'preflight-rejected');
}

if (options.signal.aborted) {
return settleError(call.args, abortedToolOutput(call.toolName, options.signal), 'aborted');
}

let execution: ToolExecution;
try {
execution = await call.tool.resolveExecution(call.args);
Expand Down Expand Up @@ -458,7 +473,7 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
tasks: ToolExecutionTask[],
signal: AbortSignal,
): AsyncIterable<TimedToolResult> {
const scheduler = new ToolScheduler<TimedToolResult>();
const scheduler = new ToolScheduler<TimedToolResult>(this.outstandingEffects);
const allResults: Array<Promise<TimedToolResult>> = [];
const pendingResults = new Map<number, Promise<SettledTimedToolResult>>();

Expand All @@ -468,13 +483,24 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
accesses: task.accesses,
start: async () => {
const startedAt = Date.now();
const run = task.execute(signal);
return {
result: task.execute(signal).then(({ result, outcome }) => ({
index,
result,
outcome,
durationMs: Math.max(0, Date.now() - startedAt),
})),
result: run.then(({ result, outcome, cancelled, effectsSettled }) => {
if (effectsSettled !== undefined) {
this.trackOutstandingEffect(task.accesses, effectsSettled);
}
return {
index,
result,
outcome,
cancelled: cancelled === true,
durationMs: Math.max(0, Date.now() - startedAt),
};
}),
effectsSettled: run.then(
(value) => value.effectsSettled,
() => undefined,
),
};
},
});
Expand All @@ -501,6 +527,14 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
}
}

private trackOutstandingEffect(accesses: ToolAccesses, settled: Promise<void>): void {
const effect: OutstandingEffect = { accesses, settled };
this.outstandingEffects.add(effect);
void settled.finally(() => {
this.outstandingEffects.delete(effect);
});
}

private async runSingleExecution(
call: RunnableToolCall,
execution: RunnableToolExecution,
Expand All @@ -516,12 +550,14 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
abortedToolOutput(call.toolName, signal),
).result,
outcome: 'aborted',
cancelled: true,
};
}

let rawResult: ExecutableToolResult;
let executePromise: Promise<ExecutableToolResult>;
try {
const executePromise = execution.execute({
executePromise = execution.execute({
turnId: options.turnId,
toolCallId: call.toolCall.id,
trace: options.trace,
Expand All @@ -532,7 +568,20 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
this.dispatchToolProgress(call, update, options);
},
});
rawResult = await raceWithAbortGrace(executePromise, signal, call.toolName);
const raced = await raceWithAbortGrace(executePromise, signal);
if (raced.graceExpired) {
return {
result: makeErrorToolResult(call, call.args, abortedToolOutput(call.toolName, signal))
.result,
outcome: 'executed',
cancelled: true,
effectsSettled: executePromise.then(
() => undefined,
() => undefined,
),
};
}
rawResult = raced.value;
} catch (error) {
const aborted = isAbortError(error) || signal.aborted;
const output = aborted
Expand All @@ -541,6 +590,7 @@ export class AgentToolExecutorService implements IAgentToolExecutorService {
return {
result: makeErrorToolResult(call, call.args, output).result,
outcome: 'executed',
cancelled: aborted,
};
}

Expand Down Expand Up @@ -904,28 +954,19 @@ function normalizeToolResult(result: ExecutableToolResult): ToolResult {
return base;
}

function toolTelemetryOutcome(result: ToolResult): 'success' | 'error' | 'cancelled' {
if (result.isError !== true) return 'success';
const text = toolOutputText(result.output).toLowerCase();
return text.includes('aborted') ||
text.includes('cancelled') ||
text.includes('manually interrupted')
? 'cancelled'
: 'error';
function toolTelemetryOutcome(
result: ToolResult,
cancelled: boolean,
): 'success' | 'error' | 'cancelled' {
if (cancelled) return 'cancelled';
return result.isError === true ? 'error' : 'success';
}

function toolTelemetryErrorType(outcome: 'success' | 'error' | 'cancelled'): 'cancelled' | 'error' {
if (outcome === 'cancelled') return 'cancelled';
return 'error';
}

function toolOutputText(output: ToolResult['output']): string {
if (typeof output === 'string') return output;
return output
.filter((part): part is Extract<ContentPart, { type: 'text' }> => part.type === 'text')
.map((part) => part.text)
.join('');
}

function isMediaContentPart(part: ContentPart): boolean {
return part.type === 'image_url' || part.type === 'audio_url' || part.type === 'video_url';
Expand All @@ -938,21 +979,21 @@ function abortedToolOutput(toolName: string, signal: AbortSignal): string {
return `Tool "${toolName}" was aborted`;
}

type AbortGraceOutcome<Result> =
| { readonly graceExpired: false; readonly value: Result }
| { readonly graceExpired: true };

async function raceWithAbortGrace<Result>(
executePromise: Promise<Result>,
signal: AbortSignal,
toolName: string,
): Promise<Result> {
): Promise<AbortGraceOutcome<Result>> {
let graceTimer: ReturnType<typeof setTimeout> | undefined;
let onAbort: (() => void) | undefined;

const graceSentinel: Promise<Result> = new Promise((resolve) => {
const graceSentinel: Promise<AbortGraceOutcome<Result>> = new Promise((resolve) => {
const armTimer = (): void => {
graceTimer = setTimeout(() => {
resolve({
output: abortedToolOutput(toolName, signal),
isError: true,
} as unknown as Result);
resolve({ graceExpired: true });
}, ABORT_GRACE_MS);
};
if (signal.aborted) {
Expand All @@ -964,7 +1005,10 @@ async function raceWithAbortGrace<Result>(
});

try {
return await Promise.race([executePromise, graceSentinel]);
return await Promise.race([
executePromise.then((value) => ({ graceExpired: false, value }) as const),
graceSentinel,
]);
} finally {
if (graceTimer !== undefined) clearTimeout(graceTimer);
if (onAbort !== undefined) {
Expand Down
Loading
Loading