fix(agent-core-v2): use domain errors for wire replay conflicts

This commit is contained in:
7Sageer 2026-07-08 21:27:16 +08:00
commit 214def97cf
49 changed files with 1454 additions and 534 deletions

1
.gitignore vendored
View file

@ -19,6 +19,7 @@ plugins/cdn/
.worktrees/
.kimi-code/local.toml
.kimi-sandbox/
.vscode/
Dockerfile
docker-compose.yml

View file

@ -317,6 +317,7 @@ externalHooks --> config #34495E
externalHooks --> bootstrap #34495E
externalHooks --> plugin #34495E
externalHooks --> contextMemory #34495E
externalHooks --> session_context #34495E
sessionExternalHooks --> session_lifecycle #34495E
sessionExternalHooks --> agent_lifecycle #34495E
sessionExternalHooks --> config #34495E

File diff suppressed because one or more lines are too long

Before

Width:  |  Height:  |  Size: 276 KiB

After

Width:  |  Height:  |  Size: 276 KiB

Before After
Before After

View file

@ -13,6 +13,7 @@ export interface ContextCompactionInput {
readonly keptUserMessageCount?: number;
readonly keptHeadUserMessageCount?: number;
readonly droppedCount?: number;
readonly legacyTail?: boolean;
}
export interface ContextCompactionResult {

View file

@ -136,6 +136,7 @@ export class AgentContextMemoryService extends Disposable implements IAgentConte
keptUserMessageCount: result.keptUserMessageCount,
keptHeadUserMessageCount: result.keptHeadUserMessageCount,
droppedCount: result.droppedCount,
legacyTail: input.legacyTail,
}),
contextSizeMeasured({ length: result.messages.length, tokens: result.tokensAfter }),
);

View file

@ -165,6 +165,7 @@ interface ContextCompactionBasePayload {
readonly keptUserMessageCount?: number;
readonly keptHeadUserMessageCount?: number;
readonly droppedCount?: number;
readonly legacyTail?: boolean;
}
export interface TextSummaryCompactionPayload extends ContextCompactionBasePayload {
@ -226,7 +227,7 @@ export function readContextCompactionShapeInput(
keptUserMessageCount,
keptHeadUserMessageCount: readOptionalNumber(fields, 'keptHeadUserMessageCount'),
droppedCount: readOptionalNumber(fields, 'droppedCount'),
legacyTail: keptUserMessageCount === undefined,
legacyTail: readOptionalBoolean(fields, 'legacyTail') ?? keptUserMessageCount === undefined,
};
}
@ -275,6 +276,11 @@ function readOptionalString(record: UnknownRecord, key: string): string | undefi
return typeof value === 'string' ? value : undefined;
}
function readOptionalBoolean(record: UnknownRecord, key: string): boolean | undefined {
const value = record[key];
return typeof value === 'boolean' ? value : undefined;
}
function textOf(message: ContextMessage): string {
let text = '';
for (const part of message.content) {

View file

@ -237,8 +237,7 @@ function outputFromToolContent(content: readonly ContentPart[]): string | readon
}
const TOOL_INTERRUPTED_TEXT =
'<system>ERROR: Tool execution failed.</system>\n' +
'Tool execution was interrupted before its result was recorded. Do not assume the tool completed successfully.';
'Tool result is not available in the current context. Do not assume the tool completed successfully.';
// Shared inert filler for a call's slot while it awaits its recorded result;
// every slot still open at the end is overwritten with a synthetic result, so

View file

@ -10,8 +10,9 @@
* of its own). The requester-side `SubagentStart` / `SubagentStop` hooks are
* translated by the Session-scope `SessionExternalHooksService`, which observes
* the `agentLifecycle` run slots hosted on `IAgentLifecycleService`. Appends
* UserPromptSubmit hook results
* and Stop hook continuation prompts through `contextMemory`.
* UserPromptSubmit hook results and Stop hook continuation prompts through
* `contextMemory`, and passes the current session id from `sessionContext`
* into hook runner payloads.
*/
import { IInstantiationService } from '#/_base/di/instantiation';
@ -42,6 +43,7 @@ import type { ToolDidExecuteContext, ToolWillExecuteContext } from '#/agent/tool
import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor';
import { IAgentTurnService } from '#/agent/turn/turn';
import { toKimiErrorPayload } from '#/errors';
import { ISessionContext } from '#/session/sessionContext/sessionContext';
import { IAgentExternalHooksService } from './externalHooks';
import { IExternalHooksRunnerService } from '#/app/externalHooksRunner/externalHooksRunner';
@ -66,6 +68,7 @@ export class AgentExternalHooksService extends Disposable implements IAgentExter
@IAgentContextMemoryService private readonly context: IAgentContextMemoryService,
@IEventBus private readonly eventBus: IEventBus,
@IInstantiationService private readonly instantiation: IInstantiationService,
@ISessionContext private readonly sessionContext: ISessionContext,
) {
super();
this.registerListeners();
@ -82,7 +85,12 @@ export class AgentExternalHooksService extends Disposable implements IAgentExter
// output), and throwing here would clobber that with a finalize-abort error.
// The runner mirrors the legacy fire-and-forget behavior.
try {
void this.runner.fireAndForgetTrigger(event, { matcherValue, signal, inputData });
void this.runner.fireAndForgetTrigger(event, {
matcherValue,
signal,
sessionId: this.sessionContext.sessionId,
inputData,
});
} catch {}
}
@ -222,6 +230,7 @@ export class AgentExternalHooksService extends Disposable implements IAgentExter
const block = await this.runner.triggerBlock('PreToolUse', {
matcherValue: ctx.toolCall.name,
signal: ctx.signal,
sessionId: this.sessionContext.sessionId,
inputData: {
toolName: ctx.toolCall.name,
toolInput,
@ -260,6 +269,7 @@ export class AgentExternalHooksService extends Disposable implements IAgentExter
const results = await this.runner.trigger('UserPromptSubmit', {
matcherValue: input,
signal,
sessionId: this.sessionContext.sessionId,
inputData: { prompt: input, isSteer: ctx.isSteer },
});
signal.throwIfAborted();
@ -331,6 +341,7 @@ export class AgentExternalHooksService extends Disposable implements IAgentExter
const block = await this.runner.triggerBlock('Stop', {
signal: ctx.signal,
sessionId: this.sessionContext.sessionId,
inputData: { stopHookActive: false },
});
ctx.signal.throwIfAborted();
@ -343,6 +354,7 @@ export class AgentExternalHooksService extends Disposable implements IAgentExter
await this.runner.trigger('PreCompact', {
matcherValue: ctx.trigger,
signal,
sessionId: this.sessionContext.sessionId,
inputData: {
trigger: ctx.trigger,
tokenCount: ctx.tokenCount,

View file

@ -2,7 +2,7 @@ import { Disposable } from "#/_base/di/lifecycle";
import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { renderPrompt } from "#/_base/utils/render-prompt";
import { estimateTokensForMessages } from "#/_base/utils/tokens";
import { estimateTokensForMessage, estimateTokensForMessages } from "#/_base/utils/tokens";
import { buildCompactionSummaryText, isRealUserInput } from '#/agent/contextMemory/compactionHandoff';
import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory';
import type { ContextMessage } from '#/agent/contextMemory/types';
@ -19,6 +19,7 @@ import {
APIContextOverflowError,
APIEmptyResponseError,
APIStatusError,
isRetryableGenerateError,
} from '#/app/llmProtocol/errors';
import { createUserMessage, type Message } from '#/app/llmProtocol/message';
import { type TokenUsage } from '#/app/llmProtocol/usage';
@ -46,6 +47,7 @@ import {
} from './compactionOps';
import {
type CompactionBeginData,
type FullCompactionCompleteData,
type CompactionResult,
} from './types';
import { OrderedHookSlot } from '#/hooks';
@ -67,6 +69,8 @@ declare module '#/agent/wireRecord/wireRecord' {
export const MAX_COMPACTION_RETRY_ATTEMPTS = 5;
const DEFAULT_COMPACTION_MAX_COMPLETION_TOKENS = 128 * 1024;
const OVERFLOW_STATUS_RECOVERY_RATIO = 0.5;
const MAX_COMPACTION_OVERFLOW_SHRINK_ATTEMPTS = 3;
const COMPACTION_OVERFLOW_SHRINK_RATIOS = [0.7, 0.5, 0.35] as const;
type CompactionTelemetryProperties = Record<string, string | number | boolean | undefined>;
@ -158,6 +162,9 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
if (this.compactionCountInTurn > this.strategy.maxCompactionPerTurn) return false;
const history = this.context.get();
if (history.length === 0) {
throw new KimiError(ErrorCodes.COMPACTION_UNABLE, 'No messages to compact in current history.');
}
if (data.source === 'manual' && this.turn.getActiveTurn() !== undefined) {
throw new KimiError(
ErrorCodes.COMPACTION_UNABLE,
@ -165,10 +172,6 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
);
}
const tokenCount = estimateTokensForMessages(history);
const compactedCount = this.strategy.computeCompactCount(history, data.source);
if (compactedCount === 0) {
throw new KimiError(ErrorCodes.COMPACTION_UNABLE, 'No prefix that can be compacted in current history.');
}
this.wire.dispatch(fullCompactionBegin(data));
@ -190,7 +193,10 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
abortController.signal.addEventListener('abort', () => {
this.cancelActive(active);
}, { once: true });
void this.compactionWorker(active, data, compactedCount)
void this.compactionWorker(
active,
data,
)
.then(resolveCompaction, rejectCompaction);
void active.promise.catch(() => undefined);
return true;
@ -207,9 +213,9 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
return true;
}
private markCompleted(active: ActiveCompaction): boolean {
private markCompleted(active: ActiveCompaction, data: FullCompactionCompleteData): boolean {
if (this._compacting !== active) return false;
this.wire.dispatch(fullCompactionComplete({}));
this.wire.dispatch(fullCompactionComplete(data));
this._compacting = null;
return true;
}
@ -234,7 +240,9 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
next: () => Promise<void>,
): Promise<void> {
const isOverflow =
isContextOverflowError(context.error) || this.shouldRecoverFromPlain413(context.error);
isContextOverflowError(context.error) ||
findAPIStatusError(context.error) instanceof APIContextOverflowError ||
this.shouldRecoverFromPlain413(context.error);
if (!isOverflow) {
await next();
return;
@ -315,7 +323,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
try {
await active.promise;
} catch (error) {
if (active.abortController.signal.aborted || isAbortError(error)) return;
if (signal?.aborted === true && (active.abortController.signal.aborted || isAbortError(error))) return;
throw error;
}
}
@ -323,47 +331,18 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
private async compactionWorker(
active: ActiveCompaction,
data: Readonly<CompactionBeginData>,
initialCompactedCount: number,
): Promise<CompactionResult> {
try {
const finalResult: CompactionResult = {
summary: '',
contextSummary: '',
compactedCount: 1,
tokensBefore: 0,
tokensAfter: 0,
keptUserMessageCount: 0,
};
let compactedCount = initialCompactedCount;
for (let round = 1; ; round++) {
const result = await this.compactionRound(active, round, data, compactedCount);
if (this._compacting !== active) throw compactionCancelledReason(active);
finalResult.summary = result.summary;
finalResult.contextSummary = result.contextSummary;
finalResult.compactedCount += result.compactedCount - 1;
finalResult.tokensBefore += result.tokensBefore - finalResult.tokensAfter;
finalResult.tokensAfter = result.tokensAfter;
finalResult.keptUserMessageCount = result.keptUserMessageCount;
finalResult.keptHeadUserMessageCount = result.keptHeadUserMessageCount;
finalResult.droppedCount = result.droppedCount;
if (result.tokensBefore - result.tokensAfter < 1024) break;
if (!this.strategy.shouldBlock(result.tokensAfter)) break;
compactedCount = this.strategy.computeCompactCount(this.context.get(), data.source);
if (compactedCount === 0) break;
}
const result = await this.compactionRound(active, data);
if (this._compacting !== active) throw compactionCancelledReason(active);
this.lastCompactedTokenCount = finalResult.tokensAfter;
if (!this.markCompleted(active)) {
this.lastCompactedTokenCount = result.tokensAfter;
if (!this.markCompleted(active, completeData(result))) {
throw compactionCancelledReason(active);
}
const { contextSummary: _contextSummary, ...eventResult } = finalResult;
const { contextSummary: _contextSummary, ...eventResult } = result;
void _contextSummary;
this.eventBus.publish({ type: 'compaction.completed', result: eventResult, trigger: data.source });
return finalResult;
return result;
} catch (error) {
if (active.abortController.signal.aborted || isAbortError(error)) {
this.cancelActive(active);
@ -386,9 +365,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
private async compactionRound(
active: ActiveCompaction,
round: number,
data: Readonly<CompactionBeginData>,
initialCompactedCount: number,
): Promise<CompactionResult> {
const startedAt = Date.now();
const originalHistory = [...this.context.get()];
@ -396,15 +373,10 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
let retryCount = 0;
try {
let compactedCount = Math.min(initialCompactedCount, originalHistory.length);
const signal = active.abortController.signal;
signal.throwIfAborted();
// One logical compaction fires the hook once, even when it takes
// multiple window-sized rounds to bring the context under the ratio.
if (round === 1) {
await this.hooks.onWillCompact.run(active);
}
await this.hooks.onWillCompact.run(active);
const resolvedModel = this.profile.resolveModelContext();
const maxContextTokens = resolvedModel.modelCapabilities.max_context_tokens;
@ -420,8 +392,12 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
const delays = retryBackoffDelays(MAX_COMPACTION_RETRY_ATTEMPTS);
let attempt: CompactionAttemptResult | undefined;
let historyForModel: readonly ContextMessage[] = originalHistory;
let droppedCount = 0;
let overflowShrinkCount = 0;
let emptyOrTruncatedShrinkCount = 0;
while (true) {
const messagesToCompact = originalHistory.slice(0, compactedCount);
const messagesToCompact = historyForModel;
// Raw context slice — `llmRequester` projects every request once;
// projecting here too would run micro-compaction on shifted indices.
const messages: Message[] = [...messagesToCompact, createUserMessage(instruction)];
@ -434,10 +410,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
maxOutputSize: compactionMaxOutputSize,
source: { type: 'operation', requestKind: 'full_compaction' },
retry: {
maxAttempts: MAX_COMPACTION_RETRY_ATTEMPTS,
onRetry: () => {
retryCount += 1;
},
maxAttempts: 1,
},
},
undefined,
@ -446,19 +419,38 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
);
break;
} catch (error) {
if (
this.shouldRecoverFromCompactionOverflow(error, messages) ||
error instanceof CompactionTruncatedError ||
error instanceof APIEmptyResponseError
) {
const reduced = this.strategy.reduceCompactOnOverflow(messagesToCompact);
// An overflow that cannot shrink further would replay the same
// request; give up (v1: throws when the history cannot shrink).
if (error instanceof APIContextOverflowError && reduced >= compactedCount) {
if (this.shouldRecoverFromCompactionOverflow(error, messages)) {
overflowShrinkCount += 1;
if (
overflowShrinkCount > MAX_COMPACTION_OVERFLOW_SHRINK_ATTEMPTS ||
messagesToCompact.length <= 1
) {
throw error;
}
compactedCount = reduced;
} else {
const before = messagesToCompact.length;
historyForModel = shrinkCompactionHistoryAfterOverflow(
messagesToCompact,
overflowShrinkCount,
);
droppedCount += before - historyForModel.length;
retryCount = 0;
continue;
}
if (
(error instanceof CompactionTruncatedError || error instanceof APIEmptyResponseError) &&
messagesToCompact.length > 1
) {
emptyOrTruncatedShrinkCount += 1;
if (emptyOrTruncatedShrinkCount > MAX_COMPACTION_RETRY_ATTEMPTS) {
throw error;
}
const reduced = dropOldestMessageAndLeadingToolResults(messagesToCompact);
droppedCount += messagesToCompact.length - reduced.length;
historyForModel = reduced;
retryCount = 0;
continue;
}
if (!isRetryableGenerateError(error)) {
throw error;
}
if (retryCount + 1 >= MAX_COMPACTION_RETRY_ATTEMPTS) {
@ -487,8 +479,9 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
const result = this.context.applyCompaction({
summary,
contextSummary: buildCompactionSummaryText(summary),
compactedCount,
compactedCount: originalHistory.length,
tokensBefore,
droppedCount: droppedCount === 0 ? undefined : droppedCount,
});
this.telemetry.track('compaction_finished', {
@ -498,9 +491,10 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
tokens_after: result.tokensAfter,
duration_ms: Date.now() - startedAt,
compacted_count: result.compactedCount,
dropped_count: result.droppedCount,
retry_count: retryCount,
round,
thinking_level: this.profile.data().thinkingLevel,
round: 1,
thinking_effort: this.profile.data().thinkingLevel,
...usageTelemetry(attempt.usage),
});
return result;
@ -510,9 +504,9 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
source: data.source,
tokens_before: tokensBefore,
duration_ms: Date.now() - startedAt,
round,
round: 1,
retry_count: retryCount,
thinking_level: this.profile.data().thinkingLevel,
thinking_effort: this.profile.data().thinkingLevel,
error_type: error instanceof Error ? error.name : 'Unknown',
});
if (isKimiError(error) && error.code === ErrorCodes.AUTH_LOGIN_REQUIRED) throw error;
@ -540,7 +534,8 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
error: unknown,
estimatedRequestTokens = this.tokenCountWithPending(),
): boolean {
if (!(error instanceof APIStatusError) || error.statusCode !== 413) return false;
const statusError = findAPIStatusError(error);
if (statusError === undefined || statusError.statusCode !== 413) return false;
const maxContextTokens = this.profile.getModelCapabilities().max_context_tokens;
return (
maxContextTokens > 0 &&
@ -557,6 +552,17 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull
}
}
function findAPIStatusError(error: unknown): APIStatusError | undefined {
let current: unknown = error;
const seen = new Set<unknown>();
while (current !== undefined && current !== null && !seen.has(current)) {
if (current instanceof APIStatusError) return current;
seen.add(current);
current = current instanceof Error ? current.cause : undefined;
}
return undefined;
}
function collectSummary(finish: LLMRequestFinish): CompactionAttemptResult {
if (finish.providerFinishReason === 'truncated') {
throw new CompactionTruncatedError();
@ -576,6 +582,17 @@ function collectSummary(finish: LLMRequestFinish): CompactionAttemptResult {
return { summary, usage: finish.usage };
}
function completeData(result: CompactionResult): FullCompactionCompleteData {
return {
compactedCount: result.compactedCount,
tokensBefore: result.tokensBefore,
tokensAfter: result.tokensAfter,
keptUserMessageCount: result.keptUserMessageCount,
keptHeadUserMessageCount: result.keptHeadUserMessageCount,
droppedCount: result.droppedCount,
};
}
function historySafeToCompact(
current: readonly ContextMessage[],
original: readonly ContextMessage[],
@ -585,6 +602,49 @@ function historySafeToCompact(
return current.slice(original.length).every(isRealUserInput);
}
function shrinkCompactionHistoryAfterOverflow<T extends Message>(
messages: readonly T[],
attempt: number,
): T[] {
if (messages.length <= 1) return messages.slice();
const ratio = COMPACTION_OVERFLOW_SHRINK_RATIOS[
Math.min(attempt - 1, COMPACTION_OVERFLOW_SHRINK_RATIOS.length - 1)
]!;
const tokenBudget = Math.floor(estimateTokensForMessages(messages) * ratio);
return takeRecentMessagesWithinTokenBudget(messages, tokenBudget);
}
function takeRecentMessagesWithinTokenBudget<T extends Message>(
messages: readonly T[],
tokenBudget: number,
): T[] {
let start = messages.length;
let tokens = 0;
for (let i = messages.length - 1; i >= 0; i--) {
const messageTokens = estimateTokensForMessage(messages[i]!);
if (tokens + messageTokens > tokenBudget) break;
tokens += messageTokens;
start = i;
}
if (start === 0) start = 1;
return dropLeadingToolResults(messages.slice(start));
}
function dropOldestMessageAndLeadingToolResults<T extends { readonly role: string }>(
messages: readonly T[],
): T[] {
if (messages.length <= 1) return messages.slice();
return dropLeadingToolResults(messages.slice(1));
}
function dropLeadingToolResults<T extends { readonly role: string }>(messages: readonly T[]): T[] {
let start = 0;
while (start < messages.length && messages[start]!.role === 'tool') {
start += 1;
}
return messages.slice(start);
}
function usageTelemetry(usage: TokenUsage | null): CompactionTelemetryProperties {
if (usage === null) return {};
return {

View file

@ -8,6 +8,7 @@ export const ProfileErrors = {
codes: {
MODEL_NOT_CONFIGURED: 'model.not_configured',
MODEL_CONFIG_INVALID: 'model.config_invalid',
THINKING_ALIAS_CONFLICT: 'profile.thinking_alias_conflict',
},
} as const satisfies ErrorDomain;

View file

@ -4,8 +4,23 @@ import type { ThinkingEffort } from '#/app/llmProtocol/thinkingEffort';
import type { Model } from '#/app/model/modelInstance';
import { createDecorator } from "#/_base/di/instantiation";
import type { ErrorCode } from '#/_base/errors/codes';
import { KimiError } from '#/_base/errors/errors';
import type { ToolSource } from '#/agent/tool/toolContract';
import { ProfileErrors } from './errors';
export { ProfileErrors } from './errors';
export type ProfileErrorCode = (typeof ProfileErrors.codes)[keyof typeof ProfileErrors.codes];
export class ProfileError extends KimiError {
constructor(code: ProfileErrorCode, message: string, details?: Record<string, unknown>) {
super(code as ErrorCode, message, { details });
this.name = 'ProfileError';
}
}
/**
* Data required to configure an agent: active model id, its capability
* matrix, profile, thinking level, system prompt, and working directory.

View file

@ -29,10 +29,11 @@
*/
import type { ThinkingEffort } from '#/app/llmProtocol/thinkingEffort';
import { ErrorCodes, KimiError } from '#/errors';
import { defineModel } from '#/wire/model';
import { defineOp } from '#/wire/op';
import { ProfileError, ProfileErrors } from './profile';
export interface ProfileModelState {
readonly cwd?: string;
readonly modelAlias?: string;
@ -81,15 +82,13 @@ export const configUpdate = defineOp(ProfileModel, 'config.update', {
function configUpdateThinkingLevel(p: ConfigUpdatePayload): ThinkingEffort | undefined {
if (p.thinkingEffort !== undefined && p.thinkingLevel !== undefined) {
if (p.thinkingEffort !== p.thinkingLevel) {
throw new KimiError(
ErrorCodes.REQUEST_INVALID,
throw new ProfileError(
ProfileErrors.codes.THINKING_ALIAS_CONFLICT,
`config.update has conflicting thinkingEffort (${p.thinkingEffort}) and legacy thinkingLevel (${p.thinkingLevel})`,
{
details: {
type: 'config.update',
thinkingEffort: p.thinkingEffort,
thinkingLevel: p.thinkingLevel,
},
type: 'config.update',
thinkingEffort: p.thinkingEffort,
thinkingLevel: p.thinkingLevel,
},
);
}

View file

@ -68,7 +68,7 @@ type AgentTaskNotification = Record<string, unknown> & {
readonly id: string;
readonly category: 'task';
readonly type: string;
readonly source_kind: 'task';
readonly source_kind: 'background_task';
readonly source_id: string;
readonly agent_id?: string | undefined;
readonly title: string;
@ -93,6 +93,10 @@ interface ManagedTask {
readonly outputChunks: string[];
outputSizeBytes: number;
retainedOutputBytes: number;
/**
* True once a command has crossed `MAX_TASK_OUTPUT_BYTES` and termination has
* been requested. One-shot guard so the ceiling fires exactly once.
*/
outputLimitTripped: boolean;
status: AgentTaskStatus;
options: RegisterAgentTaskOptions & { description?: string };
@ -116,8 +120,30 @@ interface ManagedTask {
handleSubscription?: { dispose(): void };
}
const MAX_OUTPUT_BYTES = 1024 * 1024;
const MAX_TASK_OUTPUT_BYTES = 16 * 1024 * 1024;
const MAX_OUTPUT_BYTES = 1024 * 1024; // 1 MiB
/**
* Hard ceiling on the combined output a single shell command may stream before
* it is force-terminated (SIGTERM grace SIGKILL). It guards both the
* live-forward path and the on-disk `output.log` write chain from a runaway
* command (e.g. `b3sum --length <huge>`) whose output would otherwise grow
* without bound filling the disk, or retaining each pending-write chunk until
* Node aborts with an out-of-memory crash. Scoped to process tasks (foreground
* and background); subagent and user-question results are appended once and must
* always be persisted, so they are intentionally not capped here.
*/
const MAX_TASK_OUTPUT_BYTES = 16 * 1024 * 1024; // 16 MiB
/** Terminal `stopReason` recorded when a command trips the output ceiling. */
function outputLimitReason(): string {
const mib = Math.floor(MAX_TASK_OUTPUT_BYTES / (1024 * 1024));
return (
`Output limit exceeded: the command produced more than ${mib} MiB and was ` +
'terminated. Redirect large output to a file (e.g. `command > out.txt`) and ' +
'inspect it in slices instead.'
);
}
const SIGTERM_GRACE_MS = 5_000;
const TASK_ID_ALPHABET = '0123456789abcdefghijklmnopqrstuvwxyz';
const USER_INTERRUPT_REASON = 'Interrupted by user';
@ -660,6 +686,9 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
await entry.persistWriteQueue;
return this.toInfo(entry);
}
if (timeoutMs <= 0) {
return this.toInfo(entry);
}
let waiter: (() => void) | undefined;
let timeout: ReturnType<typeof setTimeout> | undefined;
@ -717,7 +746,7 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
if (maxRunningTasks === undefined) return;
if (!detached) return;
if (this.activeTaskCount() < maxRunningTasks) return;
throw new Error('Too many detached tasks are already running.');
throw new Error('Too many background tasks are already running.');
}
private taskConfig(): AgentTaskConfig | undefined {
@ -774,6 +803,13 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
entry.outputSizeBytes += chunkBytes;
this.appendRetainedOutput(entry, chunk, chunkBytes);
// Output ceiling: a single shell command must not grow the (unbounded)
// live-forward buffer or the on-disk write chain until the process runs out
// of memory or fills the disk. Trip once, then request graceful termination
// through the shared stop path (SIGTERM → grace → SIGKILL). Scoped to
// process tasks (foreground and background): subagent and user-question tasks
// append their bounded result in one shot and must always persist it, so they
// are intentionally not capped here.
if (
!entry.outputLimitTripped &&
entry.task?.kind === 'process' &&
@ -783,6 +819,11 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
void this.stop(entry.taskId, outputLimitReason());
}
// Once the cap has tripped the task is being terminated: keep only the
// bounded in-memory ring buffer above and stop feeding the (unbounded) disk
// write chain. A producer that ignores SIGTERM could otherwise keep the
// chain — and the chunk strings each pending write retains — growing through
// the grace window until SIGKILL, re-introducing the OOM this cap prevents.
if (entry.outputLimitTripped) return;
if (!entry.outputPersistStarted) {
@ -874,16 +915,16 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
private recordTaskStarted(info: AgentTaskInfo): void {
this.wire.dispatch(taskStarted({ info }));
this.telemetry.track('task_created', {
this.telemetry.track('background_task_created', {
kind: info.kind === 'process' ? 'bash' : info.kind,
});
}
private recordTaskTerminated(info: AgentTaskInfo): void {
this.wire.dispatch(taskTerminated({ info }));
this.telemetry.track('task_completed', {
this.telemetry.track('background_task_completed', {
kind: info.kind,
duration: info.endedAt !== null ? info.endedAt - info.startedAt : null,
duration_ms: info.endedAt !== null ? info.endedAt - info.startedAt : null,
status: info.status,
});
}
@ -945,10 +986,10 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
id: origin.notificationId,
category: 'task',
type: `task.${info.status}`,
source_kind: 'task',
source_kind: 'background_task',
source_id: info.taskId,
agent_id: info.kind === 'agent' ? info.agentId : undefined,
title: `Task ${info.kind} ${info.status}`,
title: `Background ${info.kind} ${info.status}`,
severity: info.status === 'completed' ? 'info' : 'warning',
body: buildAgentTaskNotificationBody(info),
children: agentTaskNotificationChildren(output),
@ -981,7 +1022,7 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
);
}
private markDeliveredNotification(origin: TaskOrigin): void {
private markDeliveredNotification(origin: TaskNotificationOrigin): void {
this.deliveredNotificationKeys.add(notificationKey(origin));
}
@ -1045,15 +1086,6 @@ function emptyOutputSnapshot(): AgentTaskOutputSnapshot {
};
}
function outputLimitReason(): string {
const mib = Math.floor(MAX_TASK_OUTPUT_BYTES / (1024 * 1024));
return (
`Output limit exceeded: the command produced more than ${String(mib)} MiB and was ` +
'terminated. Redirect large output to a file (e.g. `command > out.txt`) and ' +
'inspect it in slices instead.'
);
}
function agentTaskNotificationChildren(
output: AgentTaskOutputSnapshot,
): readonly string[] | undefined {
@ -1105,22 +1137,24 @@ function newerRestoredTask(
return loaded;
}
function isTaskOrigin(origin: unknown): origin is TaskOrigin {
type TaskNotificationOrigin = Pick<TaskOrigin, 'taskId' | 'status' | 'notificationId'>;
function isTaskOrigin(origin: unknown): origin is TaskNotificationOrigin {
if (typeof origin !== 'object' || origin === null) return false;
const value = origin as Record<string, unknown>;
return (
value['kind'] === 'task' &&
(value['kind'] === 'background_task' || value['kind'] === 'task') &&
typeof value['taskId'] === 'string' &&
typeof value['status'] === 'string' &&
typeof value['notificationId'] === 'string'
);
}
function notificationKey(origin: TaskOrigin): string {
function notificationKey(origin: TaskNotificationOrigin): string {
return `${origin.taskId}\0${origin.status}\0${origin.notificationId}`;
}
function taskOriginsFromRecord(record: WireRecord): readonly TaskOrigin[] {
function taskOriginsFromRecord(record: WireRecord): readonly TaskNotificationOrigin[] {
const raw = record as {
readonly type: string;
readonly message?: unknown;
@ -1135,7 +1169,7 @@ function taskOriginsFromRecord(record: WireRecord): readonly TaskOrigin[] {
return [];
}
function taskOriginFromMessage(message: unknown): readonly TaskOrigin[] {
function taskOriginFromMessage(message: unknown): readonly TaskNotificationOrigin[] {
if (typeof message !== 'object' || message === null) return [];
const origin = (message as { readonly origin?: unknown }).origin;
return isTaskOrigin(origin) ? [origin] : [];
@ -1157,7 +1191,7 @@ function buildAgentTaskNotificationBody(info: AgentTaskInfo): string {
const recovery = [
'',
`To recover or continue this subagent, call Agent(resume="${agentId}", prompt="Pick up where you left off; redo the last tool call if its result was never observed.").`,
`Use agent_id ("${agentId}"), NOT source_id / task_id ("${info.taskId}") because the two look alike but only agent_id is accepted by the resume parameter.`,
`Use agent_id ("${agentId}"), NOT source_id / task_id ("${info.taskId}") the two look alike but only agent_id is accepted by the resume parameter.`,
'Add run_in_background=true to keep it backgrounded, or omit it to take the result inline in the current turn.',
'The subagent retains its full prior context across the restart, but any in-flight tool call lost its result and may need to be redone.',
].join('\n');

View file

@ -2,8 +2,9 @@ List background tasks and their current status.
Use this tool to discover which background tasks exist and where each one
stands. It is the entry point for inspecting background work: it returns a
task ID, status, command, description, PID, exit code, and stop reason
when available.
task ID, status, and description for every task it reports, plus the command,
PID, and (once finished) exit code for shell tasks, and a stop reason for any
task that ended early.
Guidelines:

View file

@ -7,7 +7,7 @@ Guidelines:
- Do not use TaskOutput to wait for a result you need before continuing — if your next step depends on the task's result, run that task in the foreground instead. TaskOutput is for a deliberate progress check you will act on without blocking, not a way to sit and wait for a background task you just launched.
- By default this tool is non-blocking and returns a current status/output snapshot.
- Use block=true only when you intentionally want to wait for completion or timeout.
- This tool returns structured task metadata, a fixed-size output preview, and, when available, an `output_path` for the full log.
- This tool returns structured task metadata, a fixed-size output preview, and an output_path for the full log.
- For a terminal task, the metadata also explains why it ended. A shell command that runs to completion reports `status: completed` on a zero exit, or `status: failed` with its non-zero `exit_code` — judge that failure from the `exit_code`, because a plain command failure carries no `stop_reason` and no `terminal_reason`. `terminal_reason` is a categorical label emitted only when the end is not an ordinary exit: `timed_out` when the deadline aborted it, `stopped` when it was explicitly stopped, or `failed` when it errored without producing an exit code; the `stopped` and `failed` cases also carry a human-readable `stop_reason`. A task that finished on its own with a clean exit carries neither `stop_reason` nor `terminal_reason`.
- When `output_path` is present, the full, never-truncated log is available there; use the `Read` tool with that path to page through it, whether or not the preview was truncated.
- The full, never-truncated log is always available at output_path; use the `Read` tool with that path to page through it, whether or not the preview was truncated.
- This tool works with the generic background task system and should remain the primary read path for future task types, not just bash.

View file

@ -41,7 +41,7 @@ const PAGING_HINT_LINES = 300;
// ── Input schema ─────────────────────────────────────────────────────
export const TaskOutputInputSchema = z.object({
task_id: z.string().describe('The task ID to inspect.'),
task_id: z.string().describe('The background task ID to inspect.'),
block: z
.boolean()
.default(false)

View file

@ -16,7 +16,7 @@ import TASK_STOP_DESCRIPTION from './task-stop.md?raw';
// ── Input schema ─────────────────────────────────────────────────────
export const TaskStopInputSchema = z.object({
task_id: z.string().describe('The task ID to stop.'),
task_id: z.string().describe('The background task ID to stop.'),
reason: z
.string()
.default('Stopped by TaskStop')

View file

@ -27,6 +27,7 @@ export type ExecutableToolResultBuilderResult = (
readonly output: string;
readonly message: string;
readonly truncated: boolean;
readonly brief?: string;
};
export class ToolResultBuilder {
@ -103,7 +104,7 @@ export class ToolResultBuilder {
return charsWritten;
}
ok(message = ''): ExecutableToolResultBuilderResult {
ok(message = '', options: { readonly brief?: string } = {}): ExecutableToolResultBuilderResult {
let finalMessage = message;
if (finalMessage.length > 0 && !finalMessage.endsWith('.')) {
finalMessage += '.';
@ -127,10 +128,14 @@ export class ToolResultBuilder {
: output,
message: finalMessage,
truncated: this.truncationHappened,
brief: options.brief,
};
}
error(message: string): ExecutableToolResultBuilderResult {
error(
message: string,
options: { readonly brief?: string } = {},
): ExecutableToolResultBuilderResult {
const finalMessage = this.truncationHappened
? message.length === 0
? TRUNCATION_MESSAGE
@ -149,6 +154,7 @@ export class ToolResultBuilder {
: `${output}\n${finalMessage}`,
message: finalMessage,
truncated: this.truncationHappened,
brief: options.brief,
};
}
}

View file

@ -0,0 +1,13 @@
/**
* `usage` domain error codes invalid persisted usage records.
*/
import { registerErrorDomain, type ErrorDomain } from '#/_base/errors/codes';
export const UsageErrors = {
codes: {
TURN_ID_CONFLICT: 'usage.turn_id_conflict',
},
} as const satisfies ErrorDomain;
registerErrorDomain(UsageErrors);

View file

@ -2,6 +2,21 @@ import type { LLMRequestSource } from '#/agent/llmRequester/llmRequester';
import type { TokenUsage } from '#/app/llmProtocol/usage';
import { createDecorator } from '#/_base/di/instantiation';
import type { ErrorCode } from '#/_base/errors/codes';
import { KimiError } from '#/_base/errors/errors';
import { UsageErrors } from './errors';
export { UsageErrors } from './errors';
export type UsageErrorCode = (typeof UsageErrors.codes)[keyof typeof UsageErrors.codes];
export class UsageError extends KimiError {
constructor(code: UsageErrorCode, message: string, details?: Record<string, unknown>) {
super(code as ErrorCode, message, { details });
this.name = 'UsageError';
}
}
export interface UsageStatus {
readonly byModel?: Record<string, TokenUsage>;

View file

@ -15,11 +15,10 @@
import { addUsage, type TokenUsage } from '#/app/llmProtocol/usage';
import type { LLMRequestSource } from '#/agent/llmRequester/llmRequester';
import type { AgentPhase } from '#/agent/runtime/runtime';
import { ErrorCodes, KimiError } from '#/errors';
import { defineModel } from '#/wire/model';
import { defineOp } from '#/wire/op';
import type { UsageStatus } from './usage';
import { UsageError, UsageErrors, type UsageStatus } from './usage';
export type UsageRecordScope = 'session' | 'turn';
@ -88,15 +87,13 @@ function turnIdFromUsagePayload(p: UsageRecordPayload): number | undefined {
const legacyContext = p.context;
const legacyTurnId = legacyContext?.type === 'turn' ? legacyContext.turnId : undefined;
if (p.turnId !== undefined && legacyTurnId !== undefined && p.turnId !== legacyTurnId) {
throw new KimiError(
ErrorCodes.REQUEST_INVALID,
throw new UsageError(
UsageErrors.codes.TURN_ID_CONFLICT,
`usage.record has conflicting turnId (${p.turnId}) and legacy context turnId (${legacyTurnId})`,
{
details: {
type: 'usage.record',
turnId: p.turnId,
legacyTurnId,
},
type: 'usage.record',
turnId: p.turnId,
legacyTurnId,
},
);
}

View file

@ -48,18 +48,11 @@ export class MoonshotWebSearchProvider implements WebSearchProvider {
async search(
query: string,
options?: {
limit?: number;
includeContent?: boolean;
toolCallId?: string;
signal?: AbortSignal;
},
): Promise<WebSearchResult[]> {
const body = {
text_query: query,
limit: options?.limit ?? 5,
enable_page_crawling: options?.includeContent ?? false,
timeout_seconds: 30,
};
const body = { text_query: query };
const bodyJson = JSON.stringify(body);
const toolCallId = options?.toolCallId;
@ -90,7 +83,6 @@ export class MoonshotWebSearchProvider implements WebSearchProvider {
};
if (typeof r.date === 'string' && r.date.length > 0) out.date = r.date;
if (typeof r.site_name === 'string' && r.site_name.length > 0) out.siteName = r.site_name;
if (typeof r.content === 'string' && r.content.length > 0) out.content = r.content;
return out;
});
}

View file

@ -36,15 +36,12 @@ export interface WebSearchResult {
snippet: string;
date?: string;
siteName?: string;
content?: string;
}
export interface WebSearchProvider {
search(
query: string,
options?: {
limit?: number;
includeContent?: boolean;
toolCallId?: string;
signal?: AbortSignal;
},
@ -55,23 +52,6 @@ export interface WebSearchProvider {
export const WebSearchInputSchema = z.object({
query: z.string().describe('The query text to search for.'),
limit: z
.number()
.int()
.min(1)
.max(20)
.default(5)
.describe(
'The number of results to return. Typically you do not need to set this value. When the results do not contain what you need, you probably want to give a more concrete query.',
)
.optional(),
include_content: z
.boolean()
.default(false)
.describe(
'Whether to include the content of the web pages in the results. It can consume a large amount of tokens when this is set to true. You should avoid enabling this when `limit` is set to a large value.',
)
.optional(),
});
export type WebSearchInput = z.infer<typeof WebSearchInputSchema>;
@ -102,18 +82,7 @@ export class WebSearchTool implements BuiltinTool<WebSearchInput> {
{ toolCallId, signal }: ExecutableToolContext,
): Promise<ExecutableToolResult> {
try {
const opts: {
limit?: number;
includeContent?: boolean;
toolCallId?: string;
signal?: AbortSignal;
} = {
toolCallId,
signal,
};
if (args.limit !== undefined) opts.limit = args.limit;
if (args.include_content !== undefined) opts.includeContent = args.include_content;
const results = await this.provider.search(args.query, opts);
const results = await this.provider.search(args.query, { toolCallId, signal });
const builder = new ToolResultBuilder({ maxLineLength: null });
if (results.length === 0) {
@ -131,7 +100,6 @@ export class WebSearchTool implements BuiltinTool<WebSearchInput> {
if (result.date) builder.write(`Date: ${result.date}\n`);
builder.write(`URL: ${result.url}\n`);
builder.write(`Snippet: ${result.snippet}\n\n`);
if (result.content) builder.write(`${result.content}\n\n`);
}
// Keep the citation reminder next to the data (not just in the static tool

View file

@ -70,15 +70,20 @@ export class FetchURLTool implements BuiltinTool<FetchURLInput> {
}
const builder = new ToolResultBuilder({ maxLineLength: null });
builder.write(content);
// Tell the LLM whether it received the whole body or only the
// extracted article text, so it can judge how complete the
// content is.
const message =
// Tell the LLM whether it received the whole body or only the extracted
// article text, so it can judge how complete the content is, and remind it
// to cite this page when it uses the content. Both notes must ride in
// `output`: the result's `message` field is dropped from the transcript, so
// `output` is the only place the model can read them. Put them at the front
// so they survive any downstream truncation of the body.
const note =
kind === 'passthrough'
? 'The returned content is the full response body, returned verbatim.'
: 'The returned content is the main text extracted from the page.';
return builder.ok(message);
const citeReminder =
'If you use it in your answer, cite this page as a markdown link, e.g. [title](url).';
builder.write(`${note} ${citeReminder}\n\n${content}`);
return builder.ok();
} catch (error) {
// An in-flight abort rejects the signal-aware fetch promptly. Re-throw
// so the executor can classify it (including user cancellation) and

View file

@ -30,6 +30,7 @@ import { SessionErrors } from '#/session/errors';
import { SkillErrors } from '#/app/skillCatalog/errors';
import { TerminalErrors } from '#/os/interface/terminalErrors';
import { TurnErrors } from '#/agent/turn/errors';
import { UsageErrors } from '#/agent/usage/errors';
import { WireRecordErrors } from '#/agent/wireRecord/errors';
export * from '#/_base/errors/codes';
@ -59,6 +60,7 @@ export { SessionErrors } from '#/session/errors';
export { SkillErrors } from '#/app/skillCatalog/errors';
export { TerminalErrors } from '#/os/interface/terminalErrors';
export { TurnErrors } from '#/agent/turn/errors';
export { UsageErrors } from '#/agent/usage/errors';
export { WireRecordErrors } from '#/agent/wireRecord/errors';
export const ErrorCodes = {
@ -85,5 +87,6 @@ export const ErrorCodes = {
...SkillErrors.codes,
...TerminalErrors.codes,
...TurnErrors.codes,
...UsageErrors.codes,
...WireRecordErrors.codes,
} as const;

View file

@ -38,7 +38,10 @@ import { ISessionContext } from '#/session/sessionContext/sessionContext';
import { ISessionProcessRunner, type IProcess } from '#/session/process/processRunner';
import { IAgentProfileService } from '#/agent/profile/profile';
import type { BuiltinTool, ExecutableToolResult, ToolExecution, ToolUpdate } from '#/agent/tool/toolContract';
import { ToolResultBuilder } from '#/agent/tool/result-builder';
import {
type ExecutableToolResultBuilderResult,
ToolResultBuilder,
} from '#/agent/tool/result-builder';
import { registerTool } from '#/agent/toolRegistry/toolContribution';
import { toInputJsonSchema } from '#/_base/tools/support/input-schema';
import { literalRulePattern, matchesGlobRuleSubject } from '#/_base/tools/support/rule-match';
@ -146,7 +149,7 @@ function renderBashDescription(shellName: string): string {
function withoutBackgroundDescription(description: string): string {
return description
.replace(
/\n\nIf `run_in_background=true`,[\s\S]*?point them to the `\/tasks` command, which opens an interactive panel; it has no subcommands\./,
/\r?\n\r?\nIf `run_in_background=true`,[\s\S]*?point them to the `\/tasks` command, which opens an interactive panel; it has no subcommands\./,
'\n\nBackground execution is disabled for this agent. Do not set `run_in_background=true`.',
)
.replace(
@ -154,7 +157,7 @@ function withoutBackgroundDescription(description: string): string {
` For possibly long-running commands, set the \`timeout\` argument in seconds. The default is ${String(DEFAULT_TIMEOUT_S)}s; foreground commands allow up to ${String(MAX_TIMEOUT_S)}s.`,
)
.replace(
/\n- Prefer `run_in_background=true`[\s\S]*?conversation to continue before the command finishes\./,
/\r?\n- Prefer `run_in_background=true`[\s\S]*?conversation to continue before the command finishes\./,
'\n- Do not set `run_in_background=true`; background task management tools are not available.',
);
}
@ -179,7 +182,11 @@ export class BashTool implements BuiltinTool<BashInput> {
}
private allowBackground(): boolean {
return this.profile.isToolActive('TaskOutput') && this.profile.isToolActive('TaskStop');
return (
this.profile.isToolActive('TaskList') &&
this.profile.isToolActive('TaskOutput') &&
this.profile.isToolActive('TaskStop')
);
}
get description(): string {
@ -286,6 +293,9 @@ export class BashTool implements BuiltinTool<BashInput> {
{
detached: startsInBackground,
timeoutMs,
// Detaching (ctrl+b) moves a foreground command to the background;
// give it the background timeout so it is not still bounded by the
// shorter foreground deadline.
detachTimeoutMs: DEFAULT_BACKGROUND_TIMEOUT_S * MS_PER_SECOND,
signal: startsInBackground ? undefined : signal,
},
@ -300,11 +310,14 @@ export class BashTool implements BuiltinTool<BashInput> {
};
}
// Foreground `!` shell commands surface their task id so the TUI can detach
// (ctrl+b) this exact task. Background runs are already detached.
if (!startsInBackground) onForegroundTaskStart?.(taskId);
if (startsInBackground) {
return this.detachedTaskResult(taskId, proc, description, {
title: 'Task started in background',
return this.backgroundStartedResult(taskId, proc, description, {
title: 'Background task started',
brief: `Started ${taskId}`,
});
}
@ -312,19 +325,20 @@ export class BashTool implements BuiltinTool<BashInput> {
const release = await this.tasks.waitForForegroundRelease(taskId);
if (release === 'detached') {
collectForegroundOutput = false;
return this.detachedTaskResult(
return this.backgroundStartedResult(
taskId,
proc,
description,
{
title: 'Task moved to background',
brief: `Backgrounded ${taskId}`,
},
builder,
'foreground_detached',
);
}
return this.foregroundCompletionResult(taskId, proc, builder, foregroundTimeoutMs);
return await this.foregroundCompletionResult(taskId, proc, builder, foregroundTimeoutMs);
} finally {
collectForegroundOutput = false;
}
@ -353,46 +367,65 @@ export class BashTool implements BuiltinTool<BashInput> {
return undefined;
}
private foregroundCompletionResult(
private async foregroundCompletionResult(
taskId: string,
proc: IProcess,
builder: ToolResultBuilder,
foregroundTimeoutMs: number,
): ExecutableToolResult {
): Promise<ExecutableToolResult> {
const current = this.tasks.getTask(taskId);
const exitCode = current?.kind === 'process' ? current.exitCode : proc.exitCode;
let result: ExecutableToolResultBuilderResult;
if (current?.status === 'timed_out') {
const timeoutLabel = formatTimeoutLabel(foregroundTimeoutMs);
return builder.error(`Command killed by timeout (${timeoutLabel})`);
}
if (current?.status === 'killed' && current.stopReason === USER_INTERRUPT_REASON) {
return builder.error(USER_INTERRUPT_REASON);
}
if (
result = builder.error(`Command killed by timeout (${timeoutLabel})`, {
brief: `Killed by timeout (${timeoutLabel})`,
});
} else if (current?.status === 'killed' && current.stopReason === USER_INTERRUPT_REASON) {
result = builder.error(USER_INTERRUPT_REASON, { brief: USER_INTERRUPT_REASON });
} else if (
(current?.status === 'failed' || current?.status === 'killed') &&
current.stopReason !== undefined
) {
return builder.error(current.stopReason);
result = builder.error(current.stopReason, { brief: current.stopReason });
} else if (exitCode === 0) {
result = builder.ok('Command executed successfully.');
} else {
if (builder.nChars === 0) builder.write(`Process exited with code ${String(exitCode)}`);
result = builder.error(`Command failed with exit code: ${String(exitCode)}.`, {
brief: `Failed with exit code: ${String(exitCode)}`,
});
}
const isError = exitCode !== 0;
if (isError && builder.nChars === 0) {
builder.write(`Process exited with code ${String(exitCode)}`);
}
if (!isError) {
return builder.ok('Command executed successfully.');
}
return builder.error(`Command failed with exit code: ${String(exitCode)}.`);
return this.addForegroundOutputReference(taskId, result);
}
private detachedTaskResult(
private async addForegroundOutputReference(
taskId: string,
result: ExecutableToolResultBuilderResult,
): Promise<ExecutableToolResult> {
if (!result.truncated) return result;
const output = await this.tasks.getOutputSnapshot(taskId, 0);
if (!output.fullOutputAvailable || output.outputPath === undefined) return result;
const taskOutputHint = this.allowBackground()
? `, or TaskOutput(task_id="${taskId}", block=false)`
: '';
const reference =
`\n\n[Full output saved]\n` +
`task_id: ${taskId}\n` +
`output_path: ${output.outputPath}\n` +
`output_size_bytes: ${String(output.outputSizeBytes)}\n` +
`next_step: Use Read with output_path to page through the full log${taskOutputHint}.`;
return { ...result, output: `${result.output}${reference}` };
}
private backgroundStartedResult(
taskId: string,
proc: IProcess,
description: string,
labels: { title: string },
labels: { title: string; brief: string },
builder = new ToolResultBuilder(),
scenario: 'detached_started' | 'foreground_detached' = 'detached_started',
scenario: 'background_started' | 'foreground_detached' = 'background_started',
): ExecutableToolResult {
const status = this.tasks.getTask(taskId)?.status ?? 'running';
const metadata =
@ -402,23 +435,31 @@ export class BashTool implements BuiltinTool<BashInput> {
`status: ${status}\n` +
`automatic_notification: true\n` +
this.nextStepLines(scenario) +
'human_shell_hint: Tell the human to run /tasks to open the interactive task panel.';
'human_shell_hint: Tell the human to run /tasks to open the interactive background-task panel.';
const foregroundResult = builder.ok('');
const foregroundOutput = foregroundResult.output.length > 0 ? foregroundResult.output : '';
const message = taskResultMessage(labels.title, foregroundResult.message);
return {
const message = backgroundResultMessage(labels.title, foregroundResult.message);
const result: ExecutableToolResult & {
readonly message: string;
readonly brief: string;
readonly truncated: boolean;
} = {
isError: false,
output:
foregroundOutput.length === 0
? metadata
: `${metadata}\n\nforeground_output:\n${foregroundOutput}`,
message,
brief: labels.brief,
truncated: foregroundResult.truncated,
};
return result;
}
private nextStepLines(scenario: 'detached_started' | 'foreground_detached'): string {
private nextStepLines(
scenario: 'background_started' | 'foreground_detached',
): string {
if (scenario === 'foreground_detached') {
// The user explicitly moved a foreground call to the background to avoid
// blocking the current turn. Steer the model away from waiting on it.
@ -431,7 +472,9 @@ export class BashTool implements BuiltinTool<BashInput> {
`when it completes — ${avoid}; continue with your current work.\n`
);
}
// detached_started: the model chose to launch in the background.
// background_started: the model chose to launch in the background. Same anti-wait
// stance — immediately waiting on a background task is just a blocked turn, so do
// not invite a TaskOutput peek here.
if (!this.allowBackground()) {
return 'next_step: You will be automatically notified when it completes.\n';
}
@ -445,7 +488,7 @@ export class BashTool implements BuiltinTool<BashInput> {
registerTool(BashTool);
function taskResultMessage(title: string, suffix: string): string {
function backgroundResultMessage(title: string, suffix: string): string {
const normalized = title.endsWith('.') ? title : `${title}.`;
if (suffix.length === 0) return normalized;
return suffix.endsWith('.') ? `${normalized} ${suffix}` : `${normalized} ${suffix}.`;

View file

@ -145,6 +145,11 @@ function observeProcessStream(
const onData = (chunk: string): void => {
if (chunk.length === 0) return;
sink.appendOutput(chunk);
// Once the manager has begun terminating the task — an output-limit trip
// (see MAX_TASK_OUTPUT_BYTES), a user interrupt, or a timeout —
// `appendOutput` above may synchronously abort the signal. Stop forwarding
// live output from that point so the unbounded forward buffer cannot keep
// growing while the process is being killed.
if (sink.signal.aborted) return;
onOutput?.(kind, chunk);
};

View file

@ -1 +1 @@
Background agent execution is disabled for this agent. Do not set `run_in_background=true`.
Background agent execution is disabled for this agent. Do not set `run_in_background=true` — any call that sets it is rejected before the subagent launches. Run every subagent in the foreground and wait for its result.

View file

@ -1,3 +1,3 @@
Default to a foreground subagent unless the task can run independently and there is a clear benefit to not waiting.
When `run_in_background=true`, the subagent runs detached from this turn. The completion arrives in a later turn as a synthetic user-role message containing its result — you do not need to poll, sleep, or check on its progress. Continue with other work or respond to the user. Never fabricate or predict what the result will say.
Default to a foreground subagent (omit `run_in_background`) when your next step needs its result — foreground hands the result straight back. Reach for `run_in_background=true` only when you have other work to do while it runs and do not need its result to proceed. Never launch in the background and then immediately wait on it (with `TaskOutput block=true`, sleeping, or otherwise): that just blocks the turn for no benefit — run it in the foreground instead.

View file

@ -457,7 +457,7 @@ function formatBackgroundAgentResult(
`description: ${description}`,
'',
allowBackground
? 'next_step: The completion arrives automatically in a later turn; do NOT wait, poll, or call TaskOutput on it. Continue with other work or respond to the user.'
? `next_step: The completion arrives automatically in a later turn — do NOT wait, poll, or call TaskOutput on it; continue with other work or hand back to the user. (If you have nothing to do until it finishes, run such tasks in the foreground next time.)`
: 'next_step: The completion arrives automatically in a later turn.',
`resume_hint: To continue or recover this same subagent later, call Agent(resume="${handle.agentId}", prompt="..."). The parameter is agent_id ("${handle.agentId}"), NOT task_id ("${taskId}") or source_id from a later <notification>. Recovery cases: a later <notification type="task.lost" | "task.failed" | "task.killed"> for this subagent — its conversation history is preserved across session restarts and resume will pick it up.`,
].join('\n');

View file

@ -21,7 +21,7 @@ export interface ISessionTodoService {
setTodos(todos: readonly TodoItem[]): void;
/** Clear the list (equivalent to `setTodos([])`). */
clear(): void;
/** Fires after every `setTodos` with the new list. */
/** Fires when the materialized list changes (after a `todo.set` is applied); carries the sanitized list. */
readonly onDidChange: Event<readonly TodoItem[]>;
}

View file

@ -1,15 +1,23 @@
/**
* `todo` domain (L4) `ISessionTodoService` implementation.
*
* Holds the session's shared in-memory todo list. Every mutation dispatches a
* `todo.set` Op to the main agent's wire (the single source of truth and
* replayable timeline); on resume the main agent's `wire.replay` rebuilds the
* `TodoModel` and the `wire.onRestored` handler copies it back into the
* in-memory list. Binds the `TodoListTool` and the stale-todo reminder into
* every agent (`onDidCreate`), and the restore handler into the main agent
* (`onDidCreateMain`), borrowing each agent's services through its
* `IAgentScopeHandle.accessor`. Per-agent bindings are disposed when the agent
* is disposed. Bound at Session scope.
* Holds the session's shared todo list as a stateless facade over the main
* agent's `TodoModel`: `getTodos` reads `wire.getModel(TodoModel)` live, and
* every mutation only dispatches a `todo.set` Op to the main agent's wire (the
* single source of truth and replayable timeline); `onDidChange` is bridged
* from `wire.subscribe(TodoModel)`. The service keeps no list copy of its own,
* so the live view and the post-replay view can never drift. Binds the
* `TodoListTool` and the stale-todo reminder into every agent (`onDidCreate`),
* and the model subscription into the main agent (`onDidCreateMain`),
* borrowing each agent's services through its `IAgentScopeHandle.accessor`.
* Per-agent bindings are disposed when the agent is disposed. Bound at Session
* scope.
*
* Debt: the session's todo list is still persisted on the MAIN agent's wire (a
* Session Agent edge), so it follows the main agent's lifetime. Once
* `ISessionWireService` is wired up with its own log + replay, move `TodoModel`
* there swap `@IAgentWireService` for `@ISessionWireService` and drop the
* main-agent subscription. The stateless facade makes that a one-line change.
*/
import { Disposable, toDisposable, type IDisposable } from '#/_base/di/lifecycle';
@ -41,11 +49,10 @@ const MAIN_AGENT_ID = 'main';
export class SessionTodoService extends Disposable implements ISessionTodoService {
declare readonly _serviceBrand: undefined;
private todos: readonly TodoItem[] = [];
private readonly onDidChangeEmitter = this._register(new Emitter<readonly TodoItem[]>());
readonly onDidChange = this.onDidChangeEmitter.event;
/** Per-agent bindings (tool + reminder, plus the resume resumer for main). */
/** Per-agent bindings (reminder per agent, plus the model subscription for main). */
private readonly agentBindings = new Map<string, IDisposable[]>();
constructor(
@ -72,13 +79,14 @@ export class SessionTodoService extends Disposable implements ISessionTodoServic
for (const agentId of Array.from(this.agentBindings.keys())) {
this.disposeAgentBindings(agentId);
}
this.todos = [];
}),
);
}
getTodos(): readonly TodoItem[] {
return this.todos;
const main = this.agentLifecycle.getHandle(MAIN_AGENT_ID);
if (main === undefined) return [];
return main.accessor.get(IAgentWireService).getModel(TodoModel);
}
setTodos(todos: readonly TodoItem[]): void {
@ -86,9 +94,7 @@ export class SessionTodoService extends Disposable implements ISessionTodoServic
title: todo.title,
status: todo.status,
}));
this.todos = next;
this.dispatchTodoSet(next);
this.onDidChangeEmitter.fire(next);
}
clear(): void {
@ -105,10 +111,11 @@ export class SessionTodoService extends Disposable implements ISessionTodoServic
private bindMainWire(handle: IAgentScopeHandle): void {
const wire = handle.accessor.get(IAgentWireService);
// Registered on the main agent's wire by `onDidCreateMain`, which fires in
// `ensureMainAgent` strictly before that wire's `replay`, so this handler
// runs at the end of the main agent's restore and copies the rebuilt list.
const disposable = wire.onRestored(() => {
this.todos = wire.getModel(TodoModel);
// `ensureMainAgent` strictly before that wire's `replay`. Bridge model
// changes to `onDidChange`: replay applies silently (no notification), so
// this fires only for live `todo.set` writes, carrying the sanitized model.
const disposable = wire.subscribe(TodoModel, (state) => {
this.onDidChangeEmitter.fire(state);
});
this.trackAgentBinding(handle.id, disposable);
}
@ -127,7 +134,7 @@ export class SessionTodoService extends Disposable implements ISessionTodoServic
return todoListStaleReminder({
active: profile.isToolActive(TODO_LIST_TOOL_NAME, 'builtin'),
history: memory.get(),
todos: this.todos,
todos: this.getTodos(),
});
}

View file

@ -3,10 +3,12 @@
* (`todoSet`) for the session's shared todo list.
*
* Declares the todo list as `readonly TodoItem[]` (initial `[]`) plus the single
* `todo.set` Op whose `apply` is a pure replace the whole list is carried in
* the payload returning the same reference when the payload is already the
* current list so the wire's reference-equality gate stays quiet. The Op type
* (`todo.set`) matches the legacy record type, so `wire.replay` rebuilds the
* `todo.set` Op whose `apply` replaces the whole list with the payload after
* sanitizing it through `readTodoItems`. Replayed / hand-written records may
* carry malformed items, and `apply` is the single logmodel boundary that
* keeps the model clean so every consumer (`getTodos`, the tool render, the
* stale reminder, the compaction summary) can trust it without re-validating.
* The Op type (`todo.set`) matches the legacy record type, so `wire.replay`
* Model from the existing shared append log. Consumed cross-scope by the
* Session-scope `SessionTodoService`: it dispatches `todo.set` to the MAIN
* agent's wire (the single source of truth and replayable timeline) and, on
@ -18,7 +20,7 @@
import { defineModel } from '#/wire/model';
import { defineOp } from '#/wire/op';
import type { TodoItem } from './todoItem';
import { readTodoItems, type TodoItem } from './todoItem';
export type TodoModelState = readonly TodoItem[];
@ -29,5 +31,5 @@ export interface TodoSetPayload {
}
export const todoSet = defineOp(TodoModel, 'todo.set', {
apply: (s, p: TodoSetPayload): TodoModelState => (p.todos === s ? s : p.todos),
apply: (_s, p: TodoSetPayload): TodoModelState => readTodoItems(p.todos),
});

View file

@ -78,6 +78,35 @@ describe('FetchURLTool abort signal', () => {
});
});
describe('FetchURLTool output note', () => {
async function runKind(kind: UrlFetchResult['kind']): Promise<string> {
const fetch = vi
.fn<UrlFetcher['fetch']>()
.mockResolvedValue({ content: 'BODY', kind } satisfies UrlFetchResult);
const tool = new FetchURLTool({ fetch });
const result = await execute(tool, 'https://example.com', new AbortController().signal);
expect(result.isError).toBe(false);
if (typeof result.output !== 'string') throw new Error('expected string output');
return result.output;
}
it('puts the passthrough note and citation reminder at the front of output', async () => {
const output = await runKind('passthrough');
expect(output).toBe(
'The returned content is the full response body, returned verbatim. ' +
'If you use it in your answer, cite this page as a markdown link, e.g. [title](url).\n\nBODY',
);
});
it('puts the extracted note and citation reminder at the front of output', async () => {
const output = await runKind('extracted');
expect(output).toBe(
'The returned content is the main text extracted from the page. ' +
'If you use it in your answer, cite this page as a markdown link, e.g. [title](url).\n\nBODY',
);
});
});
describe('LocalFetchURLProvider abort signal', () => {
it('passes the signal through to fetchImpl', async () => {
const controller = new AbortController();

View file

@ -678,7 +678,7 @@ describe('WebSearchProviderService', () => {
const provider = createService().getWebSearchProvider();
expect(provider).not.toBeUndefined();
const results = await provider!.search('hello', { limit: 2 });
const results = await provider!.search('hello');
expect(results).toEqual([
{ title: 'Title', url: 'https://example.com', snippet: 'Snippet' },
@ -689,6 +689,7 @@ describe('WebSearchProviderService', () => {
const headers = init.headers as Record<string, string>;
expect(headers['Authorization']).toBe('Bearer access-token');
expect(headers['X-Custom']).toBe('yes');
expect(JSON.parse(init.body as string)).toEqual({ text_query: 'hello' });
});
});

View file

@ -16,7 +16,7 @@ import type { Message } from '#/app/llmProtocol/message';
// synthetic error result, stale duplicate results are dropped, and orphan
// results are dropped in a real projection (but kept in a bare slice).
const INTERRUPTED = 'Tool execution was interrupted before its result was recorded';
const INTERRUPTED = 'Tool result is not available in the current context';
function user(text: string): ContextMessage {
return { role: 'user', content: [{ type: 'text', text }], toolCalls: [], origin: { kind: 'user' } };

View file

@ -1,4 +1,4 @@
import { existsSync, mkdtempSync, readFileSync } from 'node:fs';
import { existsSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'pathe';
@ -16,10 +16,11 @@ import { makeHookRunner } from '../externalHooks/runner-stub';
import type { IExternalHooksRunnerService } from '#/app/externalHooksRunner/externalHooksRunner';
import { MASTER_ENV } from '#/app/flag/flagService';
import { microCompactionFlag } from '#/agent/microCompaction/flag';
import { COMPACTION_SUMMARY_PREFIX } from '#/agent/contextMemory/compactionHandoff';
import { estimateTokensForMessages } from '#/_base/utils/tokens';
import { recordingTelemetry, type TelemetryRecord } from '../telemetry/stubs';
import type { TestAgentContext, TestAgentOptions, TestAgentServiceOverride } from '../harness';
import { appServices, createCommandRunner, execEnvServices, testAgent } from '../harness';
import { appServices, createCommandRunner, execEnvServices, sessionServices, testAgent } from '../harness';
import {
IAgentFullCompactionService,
IAgentMicroCompactionService,
@ -273,7 +274,7 @@ describe('FullCompaction', () => {
duration_ms: expect.any(Number),
compacted_count: 6,
retry_count: 0,
thinking_level: 'off',
thinking_effort: 'off',
input_other: 1181,
output: 8,
input_cache_read: 0,
@ -511,14 +512,14 @@ describe('FullCompaction', () => {
const [pre, post] = readHookPayloads(hookLog);
expect(pre).toMatchObject({
hook_event_name: 'PreCompact',
session_id: 'session-hooks',
session_id: 'test-session',
cwd: dir,
trigger: 'auto',
token_count: 39,
});
expect(post).toMatchObject({
hook_event_name: 'PostCompact',
session_id: 'session-hooks',
session_id: 'test-session',
cwd: dir,
trigger: 'auto',
estimated_token_count: ctx.contextData().tokenCount,
@ -636,23 +637,26 @@ describe('FullCompaction', () => {
await completed;
expect(attempts).toBe(3);
// Each empty summary shrinks the compacted prefix before retrying, so the
// recovered summary compacts only the older exchange and leaves the recent
// one in history.
// Empty summaries are retried without shrinking the history; the recovered
// summary replaces the whole history with the real user messages plus the
// prefixed summary.
expect(ctx.compactHistory()).toEqual([
{ role: 'assistant', text: 'Recovered compacted summary.' },
{ role: 'user', text: 'old user one' },
{ role: 'user', text: 'recent user two' },
{ role: 'assistant', text: 'recent assistant two' },
{ role: 'user', text: `${COMPACTION_SUMMARY_PREFIX}\nRecovered compacted summary.` },
]);
expect(
ctx.allEvents.filter((event) => event.event === 'compaction.completed'),
).toEqual([
expect.objectContaining({
args: expect.objectContaining({
result: expect.objectContaining({ summary: 'Recovered compacted summary.' }),
result: expect.objectContaining({
summary: expect.stringContaining('Recovered compacted summary.'),
}),
}),
}),
]);
vi.useRealTimers();
await ctx.expectResumeMatches();
});
@ -692,13 +696,14 @@ describe('FullCompaction', () => {
await completed;
expect(inputs).toHaveLength(2);
// The retry compacts a strictly smaller prefix than the first attempt.
// The retry sends a strictly smaller input than the first attempt.
expect(inputs[1]!.length).toBeLessThan(inputs[0]!.length);
expect(ctx.compactHistory()).toEqual([
{ role: 'assistant', text: 'Recovered compacted summary.' },
{ role: 'user', text: 'old user one' },
{ role: 'user', text: 'recent user two' },
{ role: 'assistant', text: 'recent assistant two' },
{ role: 'user', text: `${COMPACTION_SUMMARY_PREFIX}\nRecovered compacted summary.` },
]);
vi.useRealTimers();
await ctx.expectResumeMatches();
});
@ -771,8 +776,10 @@ describe('FullCompaction', () => {
await vi.advanceTimersByTimeAsync(60_000);
await failed;
// MAX_COMPACTION_RETRY_ATTEMPTS attempts, with prefix reduction between them.
expect(inputs).toHaveLength(5);
// Each empty/think-only response drops the oldest item and resets the retry
// counter; once only one item remains, MAX_COMPACTION_RETRY_ATTEMPTS more
// retries run before failing. 3 drops + 5 retries = 8 generate calls.
expect(inputs).toHaveLength(8);
expect(inputs[1]!.length).toBeLessThan(inputs[0]!.length);
expect(records).toContainEqual({
event: 'compaction_failed',
@ -979,7 +986,11 @@ describe('FullCompaction', () => {
await vi.advanceTimersByTimeAsync(60_000);
await failed;
expect(attempts).toBe(5);
// The four-message compacted prefix shrinks on each truncated response.
// Once only one message remains, it cannot shrink further, so the
// CompactionTruncatedError fails immediately instead of falling through to
// generic retry attempts.
expect(attempts).toBe(4);
expect(ctx.newEvents()).toContainEqual(
expect.objectContaining({
event: 'error',
@ -991,6 +1002,7 @@ describe('FullCompaction', () => {
}),
}),
);
vi.useRealTimers();
await ctx.expectResumeMatches();
});
@ -1053,7 +1065,7 @@ describe('FullCompaction', () => {
await ctx.expectResumeMatches();
});
it('keeps an unresolved tool exchange out of the compaction prompt', async () => {
it('closes an unresolved tool exchange in the compaction prompt with a synthetic result', async () => {
const ctx = testAgent();
ctx.configure({
provider: CATALOGUED_PROVIDER,
@ -1072,37 +1084,34 @@ describe('FullCompaction', () => {
expect(ctx.lastLlmInput()).toMatchInlineSnapshot(`
system: <system-prompt>
tools: Agent, AgentSwarm, CronCreate, CronDelete, CronList, EnterPlanMode, ExitPlanMode
tools: Agent, AgentSwarm, EnterPlanMode, ExitPlanMode
messages:
user: text "old user one"
assistant: text "old assistant one"
user: text "run both tools"
assistant: [] calls call_open_one:LookupOne { "query": "one" }, call_open_two:LookupTwo { "query": "two" }
tool[call_open_one]: text "one result"
tool[call_open_two]: text "Tool result is not available in the current context. Do not assume the tool completed successfully."
user: text <compaction-instruction>
`);
expect(ctx.context.get().map((message) => message.role)).toEqual([
'assistant',
'user',
'assistant',
'tool',
'user',
'user',
]);
await ctx.dispatch({
type: 'context.splice',
start: ctx.context.get().length,
deleteCount: 0,
messages: [
{
role: 'tool',
content: [{ type: 'text', text: 'two result' }],
toolCalls: [],
toolCallId: 'call_open_two',
},
],
type: 'context.append_loop_event',
event: {
type: 'tool.result',
parentUuid: 'call_open_two',
toolCallId: 'call_open_two',
result: { output: 'two result' },
},
});
expect(ctx.context.get().map((message) => message.role)).toEqual([
'assistant',
'user',
'assistant',
'tool',
'tool',
'user',
'user',
]);
await ctx.expectResumeMatches();
});
@ -1136,7 +1145,7 @@ describe('FullCompaction', () => {
);
expect(ctx.lastLlmInput()).toMatchInlineSnapshot(`
system: <system-prompt>
tools: Agent, AgentSwarm, CronCreate, CronDelete, CronList, EnterPlanMode, ExitPlanMode
tools: Agent, AgentSwarm, EnterPlanMode, ExitPlanMode
messages:
user: text "old user one"
assistant: text "old assistant one"
@ -1147,13 +1156,22 @@ describe('FullCompaction', () => {
expect(ctx.compactHistory()).toMatchInlineSnapshot(`
[
{
"role": "assistant",
"text": "Compacted prefix.",
"role": "user",
"text": "old user one",
},
{
"role": "user",
"text": "recent user two",
},
{
"role": "user",
"text": "new user while compacting",
},
{
"role": "user",
"text": "The conversation so far has been compacted to free up context. What follows is your own working summary of this task use it to continue your train of thought rather than starting over. Treat it as notes, not proof: where it says a step was done, tests passed, or a fix worked, verify that yourself before relying on it. Any user messages earlier in this context are preserved verbatim from the compacted conversation; where a system-reminder note among them marks an omitted middle section, the user messages it replaced are covered by this summary.
Compacted prefix.",
},
]
`);
await ctx.expectResumeMatches();
@ -1209,7 +1227,7 @@ describe('FullCompaction', () => {
await ctx.expectResumeMatches();
});
it('auto-compacts very large context in window-sized rounds', async () => {
it('auto-compacts very large context in one full-history round when the summarizer accepts it', async () => {
const maxContextTokens = 4_000;
const ctx = testAgent();
ctx.configure({
@ -1228,9 +1246,7 @@ describe('FullCompaction', () => {
}
const initialTokens = estimateTokensForMessages(ctx.context.get());
const completed = ctx.once('compaction.completed');
for (let i = 1; i <= 30; i++) {
ctx.mockNextResponse({ type: 'text', text: `Auto summary ${String(i)}.` });
}
ctx.mockNextResponse({ type: 'text', text: 'Auto summary.' });
ctx.get(IAgentFullCompactionService).begin({ source: 'auto', instruction: undefined });
await completed;
@ -1242,8 +1258,8 @@ describe('FullCompaction', () => {
expect(initialTokens).toBeGreaterThan(maxContextTokens * 9);
expect(countEvents(events, 'full_compaction.complete')).toBe(1);
expect(countEvents(events, 'compaction.completed')).toBe(1);
expect(compactedPrefixSizes.length).toBeGreaterThan(1);
expect(compactedPrefixSizes.every((size) => size <= maxContextTokens)).toBe(true);
expect(compactedPrefixSizes).toHaveLength(1);
expect(compactedPrefixSizes[0]).toBe(initialTokens);
expect(ctx.contextData().tokenCount).toBeLessThan(maxContextTokens * 0.85);
await ctx.expectResumeMatches();
});
@ -1267,22 +1283,23 @@ describe('FullCompaction', () => {
const events = ctx.newEvents();
expect(events).toEqual(
expect.arrayContaining([
expect.objectContaining({ type: '[wire]', event: 'context.splice' }),
expect.objectContaining({ type: '[wire]', event: 'full_compaction.begin' }),
// Clearing context is a full-history `context.splice` (the v1.5
// equivalent of the legacy `context.clear` record).
expect.objectContaining({
type: '[wire]',
event: 'context.splice',
args: expect.objectContaining({ start: 0, deleteCount: 4, messages: [] }),
}),
expect.objectContaining({ type: '[wire]', event: 'context.clear' }),
expect.objectContaining({ type: '[wire]', event: 'full_compaction.cancel' }),
expect.objectContaining({ type: '[rpc]', event: 'compaction.cancelled' }),
]),
);
expect(eventIndex(events, 'full_compaction.begin')).toBeLessThan(
eventIndex(events, 'context.clear'),
);
expect(eventIndex(events, 'context.clear')).toBeLessThan(
eventIndex(events, 'full_compaction.cancel'),
);
expect(countEvents(events, 'context.apply_compaction')).toBe(0);
expect(countEvents(events, 'full_compaction.complete')).toBe(0);
expect(ctx.lastLlmInput()).toMatchInlineSnapshot(`
system: <system-prompt>
tools: Agent, AgentSwarm, CronCreate, CronDelete, CronList, EnterPlanMode, ExitPlanMode
tools: Agent, AgentSwarm, EnterPlanMode, ExitPlanMode
messages:
user: text "old user one"
assistant: text "old assistant one"
@ -1342,37 +1359,53 @@ describe('FullCompaction', () => {
expect(events).toEqual(
expect.arrayContaining([
expect.objectContaining({ type: '[wire]', event: 'context.splice' }),
expect.objectContaining({ type: '[wire]', event: 'turn.launch' }),
expect.objectContaining({ type: '[wire]', event: 'turn.prompt' }),
expect.objectContaining({ type: '[rpc]', event: 'turn.started' }),
expect.objectContaining({ type: '[wire]', event: 'full_compaction.begin' }),
expect.objectContaining({ type: '[rpc]', event: 'compaction.blocked' }),
expect.objectContaining({ type: '[wire]', event: 'full_compaction.complete' }),
expect.objectContaining({ type: '[rpc]', event: 'turn.step.started' }),
expect.objectContaining({ type: '[rpc]', event: 'turn.ended' }),
]),
);
expect(eventIndex(events, 'turn.prompt')).toBeLessThan(
eventIndex(events, 'full_compaction.begin'),
);
expect(eventIndex(events, 'full_compaction.begin')).toBeLessThan(
eventIndex(events, 'full_compaction.complete'),
);
expect(eventIndex(events, 'compaction.blocked')).toBeLessThan(
eventIndex(events, 'full_compaction.complete'),
);
expect(eventIndex(events, 'full_compaction.complete')).toBeLessThan(
eventIndex(events, 'turn.step.started'),
);
expect(ctx.llmInputs()).toMatchInlineSnapshot(`
call 1:
system: <system-prompt>
tools: Agent, AgentSwarm, CronCreate, CronDelete, CronList, EnterPlanMode, ExitPlanMode
tools: Agent, AgentSwarm, EnterPlanMode, ExitPlanMode
messages:
user: text "old user one"
assistant: text "old assistant one"
user: text "old user two"
assistant: text "old assistant two"
user: text "recent user three"
assistant: text "recent assistant three"
user: text "Answer after compacting"
user: text <compaction-instruction>
call 2:
messages:
assistant: text "Auto compacted summary."
user: text "recent user three"
assistant: text "recent assistant three"
user: text "Answer after compacting"
user: text "old user one\\n\\nold user two\\n\\nrecent user three\\n\\nAnswer after compacting"
user: text "The conversation so far has been compacted to free up context. What follows is your own working summary of this task — use it to continue your train of thought rather than starting over. Treat it as notes, not proof: where it says a step was done, tests passed, or a fix worked, verify that yourself before relying on it. Any user messages earlier in this context are preserved verbatim from the compacted conversation; where a system-reminder note among them marks an omitted middle section, the user messages it replaced are covered by this summary.\\nAuto compacted summary."
`);
expect(records).toContainEqual({
event: 'compaction_finished',
properties: expect.objectContaining({
source: 'auto',
tokens_before: 46,
tokens_after: 28,
compacted_count: 4,
tokens_after: 166,
compacted_count: 7,
retry_count: 0,
}),
});
@ -1419,67 +1452,39 @@ describe('FullCompaction', () => {
await ctx.rpc.beginCompaction({});
await compacted;
// Compaction preserves the in-flight tool exchange (and the reminder behind
// it) in recent; the projection closes the open calls and keeps the
// reminder after them.
// Compaction drops the in-flight tool exchange and the deferred reminder;
// only real user messages and the compaction summary remain.
expect(ctx.context.get().map((m) => m.role)).toEqual([
'assistant',
'user',
'assistant',
'user',
]);
expect(ctx.project().map((m) => m.role)).toEqual([
'assistant',
'user',
'assistant',
'tool',
'tool',
'user',
'user',
]);
expect(ctx.context.get().at(-1)?.origin).toEqual({ kind: 'compaction_summary' });
// Closing the exchange (both results together) lets the projector place the
// reminder after the tool results.
// The dropped tool calls no longer exist, so late tool results are orphans
// and do not change history.
await ctx.dispatch({
type: 'context.splice',
start: ctx.context.get().length,
deleteCount: 0,
messages: [
{
role: 'tool',
content: [{ type: 'text', text: 'one result' }],
toolCalls: [],
toolCallId: 'call_unresolved_one',
},
{
role: 'tool',
content: [{ type: 'text', text: 'two result' }],
toolCalls: [],
toolCallId: 'call_unresolved_two',
},
],
type: 'context.append_loop_event',
event: {
type: 'tool.result',
parentUuid: 'call_unresolved_one',
toolCallId: 'call_unresolved_one',
result: { output: 'one result' },
},
});
await ctx.dispatch({
type: 'context.append_loop_event',
event: {
type: 'tool.result',
parentUuid: 'call_unresolved_two',
toolCallId: 'call_unresolved_two',
result: { output: 'two result' },
},
});
// Raw history keeps insertion order (reminder before the trailing results).
expect(ctx.context.get().map((m) => m.role)).toEqual([
'assistant',
'user',
'assistant',
'user',
'tool',
'tool',
]);
// Projection moves the reminder to after the now-closed tool exchange.
const projected = ctx.project();
expect(projected.map((m) => m.role)).toEqual([
'assistant',
'user',
'assistant',
'tool',
'tool',
'user',
]);
expect(projected.at(-1)?.content).toEqual([
{ type: 'text', text: '<system-reminder>\nhost note\n</system-reminder>' },
]);
});
@ -1523,100 +1528,70 @@ describe('FullCompaction', () => {
await ctx.rpc.beginCompaction({});
await compacted;
// Compaction drops the partially-resolved tool exchange and the deferred
// reminder; only real user messages and the compaction summary remain.
expect(ctx.context.get().map((m) => m.role)).toEqual([
'assistant',
'user',
'assistant',
'tool',
'user',
]);
expect(ctx.project().map((m) => m.role)).toEqual([
'assistant',
'user',
'assistant',
'tool',
'tool',
'user',
'user',
]);
expect(ctx.context.get().at(-1)?.origin).toEqual({ kind: 'compaction_summary' });
// The dropped tool calls no longer exist, so a late tool result is an
// orphan and does not change history.
await ctx.dispatch({
type: 'context.splice',
start: ctx.context.get().length,
deleteCount: 0,
messages: [
{
role: 'tool',
content: [{ type: 'text', text: 'two result' }],
toolCalls: [],
toolCallId: 'call_unresolved_two',
},
],
});
// Raw history keeps insertion order; the projector moves the reminder to
// after the now-closed tool exchange.
expect(ctx.context.get().map((m) => m.role)).toEqual([
'assistant',
'user',
'assistant',
'tool',
'user',
'tool',
]);
const projected = ctx.project();
expect(projected.map((m) => m.role)).toEqual([
'assistant',
'user',
'assistant',
'tool',
'tool',
'user',
]);
expect(projected.at(-1)?.content).toEqual([
{ type: 'text', text: '<system-reminder>\nhost note\n</system-reminder>' },
]);
});
it('fails the turn with compaction.unable when auto compaction has no compactable prefix', async () => {
const ctx = testAgent();
ctx.configure({
provider: CATALOGUED_PROVIDER,
modelCapabilities: {
...CATALOGUED_MODEL_CAPABILITIES,
max_context_tokens: 2_000,
type: 'context.append_loop_event',
event: {
type: 'tool.result',
parentUuid: 'call_unresolved_two',
toolCallId: 'call_unresolved_two',
result: { output: 'two result' },
},
});
const oversizedPrompt = `initial-pending-verbatim:${'x'.repeat(8_000)}`;
await ctx.rpc.prompt({ input: [{ type: 'text', text: oversizedPrompt }] });
const events = await ctx.untilTurnEnd();
expect(eventIndex(events, 'compaction.started')).toBe(-1);
expect(ctx.llmCalls).toHaveLength(0);
expect(events).toContainEqual(
expect.objectContaining({
event: 'turn.ended',
args: expect.objectContaining({
reason: 'failed',
error: expect.objectContaining({ code: 'compaction.unable' }),
}),
}),
);
await ctx.expectResumeMatches();
expect(ctx.context.get().map((m) => m.role)).toEqual([
'user',
'user',
'user',
]);
});
it('rejects manual compaction with compaction.unable when no prefix is compactable', async () => {
it('compacts a single user message and keeps it ahead of the summary', async () => {
const ctx = testAgent();
ctx.configure({
provider: CATALOGUED_PROVIDER,
modelCapabilities: CATALOGUED_MODEL_CAPABILITIES,
});
ctx.appendUserMessage([{ type: 'text', text: 'only pending user' }]);
const compacted = ctx.once('full_compaction.complete');
const completed = ctx.once('compaction.completed');
await expect(ctx.rpc.beginCompaction({})).rejects.toMatchObject({
code: 'compaction.unable',
ctx.mockNextResponse({ type: 'text', text: 'Single message summary.' });
await ctx.rpc.beginCompaction({});
await compacted;
await completed;
expect(ctx.llmCalls).toHaveLength(1);
expect(ctx.compactHistory()).toEqual([
{ role: 'user', text: 'only pending user' },
{
role: 'user',
text: `${COMPACTION_SUMMARY_PREFIX}\nSingle message summary.`,
},
]);
await ctx.expectResumeMatches();
});
it('manual compaction can run after a previous single-message compaction', async () => {
const ctx = testAgent();
ctx.configure({
provider: CATALOGUED_PROVIDER,
modelCapabilities: CATALOGUED_MODEL_CAPABILITIES,
});
expect(ctx.llmCalls).toHaveLength(0);
ctx.appendUserMessage([{ type: 'text', text: 'only pending user' }]);
ctx.mockNextResponse({ type: 'text', text: 'Single message summary.' });
await ctx.rpc.beginCompaction({});
await ctx.once('compaction.completed');
ctx.clearContext();
ctx.appendExchange(1, 'old user one', 'old assistant one', 20);
@ -1624,23 +1599,37 @@ describe('FullCompaction', () => {
const compacted = ctx.once('full_compaction.complete');
const completed = ctx.once('compaction.completed');
ctx.mockNextResponse({ type: 'text', text: 'Compacted after no-op cancel.' });
ctx.mockNextResponse({ type: 'text', text: 'Compacted after single-message compact.' });
await ctx.rpc.beginCompaction({});
await compacted;
await completed;
expect(ctx.llmCalls).toHaveLength(1);
expect(ctx.llmCalls).toHaveLength(2);
expect(ctx.compactHistory()).toEqual([
{ role: 'user', text: 'old user one' },
{ role: 'user', text: 'recent user two' },
{
role: 'user',
text: expect.stringContaining('Compacted after no-op cancel.'),
text: expect.stringContaining('Compacted after single-message compact.'),
},
]);
await ctx.expectResumeMatches();
});
it('rejects manual compaction with compaction.unable when history is empty', async () => {
const ctx = testAgent();
ctx.configure({
provider: CATALOGUED_PROVIDER,
modelCapabilities: CATALOGUED_MODEL_CAPABILITIES,
});
await expect(ctx.rpc.beginCompaction({})).rejects.toMatchObject({
code: 'compaction.unable',
});
expect(ctx.llmCalls).toHaveLength(0);
await ctx.expectResumeMatches();
});
it('does not auto compact small contexts when reserved size exceeds the model window', async () => {
const ctx = testAgent({
initialConfig: {
@ -1698,7 +1687,7 @@ describe('FullCompaction', () => {
await ctx.expectResumeMatches();
});
it('keeps an oversized pending user prompt out of auto compaction', async () => {
it('includes an oversized pending user prompt in auto compaction', async () => {
const ctx = testAgent();
ctx.configure({
provider: CATALOGUED_PROVIDER,
@ -1718,8 +1707,15 @@ describe('FullCompaction', () => {
expect(ctx.llmCalls).toHaveLength(2);
const [compactionCall, answerCall] = ctx.llmCalls;
const compactionTexts = compactionCall?.history.map(messageText) ?? [];
expect(compactionTexts.some((text) => text.includes('keep-this-pending-verbatim'))).toBe(false);
expect(compactionCall?.history.map((message) => message.role)).toEqual(['user', 'assistant', 'user']);
// The whole history is compacted, so the pending prompt is included in the
// compaction input and kept verbatim in the post-compaction replacement.
expect(compactionTexts.some((text) => text.includes('keep-this-pending-verbatim'))).toBe(true);
expect(compactionCall?.history.map((message) => message.role)).toEqual([
'user',
'assistant',
'user',
'user',
]);
expect(
answerCall?.history.map(messageText).some((text) => text.includes('Oversized prompt summary.')),
).toBe(true);
@ -1749,8 +1745,15 @@ describe('FullCompaction', () => {
expect(ctx.llmCalls).toHaveLength(2);
const [compactionCall, answerCall] = ctx.llmCalls;
const compactionTexts = compactionCall?.history.map(messageText) ?? [];
expect(compactionTexts.some((text) => text.includes('ratio-pending-verbatim'))).toBe(false);
expect(compactionCall?.history.map((message) => message.role)).toEqual(['user', 'assistant', 'user']);
// The whole history is compacted, so the pending prompt is included in the
// compaction input and kept verbatim in the post-compaction replacement.
expect(compactionTexts.some((text) => text.includes('ratio-pending-verbatim'))).toBe(true);
expect(compactionCall?.history.map((message) => message.role)).toEqual([
'user',
'assistant',
'user',
'user',
]);
expect(
answerCall?.history.map(messageText).some((text) => text.includes('Ratio compacted summary.')),
).toBe(true);
@ -1806,7 +1809,7 @@ describe('FullCompaction', () => {
args: expect.objectContaining({
result: expect.objectContaining({
summary: 'Overflow compacted summary.',
compactedCount: 2,
compactedCount: 4,
}),
}),
}),
@ -1827,6 +1830,7 @@ describe('FullCompaction', () => {
[
"user: old user one",
"assistant: old assistant one",
"user: Retry after provider overflow",
"user: <compaction-instruction>",
],
[
@ -2012,12 +2016,16 @@ describe('FullCompaction', () => {
await ctx.untilTurnEnd();
expect(callCount).toBe(3);
expect(providerThinkingEfforts).toEqual(['high', 'high', 'high']);
// The catalogued model declares no supportEfforts, so the Kimi provider
// normalizes to boolean thinking and reports 'on' rather than the
// requested 'high'. The stored thinkingLevel still carries 'high' across
// compaction, which is asserted through telemetry below.
expect(providerThinkingEfforts).toEqual(['on', 'on', 'on']);
expect(records).toContainEqual({
event: 'compaction_finished',
properties: expect.objectContaining({
source: 'auto',
thinking_level: 'high',
thinking_effort: 'high',
}),
});
});
@ -2051,10 +2059,11 @@ describe('FullCompaction', () => {
const modelResolver = ctx.modelResolver;
if (modelResolver === undefined) throw new Error('Expected model provider');
const resolve = modelResolver.resolve.bind(modelResolver);
modelResolver.resolve = (model: string) => ({
...resolve(model),
modelCapabilities: UNKNOWN_CAPABILITY,
});
modelResolver.resolve = (model: string) => {
const resolved = resolve(model);
Object.defineProperty(resolved, 'capabilities', { value: UNKNOWN_CAPABILITY });
return resolved;
};
expect(ctx.get(IAgentProfileService).data().modelCapabilities.max_context_tokens).toBe(0);
ctx.appendExchange(1, 'old user one', 'old assistant one', 20);
ctx.newEvents();
@ -2076,7 +2085,7 @@ describe('FullCompaction', () => {
args: expect.objectContaining({
result: expect.objectContaining({
summary: 'Unknown window compacted summary.',
compactedCount: 2,
compactedCount: 4,
}),
}),
}),
@ -2236,8 +2245,10 @@ describe('FullCompaction', () => {
it('ignores filtered assistant placeholders when checking the retained overflow suffix', async () => {
let callCount = 0;
const generate: GenerateFn = async (_provider, _system, _tools, _history, callbacks) => {
const inputs: string[][] = [];
const generate: GenerateFn = async (_provider, _system, _tools, history, callbacks) => {
callCount += 1;
inputs.push(inputHistorySnapshot(history));
if (callCount === 1) {
throw new APIContextOverflowError(
400,
@ -2287,7 +2298,8 @@ describe('FullCompaction', () => {
args: expect.objectContaining({
result: expect.objectContaining({
summary: 'Placeholder compacted summary.',
compactedCount: 2,
compactedCount: 3,
droppedCount: 2,
}),
}),
}),
@ -2298,11 +2310,42 @@ describe('FullCompaction', () => {
args: { turnId: 0, reason: 'completed' },
}),
);
expect(inputs).toMatchInlineSnapshot(`
[
[
"user: old user one",
"assistant: old assistant one",
"user: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"user: <compaction-instruction>",
],
[
"user: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"user: <compaction-instruction>",
],
[
"user: old user one
xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"user: The conversation so far has been compacted to free up context. What follows is your own working summary of this task use it to continue your train of thought rather than starting over. Treat it as notes, not proof: where it says a step was done, tests passed, or a fix worked, verify that yourself before relying on it. Any user messages earlier in this context are preserved verbatim from the compacted conversation; where a system-reminder note among them marks an omitted middle section, the user messages it replaced are covered by this summary.
Placeholder compacted summary.",
],
]
`);
});
it('appends the todo list to the compaction summary', async () => {
const ctx = testAgent();
const todos = [
{ title: 'Fix the auth bug', status: 'in_progress' },
{ title: 'Add tests', status: 'pending' },
] as const;
const ctx = testAgent(
sessionServices((reg) => {
reg.definePartialInstance(ISessionTodoService, {
getTodos: () => todos,
});
}),
);
ctx.configure({
provider: CATALOGUED_PROVIDER,
modelCapabilities: CATALOGUED_MODEL_CAPABILITIES,
@ -2310,11 +2353,6 @@ describe('FullCompaction', () => {
ctx.appendExchange(1, 'old user one', 'old assistant one', 20);
ctx.appendExchange(2, 'recent user two', 'recent assistant two', 80);
ctx.get(ISessionTodoService).setTodos([
{ title: 'Fix the auth bug', status: 'in_progress' },
{ title: 'Add tests', status: 'pending' },
]);
const compacted = new Promise<void>((resolve) => {
ctx.emitter.once('full_compaction.complete', () => {
resolve();
@ -2545,6 +2583,10 @@ function messageText(message: Message | undefined): string {
}
function hookPayloadLoggerCommand(logPath: string): string {
// Write the hook script to a file and run it with node, instead of
// `node -e <json>`; cmd.exe on Windows mangles the escaped quotes in the
// inline form and corrupts the script before it can run.
const scriptPath = `${logPath}.cjs`;
const script = [
"const fs = require('node:fs');",
"let input = '';",
@ -2553,7 +2595,8 @@ function hookPayloadLoggerCommand(logPath: string): string {
` fs.appendFileSync(${JSON.stringify(logPath)}, JSON.stringify(JSON.parse(input)) + '\\n');`,
'});',
].join('');
return `node -e ${JSON.stringify(script)}`;
writeFileSync(scriptPath, script);
return `${process.execPath} ${scriptPath}`;
}
function readHookPayloads(logPath: string): Array<Record<string, unknown>> {

View file

@ -580,7 +580,11 @@ function resolveExternalHooksRunner(
function isRunnerLike(
value: Pick<IExternalHooksRunnerService, 'trigger' | 'triggerBlock' | 'fireAndForgetTrigger'>,
): value is IExternalHooksRunnerService {
return '_serviceBrand' in value;
return (
typeof value.trigger === 'function' &&
typeof value.triggerBlock === 'function' &&
typeof value.fireAndForgetTrigger === 'function'
);
}
const noopHookRunner: IExternalHooksRunnerService = {

View file

@ -287,7 +287,10 @@ describe('AgentProfileService (wire-backed config.update)', () => {
await expect(
host.wire.replay({ type: 'config.update', thinkingEffort: 'low', thinkingLevel: 'high' }),
).rejects.toThrow('conflicting thinkingEffort');
).rejects.toMatchObject({
code: 'profile.thinking_alias_conflict',
name: 'ProfileError',
});
});
it('applies thinking.keep model override when thinking is enabled', () => {

View file

@ -10,9 +10,6 @@
* semantics match production.
*
* Deviations from v1:
* - The `brief` result field does not exist on v2's `ExecutableToolResult`,
* so v1 assertions on `result.brief` are dropped; the `output` / `message`
* assertions are kept.
* - v1's `execWithEnv(args, env)` is now `runner.exec(args, { env })`, so
* spawn-call assertions read `options.env` from the second argument.
*/
@ -364,8 +361,10 @@ function errorMessage(error: unknown): string {
function createFakeTaskService(options: { maxRunningTasks?: number } = {}): {
readonly service: IAgentTaskService;
readonly tasks: Map<string, ManagedEntry>;
readonly persisted: Set<string>;
} {
const tasks = new Map<string, ManagedEntry>();
const persisted = new Set<string>();
let counter = 0;
const nextId = (prefix: string): string => {
@ -541,18 +540,20 @@ function createFakeTaskService(options: { maxRunningTasks?: number } = {}): {
return result;
},
persistOutput(): void {
/* no-op in the fake */
persistOutput(taskId: string): void {
persisted.add(taskId);
},
async getOutputSnapshot(taskId: string): Promise<AgentTaskOutputSnapshot> {
const entry = tasks.get(taskId);
const preview = entry === undefined ? '' : entry.outputChunks.join('');
const fullOutputAvailable = persisted.has(taskId);
return {
outputPath: fullOutputAvailable ? `/fake/tasks/${taskId}/output.log` : undefined,
outputSizeBytes: preview.length,
previewBytes: preview.length,
truncated: false,
fullOutputAvailable: false,
fullOutputAvailable,
preview,
};
},
@ -655,7 +656,7 @@ function createFakeTaskService(options: { maxRunningTasks?: number } = {}): {
},
};
return { service, tasks };
return { service, tasks, persisted };
}
// ── Test execution helper ────────────────────────────────────────────
@ -860,6 +861,7 @@ describe('BashTool', () => {
expect(result).toMatchObject({
isError: true,
message: 'Command failed with exit code: 2.',
brief: 'Failed with exit code: 2',
});
expect(result.output).toContain('boom\n');
expect(result.output).toContain('Command failed with exit code: 2.');
@ -889,6 +891,7 @@ describe('BashTool', () => {
expect(result).toMatchObject({
isError: true,
message: 'Command failed with exit code: 2.',
brief: 'Failed with exit code: 2',
});
expect(result.output).toContain('partial\nboom\n');
expect(result.output).toContain('Command failed with exit code: 2.');
@ -911,6 +914,7 @@ describe('BashTool', () => {
expect(result).toMatchObject({
isError: true,
message: 'wait failed',
brief: 'wait failed',
});
expect(result.output).toContain('partial output\nwait failed');
expect(result.output).not.toContain('exit code: null');
@ -992,7 +996,7 @@ describe('BashTool', () => {
await vi.advanceTimersByTimeAsync(250);
const result = await running;
expect(result).toMatchObject({ isError: true });
expect(result).toMatchObject({ isError: true, brief: 'Killed by timeout (1s)' });
expect(result.output).toContain('Command killed by timeout (1s)');
} finally {
vi.useRealTimers();
@ -1012,7 +1016,7 @@ describe('BashTool', () => {
const result = await running;
expect(proc.kill).toHaveBeenCalledWith('SIGTERM');
expect(result).toMatchObject({ isError: true });
expect(result).toMatchObject({ isError: true, brief: 'Killed by timeout (1s)' });
expect(result.output).toContain('Command killed by timeout (1s)');
expect(result.output).not.toContain('Premature close');
} finally {
@ -1111,6 +1115,40 @@ describe('BashTool', () => {
expect(output).toContain('Output is truncated');
});
it('saves full foreground output when the inline result is truncated', async () => {
const fullOutput = `${'short line\n'.repeat(6_000)}tail survives\n`;
const { runner } = createTestRunner(processWithOutput({ stdout: fullOutput }));
const { service, persisted } = createFakeTaskService();
const tool = bashTool(runner, createTestEnv(), createTestCtx(), service);
const result = await executeTool(tool, context({ command: 'flood', timeout: 60 }));
const output = result.output as string;
const taskId = /^task_id: (bash-[0-9a-z]{8})$/m.exec(output)?.[1];
expect(output).toContain('[...truncated]');
expect(output).toContain('[Full output saved]');
expect(taskId).toBeTruthy();
// The inline truncation must have started early persistence of the full log.
expect(persisted.has(taskId!)).toBe(true);
expect(output).toContain(`output_path: /fake/tasks/${taskId}/output.log`);
expect(output).toContain('Use Read with output_path');
expect(output).toContain(`TaskOutput(task_id="${taskId}", block=false)`);
});
it('omits the TaskOutput hint from the saved-output reference when background tools are disabled', async () => {
const fullOutput = 'short line\n'.repeat(6_000);
const { runner } = createTestRunner(processWithOutput({ stdout: fullOutput }));
const { service } = createFakeTaskService();
const tool = bashTool(runner, createTestEnv(), createTestCtx(), service, stubProfile(() => false));
const result = await executeTool(tool, context({ command: 'flood', timeout: 60 }));
const output = result.output as string;
expect(output).toContain('[Full output saved]');
expect(output).toContain('Use Read with output_path');
expect(output).not.toContain('TaskOutput');
});
it('rejects empty-string commands at the schema layer', () => {
expect(BashInputSchema.safeParse({ command: '' }).success).toBe(false);
});
@ -1163,6 +1201,33 @@ describe('BashTool', () => {
expect(description).toContain('**Guidelines for efficiency:**');
expect(description).toContain('run_in_background=true');
expect(description).toContain('automatically notified');
// Moved here from system.md: the "don't block on a background task" nudge belongs in
// the background-enabled Bash description, the only place that documents it.
expect(description).toContain('returning control to the user');
});
it('disables background execution when TaskList is inactive even if TaskOutput/TaskStop are active', async () => {
const { runner, exec } = createTestRunner(processWithOutput());
const tool = bashTool(
runner,
createTestEnv(),
createTestCtx(),
createFakeTaskService().service,
stubProfile((name) => name !== 'TaskList'),
);
// Background management needs TaskList, TaskOutput, and TaskStop; without
// TaskList the description must fall back to the disabled variant.
expect(tool.description).toContain('Background execution is disabled for this agent');
const result = await executeTool(
tool,
context({ command: 'sleep 10', run_in_background: true, description: 'watch' }),
);
expect(result).toMatchObject({ isError: true });
expect(result.output).toContain('Background execution is not available');
expect(exec).not.toHaveBeenCalled();
});
});
@ -1199,6 +1264,8 @@ describe('BashTool background mode', () => {
expect(result.output).toContain(`task_id: ${task.taskId}`);
expect(result.output).toContain('automatic_notification: true');
expect(result.output).toContain('do NOT wait, poll, or call TaskOutput');
expect(result).toMatchObject({ message: 'Task moved to background.' });
expect((result as { brief?: string }).brief).toBe(`Backgrounded ${task.taskId}`);
expect(service.getTask(task.taskId)).toMatchObject({ detached: true });
await vi.waitFor(async () => {
await expect(service.readOutput(task.taskId)).resolves.toContain('after detach\n');
@ -1367,6 +1434,11 @@ describe('BashTool background mode', () => {
expect(result.output).toMatch(/task_id: bash-[0-9a-z]{8}/);
expect(result.output).toContain('automatic_notification: true');
expect(result).toMatchObject({ message: 'Background task started.' });
expect((result as { brief?: string }).brief).toMatch(/^Started bash-[0-9a-z]{8}$/);
// The launch message must steer away from waiting, not invite a TaskOutput peek.
expect(result.output).toContain('do NOT wait, poll, or call TaskOutput on it');
expect(result.output).not.toContain('block=false');
expect(service.list(false)).toHaveLength(1);
});

View file

@ -523,7 +523,7 @@ describe('AgentTaskService', () => {
expect(() => {
manager.registerTask(agentTask(new Promise(() => {}), 'second background'));
}).toThrow('Too many detached tasks are already running.');
}).toThrow('Too many background tasks are already running.');
});
it('does not count foreground tasks detached later against the detached task limit', () => {
@ -539,7 +539,7 @@ describe('AgentTaskService', () => {
expect(() => {
manager.registerTask(agentTask(new Promise(() => {}), 'second background'));
}).toThrow('Too many detached tasks are already running.');
}).toThrow('Too many background tasks are already running.');
});
it('lists active tasks by default', () => {
@ -603,10 +603,10 @@ describe('AgentTaskService', () => {
expect(() => {
registerProcess(manager, pendingProcess().proc, 'sleep 60', 'second task');
}).toThrow('Too many detached tasks are already running.');
}).toThrow('Too many background tasks are already running.');
expect(() => {
manager.registerTask(agentTask(new Promise(() => {}), 'agent task'));
}).toThrow('Too many detached tasks are already running.');
}).toThrow('Too many background tasks are already running.');
});
it('captures process output', async () => {
@ -1159,6 +1159,31 @@ describe('AgentTaskService', () => {
expect(await manager.wait(runningId, 0)).toMatchObject({ status: 'running' });
});
it('wait with a zero timeout returns the immediate snapshot before next-tick completion', async () => {
const { manager } = createAgentTaskService();
const proc = manuallyResolvedProcess();
const taskId = registerProcess(
manager,
proc.proc,
'sleep 0',
'next-tick completion',
);
await Promise.resolve();
setTimeout(() => {
proc.resolve(0);
}, 0);
expect(await manager.wait(taskId, 0)).toMatchObject({
status: 'running',
exitCode: null,
});
await expect(manager.wait(taskId)).resolves.toMatchObject({
status: 'completed',
exitCode: 0,
});
});
it('clears task deadline timers when completion wins the race', async () => {
vi.useFakeTimers();
const { manager } = createAgentTaskService();

View file

@ -6,9 +6,12 @@ import { join } from 'pathe';
import type { IProcess } from '#/session/process/processRunner';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { IAgentTaskService } from '#/agent/task/task';
import { TERMINAL_STATUSES } from '#/agent/task/types';
import { TaskOutputTool } from '#/agent/task/tools/task-output';
import { ProcessTask } from '#/os/backends/node-local/tools/process-task';
import { createAgentTaskPersistence, type TaskServiceTestManager } from './stubs';
import { taskServices, createTestAgent, homeDirServices, type TestAgentContext } from '../harness';
import { executeTool, type TestExecutableToolContext } from '../tools/fixtures/execute-tool';
interface TaskServiceFixture {
readonly ctx: TestAgentContext;
@ -36,6 +39,22 @@ function registerProcess(
return manager.registerTask(new ProcessTask(proc, command, description));
}
function toolContext<Input>(
toolCallId: string,
args: Input,
): TestExecutableToolContext<Input> {
return {
turnId: 0,
toolCallId,
args,
signal: new AbortController().signal,
};
}
function outputString(result: { readonly output: string | readonly unknown[] }): string {
return typeof result.output === 'string' ? result.output : JSON.stringify(result.output);
}
async function waitForOutput(
manager: IAgentTaskService,
taskId: string,
@ -49,6 +68,31 @@ async function waitForOutput(
throw new Error(`Timed out waiting for output: ${expected}`);
}
async function waitForTaskNotifications(
ctx: TestAgentContext,
manager: TaskServiceTestManager,
): Promise<void> {
const tasks = manager.list(false).filter(
(task) =>
TERMINAL_STATUSES.has(task.status) &&
task.detached !== false &&
task.terminalNotificationSuppressed !== true,
);
if (tasks.length === 0) return;
await vi.waitFor(() => {
const origins = ctx.context.get().map((message) => message.origin);
for (const task of tasks) {
expect(origins).toContainEqual({
kind: 'task',
taskId: task.taskId,
status: task.status,
notificationId: `task:${task.taskId}:${task.status}`,
});
}
});
}
function immediateProcess(exitCode: number, stdoutText = ''): IProcess {
return {
stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable,
@ -78,6 +122,7 @@ describe('AgentTaskService — readOutput / getOutputSnapshot', () => {
afterEach(async () => {
try {
await waitForTaskNotifications(ctx, manager);
await ctx.expectResumeMatches();
} finally {
await ctx.dispose();
@ -184,6 +229,38 @@ describe('AgentTaskService — readOutput / getOutputSnapshot', () => {
}
});
it('TaskOutputTool reads persisted output for a ghost task loaded after restart', async () => {
const taskId = registerProcess(
manager,
immediateProcess(0, 'persisted output\n'),
'echo persisted output',
'persist output test',
);
await waitForOutput(manager, taskId, 'persisted output');
await manager.wait(taskId);
const freshFixture = createTaskService(sessionDir);
const fresh = freshFixture.manager;
try {
await fresh.loadFromDisk();
await fresh.reconcile();
const result = await executeTool(
new TaskOutputTool(fresh),
toolContext('task_output_restored', { task_id: taskId }),
);
const output = outputString(result);
expect(result.isError ?? false).toBe(false);
expect(output).toContain('status: completed');
expect(output).toContain('output_path:');
expect(output).toContain('persisted output');
await freshFixture.ctx.expectResumeMatches();
} finally {
await freshFixture.ctx.dispose();
}
});
it('readOutput respects tail length', async () => {
const taskId = registerProcess(
manager,

View file

@ -15,6 +15,7 @@ import {
type AgentTaskInfo,
IAgentTaskService,
} from '#/agent/task/task';
import { TaskStopTool } from '#/agent/task/tools/task-stop';
import {
SubagentTask,
type SubagentHandle,
@ -35,6 +36,7 @@ import {
type TestAgentServiceOverride,
} from '../harness';
import { recordingTelemetry } from '../telemetry/stubs';
import { executeTool, type TestExecutableToolContext } from '../tools/fixtures/execute-tool';
import {
createAgentTaskPersistence,
type TaskServiceTestManager,
@ -260,6 +262,22 @@ function firstAppendedContextMessage(agent: FakeTaskAgent): TestContextMessage {
return message;
}
function toolContext<Input>(
toolCallId: string,
args: Input,
): TestExecutableToolContext<Input> {
return {
turnId: 0,
toolCallId,
args,
signal: new AbortController().signal,
};
}
function outputString(result: { readonly output: string | readonly unknown[] }): string {
return typeof result.output === 'string' ? result.output : JSON.stringify(result.output);
}
function registerProcess(
manager: IAgentTaskService,
proc: IProcess,
@ -286,7 +304,7 @@ describe('AgentTaskService — event emission', () => {
status: 'running',
}),
});
expect(agent.telemetry.track).toHaveBeenCalledWith('task_created', {
expect(agent.telemetry.track).toHaveBeenCalledWith('background_task_created', {
kind: 'bash',
});
});
@ -305,7 +323,7 @@ describe('AgentTaskService — event emission', () => {
status: 'running',
}),
});
expect(agent.telemetry.track).toHaveBeenCalledWith('task_created', {
expect(agent.telemetry.track).toHaveBeenCalledWith('background_task_created', {
kind: 'agent',
});
});
@ -325,10 +343,10 @@ describe('AgentTaskService — event emission', () => {
}),
});
expect(agent.telemetry.track).toHaveBeenCalledWith(
'task_completed',
'background_task_completed',
expect.objectContaining({
kind: 'process',
duration: expect.any(Number),
duration_ms: expect.any(Number),
status: 'completed',
}),
);
@ -349,11 +367,11 @@ describe('AgentTaskService — event emission', () => {
await timedOut;
expect(agent.telemetry.track).toHaveBeenCalledWith(
'task_completed',
'background_task_completed',
expect.objectContaining({ kind: 'process', status: 'failed' }),
);
expect(agent.telemetry.track).toHaveBeenCalledWith(
'task_completed',
'background_task_completed',
expect.objectContaining({ kind: 'agent', status: 'timed_out' }),
);
});
@ -436,7 +454,7 @@ describe('AgentTaskService — notification delivery', () => {
});
const content = (message as { content: Array<{ text: string }> }).content;
const text = content[0]!.text;
expect(text).toContain('Task agent completed');
expect(text).toContain('Background agent completed');
expect(text).toContain('agent task completed.');
expect(text).toContain('<output-file');
expect(text).not.toContain('final subagent summary');
@ -461,7 +479,7 @@ describe('AgentTaskService — notification delivery', () => {
});
const content = (message as { content: Array<{ text: string }> }).content;
const text = content[0]!.text;
expect(text).toContain('Task process completed');
expect(text).toContain('Background process completed');
expect(text).toContain('shell task completed.');
});
@ -484,10 +502,70 @@ describe('AgentTaskService — notification delivery', () => {
});
const content = (message as { content: Array<{ text: string }> }).content;
expect(content[0]!.text).toContain(
'Task process killed',
'Background process killed',
);
});
it('TaskStopTool suppresses the real terminal notification for model-requested stops', async () => {
const { agent, manager } = createAgentTaskService();
const taskId = registerProcess(manager, pendingProcess(), 'sleep 60', 'stop test');
const result = await executeTool(
new TaskStopTool(manager),
toolContext('task_stop_silent', { task_id: taskId }),
);
await new Promise((resolve) => setTimeout(resolve, 20));
expect(result.isError ?? false).toBe(false);
expect(outputString(result)).toContain('status: killed');
expect(agent.turn.steer).not.toHaveBeenCalled();
expect(agent.context.appendUserMessage).not.toHaveBeenCalled();
expect(manager.getTask(taskId)).toMatchObject({
status: 'killed',
terminalNotificationSuppressed: true,
});
});
it('TaskStopTool persists stop reason and suppression across reload', async () => {
const sessionDir = await mkdtemp(join(tmpdir(), 'kimi-bg-tool-stop-'));
let writerFixture: TaskServiceFixture | undefined;
let readerFixture: TaskServiceFixture | undefined;
try {
writerFixture = createAgentTaskService({ sessionDir });
const taskId = registerProcess(
writerFixture.manager,
pendingProcess(),
'sleep 60',
'persist stop',
);
const result = await executeTool(
new TaskStopTool(writerFixture.manager),
toolContext('task_stop_persisted', { task_id: taskId, reason: 'operator cancelled' }),
);
expect(result.isError ?? false).toBe(false);
readerFixture = createAgentTaskService({ sessionDir });
const { agent, manager: reader } = readerFixture;
await reader.loadFromDisk();
expect(reader.getTask(taskId)).toMatchObject({
stopReason: 'operator cancelled',
terminalNotificationSuppressed: true,
});
await reader.reconcile();
await new Promise((resolve) => setTimeout(resolve, 20));
expect(agent.context.appendUserMessage).not.toHaveBeenCalled();
expect(agent.turn.steer).not.toHaveBeenCalled();
} finally {
if (readerFixture !== undefined) {
await readerFixture.ctx.dispose();
}
await cleanupSessionDir(sessionDir, writerFixture);
}
});
it('replays restored terminal agent task notifications when undelivered', async () => {
const sessionDir = await mkdtemp(join(tmpdir(), 'kimi-bg-agent-replay-'));
let fixture: TaskServiceFixture | undefined;
@ -513,7 +591,7 @@ describe('AgentTaskService — notification delivery', () => {
notificationId: 'task:agent-done0000:completed',
});
const text = message.content[0]!.text;
expect(text).toContain('Task agent completed');
expect(text).toContain('Background agent completed');
expect(text).not.toContain('restored subagent summary');
expect(text).toContain('<output-file');
expect(text).toContain(persistence.taskOutputFile('agent-done0000'));
@ -547,7 +625,7 @@ describe('AgentTaskService — notification delivery', () => {
notificationId: 'task:bash-done0000:completed',
});
const text = message.content[0]!.text;
expect(text).toContain('Task process completed');
expect(text).toContain('Background process completed');
expect(text).not.toContain('restored shell output');
expect(text).toContain('<output-file');
expect(text).toContain(persistence.taskOutputFile('bash-done0000'));
@ -654,7 +732,7 @@ describe('AgentTaskService — notification delivery', () => {
notificationId: 'task:agent-run00000:lost',
});
expect(message.content[0]!.text).toContain(
'Task agent lost',
'Background agent lost',
);
} finally {
await cleanupSessionDir(sessionDir, fixture);
@ -684,10 +762,10 @@ describe('AgentTaskService — notification delivery', () => {
inputData: {
sink: 'context',
notificationType: 'task.completed',
title: 'Task agent completed',
title: 'Background agent completed',
body: 'inspect repository completed.',
severity: 'info',
sourceKind: 'task',
sourceKind: 'background_task',
sourceId: taskId,
},
}));
@ -733,10 +811,10 @@ describe('AgentTaskService — notification delivery', () => {
inputData: {
sink: 'context',
notificationType: 'task.completed',
title: 'Task process completed',
title: 'Background process completed',
body: 'done completed.',
severity: 'info',
sourceKind: 'task',
sourceKind: 'background_task',
sourceId: taskId,
},
}));

View file

@ -1,11 +1,19 @@
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { Readable, type Writable } from 'node:stream';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { SyncDescriptor } from '#/_base/di/descriptors';
import { DisposableStore, toDisposable } from '#/_base/di/lifecycle';
import { TestInstantiationService } from '#/_base/di/test';
import { IAgentTaskService, type AgentTask } from '#/agent/task/task';
import {
IAgentTaskService,
type AgentTask,
type AgentTaskInfo,
} from '#/agent/task/task';
import { renderNotificationXml } from '#/agent/task/notificationXml';
import { AgentTaskService } from '#/agent/task/taskService';
import { ProcessTask } from '#/os/backends/node-local/tools/process-task';
import type { IProcess } from '#/session/process/processRunner';
import { IConfigRegistry, IConfigService } from '#/app/config/config';
import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory';
import { IAgentPromptService } from '#/agent/prompt/prompt';
@ -119,6 +127,254 @@ describe('AgentTaskService', () => {
expect(await svc.readOutput(id)).toBe('');
await svc.stop(id);
});
// ── Output ceiling for shell (process) tasks ─────────────────────────
//
// A single shell command that streams more output than the per-command limit
// must be force-terminated instead of growing the (unbounded) live-forward
// buffer or the on-disk write chain until the process runs out of memory or
// fills the disk. The ceiling applies to process tasks, foreground and
// background alike. Subagent and user-question tasks append their bounded
// result in one shot and must always be persisted, so they are not capped.
const MiB = 1024 * 1024;
const LIMIT_BYTES = 16 * MiB;
/**
* A process that streams `chunks` of stdout, then exits 0 on its own unless
* it is killed first, in which case `wait()` resolves with the signal's exit
* code and the stream is destroyed (simulating the child dying on SIGTERM).
*/
function streamingProcess(chunks: string[]): {
proc: IProcess;
kill: ReturnType<typeof vi.fn>;
} {
const stdout = Readable.from(chunks);
const stderr = Readable.from([]);
let resolveWait!: (code: number) => void;
const waitP = new Promise<number>((resolve) => {
resolveWait = resolve;
});
stdout.on('end', () => {
resolveWait(0);
});
const kill = vi.fn(async (signal: string) => {
stdout.destroy();
resolveWait(signal === 'SIGKILL' ? 137 : 143);
});
const proc = {
stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable,
stdout,
stderr,
pid: 4242,
exitCode: null,
wait: () => waitP,
kill,
dispose: vi.fn().mockResolvedValue(undefined),
} as unknown as IProcess;
return { proc, kill };
}
/**
* A process that keeps streaming all of `chunks` regardless of SIGTERM (only
* SIGKILL stops it) simulating a producer that ignores the graceful stop
* and keeps writing through the SIGTERM grace window.
*/
function sigtermIgnoringProcess(chunks: string[]): {
proc: IProcess;
kill: ReturnType<typeof vi.fn>;
} {
const stdout = Readable.from(chunks);
const stderr = Readable.from([]);
let resolveWait!: (code: number) => void;
const waitP = new Promise<number>((resolve) => {
resolveWait = resolve;
});
stdout.on('end', () => {
resolveWait(0);
});
const kill = vi.fn(async (signal: string) => {
if (signal === 'SIGKILL') {
stdout.destroy();
resolveWait(137);
}
// SIGTERM is intentionally ignored.
});
const proc = {
stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable,
stdout,
stderr,
pid: 4243,
exitCode: null,
wait: () => waitP,
kill,
dispose: vi.fn().mockResolvedValue(undefined),
} as unknown as IProcess;
return { proc, kill };
}
/** One-shot non-process task appending its full result at once, like a subagent. */
function agentLikeTask(result: string, description: string): AgentTask {
return {
idPrefix: 'agent',
kind: 'agent',
description,
start: async (sink) => {
sink.appendOutput(result);
await sink.settle({ status: 'completed' });
},
toInfo: (base) => ({ ...base, kind: 'agent' }),
};
}
async function waitForTerminal(
svc: IAgentTaskService,
taskId: string,
timeoutMs = 30_000,
): Promise<AgentTaskInfo | undefined> {
const deadline = Date.now() + timeoutMs;
while (Date.now() <= deadline) {
const info = await svc.wait(taskId, 5);
if (
info?.status === 'completed' ||
info?.status === 'failed' ||
info?.status === 'timed_out' ||
info?.status === 'killed' ||
info?.status === 'lost'
) {
return info;
}
await new Promise((resolve) => setTimeout(resolve, 1));
}
return svc.getTask(taskId);
}
/** Re-stub the byte store so `output.log` appends are counted, then build the service. */
function serviceWithAppendCounter(): {
svc: IAgentTaskService;
persistedChars: () => number;
} {
let persistedChars = 0;
ix.stub(IFileSystemStorageService, {
read: async () => undefined,
readStream: async function* () {},
write: async () => {},
append: async (_scope: string, _key: string, chunk: Uint8Array) => {
persistedChars += chunk.byteLength;
},
list: async () => [],
delete: async () => {},
flush: async () => {},
close: async () => {},
});
return { svc: ix.get(IAgentTaskService), persistedChars: () => persistedChars };
}
it('terminates a foreground command that exceeds the output limit and stops forwarding', async () => {
const svc = ix.get(IAgentTaskService);
// 20 MiB total, well past the 16 MiB ceiling.
const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB));
const { proc, kill } = streamingProcess(chunks);
let forwardedChars = 0;
const onOutput = vi.fn((_kind: 'stdout' | 'stderr', text: string) => {
forwardedChars += text.length;
});
const taskId = svc.registerTask(
new ProcessTask(proc, 'b3sum --length 18446744073709551615', 'hash', onOutput),
{ detached: false, signal: new AbortController().signal, timeoutMs: 60_000 },
);
const info = await waitForTerminal(svc, taskId);
expect(info?.status).toBe('killed');
expect(info?.stopReason ?? '').toMatch(/output limit/i);
expect(kill).toHaveBeenCalledWith('SIGTERM');
// The live-forward path is capped at the ceiling rather than draining the
// full 20 MiB into the (unbounded) transcript/stderr buffer.
expect(forwardedChars).toBeLessThanOrEqual(LIMIT_BYTES);
});
it('also terminates a detached (background) task that exceeds the output limit', async () => {
const svc = ix.get(IAgentTaskService);
const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB));
const { proc, kill } = streamingProcess(chunks);
const taskId = svc.registerTask(new ProcessTask(proc, 'producer', 'bg'), {
detached: true,
timeoutMs: 60_000,
});
const info = await waitForTerminal(svc, taskId);
expect(info?.status).toBe('killed');
expect(info?.stopReason ?? '').toMatch(/output limit/i);
expect(kill).toHaveBeenCalledWith('SIGTERM');
});
it('stops enqueuing output to disk once the foreground cap trips', async () => {
const { svc, persistedChars } = serviceWithAppendCounter();
// 20 MiB, and the producer ignores SIGTERM so it keeps writing through
// the whole grace window.
const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB));
const { proc } = sigtermIgnoringProcess(chunks);
const taskId = svc.registerTask(new ProcessTask(proc, 'runaway', 'hash', () => {}), {
detached: false,
signal: new AbortController().signal,
timeoutMs: 60_000,
});
const info = await waitForTerminal(svc, taskId);
expect(info?.status).toBe('killed');
// Before the fix every chunk of the 20 MiB is enqueued into the disk
// write chain (retaining each string until its write drains); afterwards
// enqueuing stops at the ceiling so the chain cannot grow unbounded.
expect(persistedChars()).toBeLessThanOrEqual(17 * MiB);
});
it('stops enqueuing output to disk once the cap trips for a background task', async () => {
const { svc, persistedChars } = serviceWithAppendCounter();
// 20 MiB, and the producer ignores SIGTERM so it keeps writing through
// the whole grace window. Background tasks share the same ceiling.
const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB));
const { proc } = sigtermIgnoringProcess(chunks);
const taskId = svc.registerTask(new ProcessTask(proc, 'runaway', 'bg', () => {}), {
detached: true,
timeoutMs: 60_000,
});
const info = await waitForTerminal(svc, taskId);
expect(info?.status).toBe('killed');
// Same guarantee as the foreground case: once the cap trips, subsequent
// chunks are dropped before they reach the disk write chain.
expect(persistedChars()).toBeLessThanOrEqual(17 * MiB);
});
it('does not cap or drop a detached subagent result larger than the limit', async () => {
const { svc, persistedChars } = serviceWithAppendCounter();
// 20 MiB result — well past the 16 MiB ceiling — delivered in one shot,
// exactly how a subagent appends its completed result.
const bigResult = 'y'.repeat(20 * MiB);
const taskId = svc.registerTask(agentLikeTask(bigResult, 'big subagent result'), {
detached: true,
timeoutMs: 60_000,
});
const info = await waitForTerminal(svc, taskId);
// Non-process tasks must complete normally and have their full result
// persisted; the shell-output ceiling must not drop it.
expect(info?.status).toBe('completed');
expect(persistedChars()).toBeGreaterThanOrEqual(bigResult.length);
});
});
describe('Agent task notification XML', () => {
@ -127,7 +383,7 @@ describe('Agent task notification XML', () => {
id: 'n_"1&2',
category: 'task',
type: 'task.done',
source_kind: 'task',
source_kind: 'background_task',
source_id: 'bg&1',
title: 'Task finished',
severity: 'info',
@ -158,12 +414,12 @@ describe('Agent task notification XML', () => {
id: 'n_lost1',
category: 'task',
type: 'task.lost',
source_kind: 'task',
source_kind: 'background_task',
source_id: 'agent-w7gq3wwj',
agent_id: 'agent-0',
title: 'Task agent lost',
title: 'Background agent lost',
severity: 'warning',
body: 'Task agent 1 lost.',
body: 'Background agent 1 lost.',
});
expect(text).toContain('source_id="agent-w7gq3wwj"');
@ -175,9 +431,9 @@ describe('Agent task notification XML', () => {
id: 'n_bash',
category: 'task',
type: 'task.completed',
source_kind: 'task',
source_kind: 'background_task',
source_id: 'bash-abcdef00',
title: 'Task completed',
title: 'Background task completed',
severity: 'info',
body: 'echo done completed.',
});

View file

@ -30,6 +30,9 @@ import {
import type { ITaskHandle } from '#/app/task/task';
import type { ProcessTaskInfo } from '#/os/backends/node-local/tools/process-task';
import type { SubagentTaskInfo } from '#/session/agentLifecycle/tools/subagent-task';
import { TaskListTool as V1TaskListTool } from '../../../agent-core/src/tools/background/task-list';
import { TaskOutputTool as V1TaskOutputTool } from '../../../agent-core/src/tools/background/task-output';
import { TaskStopTool as V1TaskStopTool } from '../../../agent-core/src/tools/background/task-stop';
import { executeTool } from '../tools/fixtures/execute-tool';
const signal = new AbortController().signal;
@ -43,6 +46,21 @@ function outputString(result: { readonly output: string | readonly unknown[] }):
return result.output as string;
}
interface ModelFacingToolContract {
readonly name: string;
readonly description: string;
readonly parameters: Record<string, unknown>;
}
function expectModelFacingParity(
actual: ModelFacingToolContract,
expected: ModelFacingToolContract,
): void {
expect(actual.name).toBe(expected.name);
expect(actual.description).toBe(expected.description);
expect(JSON.stringify(actual.parameters)).toBe(JSON.stringify(expected.parameters));
}
function processTask(
overrides: Partial<ProcessTaskInfo> = {},
): ProcessTaskInfo {
@ -714,6 +732,12 @@ describe('TaskStopTool', () => {
describe('task tool descriptions', () => {
const tasks = new FakeTaskService();
it('matches the v1 model-facing contract exactly', () => {
expectModelFacingParity(new TaskListTool(tasks), new V1TaskListTool({} as never));
expectModelFacingParity(new TaskOutputTool(tasks), new V1TaskOutputTool({} as never));
expectModelFacingParity(new TaskStopTool(tasks), new V1TaskStopTool({} as never));
});
it('TaskOutput description mentions background tasks, block, output_path, and Read', () => {
const description = new TaskOutputTool(tasks).description;

View file

@ -16,10 +16,10 @@ import {
} from '#/session/agentLifecycle/agentLifecycle';
import { ISessionTodoService } from '#/session/todo/sessionTodo';
import { SessionTodoService } from '#/session/todo/sessionTodoService';
import { type TodoItem } from '#/session/todo/todoItem';
import { readTodoItems, type TodoItem } from '#/session/todo/todoItem';
import { TODO_LIST_REMINDER_VARIANT } from '#/session/todo/todoListReminder';
import { IAgentWireService } from '#/wire/tokens';
import type { IWireService } from '#/wire/wireService';
import type { IWireService, PersistedRecord } from '#/wire/wireService';
interface RecordedTodoSet {
readonly todos: readonly TodoItem[];
@ -30,14 +30,20 @@ interface FakeAgent {
readonly registeredTools: string[];
readonly registeredVariants: string[];
readonly appended: RecordedTodoSet[];
readonly resumers: Array<(record: RecordedTodoSet) => void>;
readonly subscribed: () => number;
readonly replay: (records: readonly PersistedRecord[]) => Promise<void>;
}
function makeFakeAgent(agentId: string): FakeAgent {
const registeredTools: string[] = [];
const registeredVariants: string[] = [];
const appended: RecordedTodoSet[] = [];
const resumers: Array<(record: RecordedTodoSet) => void> = [];
let todoState: readonly TodoItem[] = [];
type Subscriber = (state: readonly TodoItem[], prev: readonly TodoItem[]) => void;
const subscribers: Subscriber[] = [];
const restoredHandlers: Array<() => void> = [];
let subscribedCount = 0;
const registryStub = {
_serviceBrand: undefined,
@ -72,36 +78,50 @@ function makeFakeAgent(agentId: string): FakeAgent {
isToolActive: () => false,
};
let todoState: readonly TodoItem[] = [];
const wireStub: IWireService = {
_serviceBrand: undefined,
dispatch: (...ops: unknown[]) => {
for (const raw of ops) {
const op = raw as { type: string; payload: unknown };
const payload = op.payload;
if (payload !== null && typeof payload === 'object' && !Array.isArray(payload)) {
const record = payload as Record<string, unknown>;
if (Array.isArray(record['todos'])) {
todoState = record['todos'] as readonly TodoItem[];
const record =
payload !== null && typeof payload === 'object' && !Array.isArray(payload)
? (payload as Record<string, unknown>)
: { payload };
appended.push({ type: op.type, ...record } as unknown as RecordedTodoSet);
if (op.type === 'todo.set') {
const prev = todoState;
todoState = readTodoItems(record['todos']);
if (prev !== todoState) {
for (const h of [...subscribers]) h(todoState, prev);
}
appended.push({ type: op.type, ...record } as unknown as RecordedTodoSet);
} else {
appended.push({ type: op.type, payload } as unknown as RecordedTodoSet);
}
}
},
replay: async () => {},
replay: async (...records: PersistedRecord[]) => {
for (const record of records) {
if (record.type === 'todo.set') {
todoState = readTodoItems(record['todos']);
}
}
// Replay is silent: subscribers are NOT notified. onRestored fires after.
for (const h of restoredHandlers) h();
},
signal: () => {},
flush: async () => {},
attach: () => toDisposable(() => {}),
getModel: () => todoState,
subscribe: () => toDisposable(() => {}),
subscribe: (_model: unknown, handler: unknown) => {
subscribedCount += 1;
subscribers.push(handler as Subscriber);
return toDisposable(() => {
const i = subscribers.indexOf(handler as Subscriber);
if (i >= 0) subscribers.splice(i, 1);
});
},
onEmission: () => toDisposable(() => {}),
onRestored: (handler: () => void) => {
resumers.push((record: RecordedTodoSet) => {
todoState = record.todos;
handler();
});
restoredHandlers.push(handler);
return toDisposable(() => {});
},
} as unknown as IWireService;
@ -125,7 +145,14 @@ function makeFakeAgent(agentId: string): FakeAgent {
dispose: () => {},
};
return { handle, registeredTools, registeredVariants, appended, resumers };
return {
handle,
registeredTools,
registeredVariants,
appended,
subscribed: () => subscribedCount,
replay: (records) => wireStub.replay(...records),
};
}
interface LifecycleStub {
@ -184,8 +211,9 @@ function makeLifecycleStub(handles: readonly IAgentScopeHandle[] = []): Lifecycl
}
describe('SessionTodoService', () => {
it('starts empty and updates in-memory list on setTodos', () => {
const lifecycle = makeLifecycleStub();
it('starts empty and updates the list on setTodos', () => {
const main = makeFakeAgent('main');
const lifecycle = makeLifecycleStub([main.handle]);
const service = new SessionTodoService(lifecycle.service);
expect(service.getTodos()).toEqual([]);
@ -202,7 +230,8 @@ describe('SessionTodoService', () => {
});
it('fires onDidChange after each setTodos', () => {
const lifecycle = makeLifecycleStub();
const main = makeFakeAgent('main');
const lifecycle = makeLifecycleStub([main.handle]);
const service = new SessionTodoService(lifecycle.service);
const seen: Array<readonly TodoItem[]> = [];
@ -232,9 +261,10 @@ describe('SessionTodoService', () => {
it('does not append to the wire when the main agent is absent', () => {
const lifecycle = makeLifecycleStub();
const service = new SessionTodoService(lifecycle.service);
// Should not throw even without a main agent.
// Should not throw even without a main agent. With no main wire there is
// no source of truth to read from, so the list stays empty.
expect(() => service.setTodos([{ title: 'x', status: 'pending' }])).not.toThrow();
expect(service.getTodos()).toEqual([{ title: 'x', status: 'pending' }]);
expect(service.getTodos()).toEqual([]);
});
it('binds the stale-todo reminder into every created agent', () => {
@ -254,25 +284,23 @@ describe('SessionTodoService', () => {
expect(sub.registeredVariants).toContain(TODO_LIST_REMINDER_VARIANT);
});
it('registers the todo.set resume resumer only on the main agent', () => {
it('subscribes to TodoModel only on the main agent', () => {
const main = makeFakeAgent('main');
const sub = makeFakeAgent('agent-1');
const lifecycle = makeLifecycleStub([main.handle, sub.handle]);
const service = new SessionTodoService(lifecycle.service);
void service;
expect(main.resumers).toHaveLength(1);
expect(sub.resumers).toHaveLength(0);
expect(main.subscribed()).toBe(1);
expect(sub.subscribed()).toBe(0);
});
it('rebuilds the in-memory list when a todo.set record is resumed', () => {
it('rebuilds the list when a todo.set record is replayed', async () => {
const main = makeFakeAgent('main');
const lifecycle = makeLifecycleStub([main.handle]);
const service = new SessionTodoService(lifecycle.service);
const resumer = main.resumers[0];
expect(resumer).toBeDefined();
resumer!({ todos: [{ title: 'restored', status: 'done' }] });
await main.replay([{ type: 'todo.set', todos: [{ title: 'restored', status: 'done' }] }]);
expect(service.getTodos()).toEqual([{ title: 'restored', status: 'done' }]);
});
@ -297,4 +325,37 @@ describe('SessionTodoService', () => {
expect(typeof service.clear).toBe('function');
expect(typeof service.onDidChange).toBe('function');
});
it('cleans malformed items from a replayed todo.set record', async () => {
const main = makeFakeAgent('main');
const lifecycle = makeLifecycleStub([main.handle]);
const service = new SessionTodoService(lifecycle.service);
await main.replay([
{
type: 'todo.set',
todos: [
{ title: 'valid', status: 'done' },
{ title: 'missing status' },
{ title: 123, status: 'pending' },
'garbage',
{ title: 'bad status', status: 'wip' },
],
} as unknown as PersistedRecord,
]);
expect(service.getTodos()).toEqual([{ title: 'valid', status: 'done' }]);
});
it('treats a non-array todo.set payload as an empty list on replay', async () => {
const main = makeFakeAgent('main');
const lifecycle = makeLifecycleStub([main.handle]);
const service = new SessionTodoService(lifecycle.service);
await main.replay([
{ type: 'todo.set', todos: 'not-an-array' } as unknown as PersistedRecord,
]);
expect(service.getTodos()).toEqual([]);
});
});

View file

@ -69,6 +69,9 @@ function agentSwarmSchemaProperties<T = unknown>(): Record<string, T> {
).properties;
}
const BACKGROUND_AGENT_NEXT_STEP =
'next_step: The completion arrives automatically in a later turn — do NOT wait, poll, or call TaskOutput on it; continue with other work or hand back to the user. (If you have nothing to do until it finishes, run such tasks in the foreground next time.)';
function deferred<T>(): {
readonly promise: Promise<T>;
resolve(value: T): void;
@ -721,6 +724,56 @@ describe('Agent tool execution contract', () => {
completions[0]?.resolve({ summary: 'finished later' });
});
it('rejects one of two concurrent background subagents when the task limit is reached', async () => {
const completions = [
deferred<{ readonly summary: string }>(),
deferred<{ readonly summary: string }>(),
];
const lifecycle = createAgentLifecycleStub({
createAgentIds: ['agent-first', 'agent-second'],
runCompletion: (_agentId, _request, options) => {
const next = completions.shift();
if (next === undefined) throw new Error('unexpected run');
options.signal.addEventListener('abort', () => next.reject(options.signal.reason), {
once: true,
});
return next.promise;
},
});
const context = createAgentToolContext(
lifecycle,
configServices(() => ({
providers: {},
task: { maxRunningTasks: 1 },
})),
);
const first = executeAgentTool(context, {
prompt: 'Investigate first',
description: 'Find first',
run_in_background: true,
});
const second = executeAgentTool(context, {
prompt: 'Investigate second',
description: 'Find second',
run_in_background: true,
});
const results = await Promise.all([first, second]);
expect(lifecycle.create).toHaveBeenCalledTimes(2);
expect(results).toContainEqual(
expect.objectContaining({ output: expect.stringContaining('status: running') }),
);
expect(results).toContainEqual(
expect.objectContaining({
isError: true,
output: 'Too many background tasks are already running.',
}),
);
completions[0]?.resolve({ summary: 'finished later' });
});
it('logs background registration failures', async () => {
const { entries, logger } = captureLogs();
const completions = [
@ -889,7 +942,7 @@ describe('Agent tool execution contract', () => {
const taskId = result.output.match(/task_id: (agent-[0-9a-z]{8})/)?.[1];
expect(taskId).toBeDefined();
expect(result.output).toContain('next_step:');
expect(result.output).toContain('do NOT wait, poll, or call TaskOutput on it');
expect(result.output).toContain(BACKGROUND_AGENT_NEXT_STEP);
expect(result.output).not.toContain('block=false');
expect(result.output).toContain('resume_hint:');
expect(result.output).toContain('Agent(resume="agent-child"');

View file

@ -205,6 +205,9 @@ describe('AgentUsageService (wire-backed)', () => {
turnId: 1,
context: { type: 'turn', turnId: 2 },
}),
).rejects.toThrow('conflicting turnId');
).rejects.toMatchObject({
code: 'usage.turn_id_conflict',
name: 'UsageError',
});
});
});

View file

@ -239,6 +239,7 @@ export type KimiErrorCode =
| 'goal.not_resumable'
| 'model.not_configured'
| 'model.config_invalid'
| 'profile.thinking_alias_conflict'
| 'model.not_found'
| 'auth.login_required'
| 'auth.provisioning_required'
@ -260,6 +261,7 @@ export type KimiErrorCode =
| 'compaction.failed'
| 'compaction.unable'
| 'task.task_id_empty'
| 'usage.turn_id_conflict'
| 'mcp.server_not_found'
| 'mcp.server_disabled'
| 'mcp.startup_failed'
@ -1087,6 +1089,7 @@ export const kimiErrorCodeSchema = z.enum([
'goal.not_resumable',
'model.not_configured',
'model.config_invalid',
'profile.thinking_alias_conflict',
'model.not_found',
'auth.login_required',
'auth.provisioning_required',
@ -1108,6 +1111,7 @@ export const kimiErrorCodeSchema = z.enum([
'compaction.failed',
'compaction.unable',
'task.task_id_empty',
'usage.turn_id_conflict',
'mcp.server_not_found',
'mcp.server_disabled',
'mcp.startup_failed',