vrr/kimi-code
2
0
Fork 0
mirror of https://github.com/MoonshotAI/kimi-code.git synced 2026-08-26 00:54:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix(agent-core-v2): reduce service constructor args

Browse source
This commit is contained in:
_Kerman 2026-06-30 17:03:55 +08:00
parent 567c384840
commit 8d4e702c16
16 changed files with 252 additions and 280 deletions
Download patch file Download diff file Expand all files Collapse all files

15
packages/agent-core-v2/src/cron/configSection.ts
View file

@ -2,8 +2,8 @@
* `cron` domain (L3) — cron operational-config section env bindings.
*
* Declares the `KIMI_CRON_*` environment bindings for the cron operational
* toggles (debug / jitter / stale / killswitch / manual tick / clock). Applied
* to the effective `cron` value by `config`; operational overrides, never
* toggles (debug / jitter / stale / killswitch / manual tick / clock /
* poll interval). Applied to the effective `cron` value by `config`; never
* persisted to `config.toml`.
*/
@ -18,6 +18,7 @@ export interface CronConfig {
readonly disabled: boolean;
readonly manualTick: boolean;
readonly clock?: string;
readonly pollIntervalMs?: number | null;
}
export const DEFAULT_CRON_CONFIG: CronConfig = {
@ -32,6 +33,15 @@ const cronConfigSchema = { parse: (value: unknown): CronConfig => value as CronC
const on = (raw: string): boolean => raw === '1';
function parsePollIntervalMs(raw: string): number | null | undefined {
const value = raw.trim();
if (value.length === 0) return undefined;
if (value === 'null') return null;
const parsed = Number(value);
if (!Number.isFinite(parsed) || !Number.isInteger(parsed) || parsed < 0) return undefined;
return parsed;
}
export const cronEnvBindings: EnvBindings<CronConfig> = envBindings(cronConfigSchema, {
debug: { env: 'KIMI_CRON_DEBUG', parse: on },
noJitter: { env: 'KIMI_CRON_NO_JITTER', parse: on },
@ -39,6 +49,7 @@ export const cronEnvBindings: EnvBindings<CronConfig> = envBindings(cronConfigSc
disabled: { env: 'KIMI_DISABLE_CRON', parse: on },
manualTick: { env: 'KIMI_CRON_MANUAL_TICK', parse: on },
clock: 'KIMI_CRON_CLOCK',
pollIntervalMs: { env: 'KIMI_CRON_POLL_INTERVAL_MS', parse: parsePollIntervalMs },
});
export const stripCronEnv: ConfigStripEnv<CronConfig> = () => undefined;

8
packages/agent-core-v2/src/cron/cron.ts
View file

@ -1,7 +1,6 @@
import type { ContentPart } from '@moonshot-ai/kosong';
import { createDecorator } from "#/_base/di";
import type { ClockSources } from './tools/clock';
import type { SessionCronTaskInit } from './tools/session-store';
import type { CronTask, CronToolManager } from './tools/types';
import type { Turn } from '#/turn';
@ -15,14 +14,7 @@ export interface CronPersistence {
}
export interface CronOptions {
readonly persistence?: CronPersistence;
readonly homedir?: string;
readonly isSubagent?: boolean;
readonly clocks?: ClockSources;
readonly pollIntervalMs?: number | null;
readonly autoStart?: boolean;
readonly registerTools?: boolean;
readonly onPersistenceError?: (error: unknown, taskId: string) => void;
}
export interface CronLoadOptions {

29
packages/agent-core-v2/src/cron/cronService.ts
View file

@ -99,7 +99,7 @@ export class CronService
private sigusr1Handler: NodeJS.SignalsListener | null = null;
constructor(
private readonly options: CronOptions = {},
options: CronOptions = {},
@IPromptService private readonly prompt: IPromptService,
@IEventSink private readonly events: IEventSink,
@IWireRecord private readonly wireRecord: IWireRecord,
@ -126,16 +126,9 @@ export class CronService
}),
);
this.clocks =
options.clocks ??
resolveClockSources(this.cronConfig.clock, this.cronConfig.debug) ??
SYSTEM_CLOCKS;
this.persistStore =
this.enabled
? options.persistence ??
(options.homedir === undefined
? undefined
: createCronPersistStore(this.atomicDocs))
: undefined;
this.persistStore = this.enabled ? createCronPersistStore(this.atomicDocs) : undefined;
this._register(
wireRecord.register('cron.add', (record) => {
@ -181,20 +174,16 @@ export class CronService
pollIntervalMs:
this.cronConfig.manualTick
? null
: options.pollIntervalMs,
: this.cronConfig.pollIntervalMs,
debug: this.cronConfig.debug,
noJitter: this.cronConfig.noJitter,
});
if (options.registerTools !== false) {
this._register(this.toolRegistry.register(new CronCreateTool(this, this.cronConfig.disabled)));
this._register(this.toolRegistry.register(new CronListTool(this)));
this._register(this.toolRegistry.register(new CronDeleteTool(this)));
}
this._register(this.toolRegistry.register(new CronCreateTool(this, this.cronConfig.disabled)));
this._register(this.toolRegistry.register(new CronListTool(this)));
this._register(this.toolRegistry.register(new CronDeleteTool(this)));
if (options.autoStart !== false) {
this.start();
}
this.start();
}
this._register(
@ -417,9 +406,7 @@ export class CronService
const next = prev
.catch(() => { })
.then(() => work())
.catch((error: unknown) => {
this.options.onPersistenceError?.(error, id);
})
.catch(() => { })
.finally(() => {
if (this.persistQueues.get(id) === next) {
this.persistQueues.delete(id);

13
packages/agent-core-v2/src/cron/tools/session-store.ts
View file

@ -2,16 +2,15 @@
* SessionCronStore — in-memory cron task store for a single CLI session.
*
* The store itself is purely in-memory; cross-restart persistence is
* layered on top by `CronManager.addTask` / `removeTasks`, which
* layered on top by `CronService.addTask` / `removeTasks`, which
* mirror every mutation to `<sessionDir>/cron/<id>.json`. On resume
* the manager calls {@link adopt} to put each persisted task back into
* the service calls {@link adopt} to put each persisted task back into
* the store with its original id and `createdAt` preserved.
*
* The store is intentionally clock-agnostic: it does NOT call
* `Date.now()` itself. Callers pass `nowMs` (which the cron manager
* sources from `ClockSources.wallNow()`), so injected clocks in tests
* and benches stay authoritative. The `no-date-now` guard does not
* currently list this file, but the discipline matches.
* `Date.now()` itself. Callers pass `nowMs`, so the service's clock
* source, tests, and benches stay authoritative. The `no-date-now`
* guard does not currently list this file, but the discipline matches.
*
* Insertion order is preserved by relying on `Map` iteration order —
* callers (CronList, scheduler `source: () => CronTask[]`) want a
@ -69,7 +68,7 @@ export class SessionCronStore {
/**
* Insert a previously-persisted task verbatim — id and createdAt
* stay as they are on disk. Used by `CronManager.loadFromDisk()` to
* stay as they are on disk. Used by `CronService.loadFromDisk()` to
* rehydrate the store on resume. Unlike {@link add}, this does NOT
* generate a new id; the caller is responsible for ensuring the id
* matches the expected shape (the persistence layer's regex /

1
packages/agent-core-v2/src/llmRequester/llmRequester.ts
View file

@ -3,7 +3,6 @@ import type { FinishReason, Message, StreamedMessagePart, TokenUsage, Tool } fro
import type { LLMRequestLogFields } from '#/loop';
import type { UsageRecordContext } from '#/usage';
export interface LLMRequestOverrides {
messages?: readonly Message[];
tools?: readonly Tool[];

10
packages/agent-core-v2/src/llmRequester/llmRequesterService.ts
View file

@ -16,7 +16,7 @@ import {
type Message,
type ProviderRequestAuth,
type Tool as KosongTool,
} from '@moonshot-ai/kosong';
} from '@moonshot-ai/kosong';
import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
@ -37,10 +37,6 @@ import { IUsageService } from '#/usage';
import { AsyncEventQueue } from './asyncEventQueue';
import { ILLMRequester } from './llmRequester';
export interface LLMRequesterServiceOptions {
readonly generate?: typeof generate;
}
const EMPTY_TOOL_PARAMETERS: Record<string, unknown> = {
type: 'object',
properties: {},
@ -48,8 +44,8 @@ const EMPTY_TOOL_PARAMETERS: Record<string, unknown> = {
export class LLMRequesterService implements ILLMRequester {
declare readonly _serviceBrand: undefined;
constructor(
private readonly options: LLMRequesterServiceOptions = {},
@IContextMemory private readonly context: IContextMemory,
@IContextProjector private readonly projector: IContextProjector,
@IToolRegistry private readonly tools: IToolRegistry,
@ -189,7 +185,7 @@ export class LLMRequesterService implements ILLMRequester {
messages: [...(overrides.messages ?? this.projector.project(this.context.get()))],
requestLogFields: overrides.requestLogFields,
usageContext: overrides.usageContext,
generate: this.options.generate ?? generate,
generate,
};
}

21
packages/agent-core-v2/src/microCompaction/configSection.ts Normal file
View file

@ -0,0 +1,21 @@
/**
* `microCompaction` domain (L4) - micro-compaction config-section schema.
*
* Owns the `[micro_compaction]` tuning section consumed by
* `MicroCompactionService`. Registered into `IConfigRegistry` by the
* micro-compaction service on construction.
*/
import { z } from 'zod';
export const MICRO_COMPACTION_SECTION = 'microCompaction';
export const MicroCompactionConfigSchema = z.object({
keepRecentMessages: z.number().int().min(0).optional(),
minContentTokens: z.number().int().min(0).optional(),
cacheMissedThresholdMs: z.number().int().min(0).optional(),
truncatedMarker: z.string().optional(),
minContextUsageRatio: z.number().min(0).max(1).optional(),
});
export type MicroCompactionConfigPatch = z.infer<typeof MicroCompactionConfigSchema>;

4
packages/agent-core-v2/src/microCompaction/index.ts
View file

@ -1,8 +1,10 @@
/**
* `microCompaction` domain barrel - re-exports the microCompaction service contract and implementation.
* `microCompaction` domain barrel - re-exports the microCompaction config
* section, flag contribution, service contract, and scoped implementation.
*/
import './flag';
export * from './configSection';
export * from './microCompaction';
export * from './flag';
export * from './microCompactionService';

12
packages/agent-core-v2/src/microCompaction/microCompaction.ts
View file

@ -1,3 +1,10 @@
/**
* `microCompaction` domain (L4) - micro-compaction service contract.
*
* Defines the truncation tuning model and the Agent-scoped
* `IMicroCompactionService` used by context projection. Bound at Agent scope.
*/
import { createDecorator } from "#/_base/di";
import type { ContextMessage } from '#/contextMemory';
@ -9,11 +16,6 @@ export interface MicroCompactionConfig {
minContextUsageRatio: number;
}
export interface MicroCompactionServiceOptions {
readonly config?: Partial<MicroCompactionConfig>;
readonly maxContextTokens?: () => number | undefined;
}
export interface MicroCompactionEffect {
readonly truncatedToolResultCount: number;
readonly truncatedToolResultTokensBefore: number;

54
packages/agent-core-v2/src/microCompaction/microCompactionService.ts
View file

@ -1,3 +1,12 @@
/**
* `microCompaction` domain (L4) - `IMicroCompactionService` implementation.
*
* Tracks cache-miss compaction cutoffs over `contextMemory`, sizes context via
* `contextSize`, resolves model capacity through `profile`, persists cutoffs
* through `wireRecord`, gates behavior through `flag`, emits telemetry, and
* participates in `turn` hooks. Bound at Agent scope.
*/
import type { ContentPart } from '@moonshot-ai/kosong';
import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
@ -10,6 +19,7 @@ import {
estimateTokensForMessages,
} from "#/_base/utils/tokens";
import type { TelemetryProperties } from '#/telemetry';
import { IConfigRegistry, IConfigService } from '#/config';
import { IContextMemory } from '#/contextMemory';
import { IContextSizeService } from '#/contextSize';
import { IFlagService } from '#/flag';
@ -22,8 +32,12 @@ import {
IMicroCompactionService,
type MicroCompactionConfig,
type MicroCompactionEffect,
type MicroCompactionServiceOptions,
} from './microCompaction';
import {
MICRO_COMPACTION_SECTION,
MicroCompactionConfigSchema,
type MicroCompactionConfigPatch,
} from './configSection';
declare module '#/wireRecord' {
interface WireRecordMap {
@ -47,11 +61,10 @@ export class MicroCompactionService
{
declare readonly _serviceBrand: undefined;
private cutoff = 0;
private readonly config: MicroCompactionConfig;
private microConfig: MicroCompactionConfig;
private _lastAssistantAt: number | null = null;
constructor(
private readonly options: MicroCompactionServiceOptions = {},
@IContextMemory private readonly context: IContextMemory,
@IContextSizeService private readonly contextSize: IContextSizeService,
@IWireRecord private readonly wireRecord: IWireRecord,
@ -59,9 +72,19 @@ export class MicroCompactionService
@IProfileService private readonly profile: IProfileService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@ITurnService turn: ITurnService,
@IConfigRegistry configRegistry: IConfigRegistry,
@IConfigService private readonly config: IConfigService,
) {
super();
this.config = { ...DEFAULT_CONFIG, ...options.config };
configRegistry.registerSection(MICRO_COMPACTION_SECTION, MicroCompactionConfigSchema);
this.microConfig = this.readConfig();
this._register(
this.config.onDidSectionChange((event) => {
if (event.domain === MICRO_COMPACTION_SECTION) {
this.microConfig = this.readConfig();
}
}),
);
this._register(
turn.hooks.beforeStep.register(
'micro-compaction',
@ -113,13 +136,13 @@ export class MicroCompactionService
if (lastAssistantAt === null) return;
const cacheAgeMs = Date.now() - lastAssistantAt;
if (cacheAgeMs < this.config.cacheMissedThresholdMs) return;
if (cacheAgeMs < this.microConfig.cacheMissedThresholdMs) return;
const history = this.context.get();
if (this.contextSizeRatio() < this.config.minContextUsageRatio) return;
if (this.contextSizeRatio() < this.microConfig.minContextUsageRatio) return;
const previousCutoff = this.cutoff;
const nextCutoff = Math.max(0, history.length - this.config.keepRecentMessages);
const nextCutoff = Math.max(0, history.length - this.microConfig.keepRecentMessages);
this.apply(nextCutoff);
if (previousCutoff === nextCutoff) return;
@ -127,7 +150,7 @@ export class MicroCompactionService
const previousEffect = this.measureEffect(history, previousCutoff);
const rawContextTokens = estimateTokensForMessages(history);
const properties: TelemetryProperties = {
...this.config,
...this.microConfig,
...effect,
tokensBefore:
rawContextTokens -
@ -156,7 +179,7 @@ export class MicroCompactionService
result.push({
...message,
content: [
{ type: 'text', text: this.config.truncatedMarker } satisfies ContentPart,
{ type: 'text', text: this.microConfig.truncatedMarker } satisfies ContentPart,
],
});
} else {
@ -193,12 +216,17 @@ export class MicroCompactionService
index < this.cutoff &&
message.role === 'tool' &&
message.toolCallId !== undefined &&
estimateTokensForContentParts(message.content) >= this.config.minContentTokens
estimateTokensForContentParts(message.content) >= this.microConfig.minContentTokens
);
}
private readConfig(): MicroCompactionConfig {
const config = this.config.get<MicroCompactionConfigPatch | undefined>(MICRO_COMPACTION_SECTION);
return { ...DEFAULT_CONFIG, ...config };
}
private contextSizeRatio(): number {
const maxContextTokens = this.options.maxContextTokens?.();
const maxContextTokens = this.profile.getModelCapabilities().max_context_tokens;
if (maxContextTokens === undefined || maxContextTokens <= 0) return 1;
return this.contextSize.getStatus().contextTokensWithPending / maxContextTokens;
}
@ -218,10 +246,10 @@ export class MicroCompactionService
}
const contentTokens = estimateTokensForContentParts(message.content);
if (contentTokens < this.config.minContentTokens) continue;
if (contentTokens < this.microConfig.minContentTokens) continue;
markerTokenCount ??= estimateTokensForContentParts([
{ type: 'text', text: this.config.truncatedMarker },
{ type: 'text', text: this.microConfig.truncatedMarker },
]);
truncatedToolResultCount += 1;
truncatedToolResultTokensBefore += contentTokens;

17
packages/agent-core-v2/test/cron/cron.e2e.test.ts
View file

@ -2,7 +2,7 @@
* Session-level cron end-to-end smoke: exercises the full
* `CronCreateTool → SessionCronStore → CronScheduler → CronManager →
* agent.turn.steer` pipeline through the real `AgentTestContext`,
* with a swapped CronManager wired to an injected clock so the
* with Date.now controlled by the test so the
* `coalescedCount = 3` calibration after a 15-minute advance is
* deterministic regardless of host TZ.
*/
@ -24,15 +24,10 @@ const LOCAL_ANCHOR_MS = new Date(2024, 5, 1, 12, 0, 0, 0).getTime();
function createClocks(initial = LOCAL_ANCHOR_MS) {
let wall = initial;
let mono = initial;
vi.spyOn(Date, 'now').mockImplementation(() => wall);
return {
clocks: {
wallNow: () => wall,
monoNowMs: () => mono,
},
advance(ms: number) {
wall += ms;
mono += ms;
},
};
}
@ -61,12 +56,9 @@ describe('Cron — session E2E (P1.9)', () => {
// itself — this flag is belt-and-braces against any future refactor
// that widens the jitter window past 10 minutes.
vi.stubEnv('KIMI_CRON_NO_JITTER', '1');
vi.stubEnv('KIMI_CRON_POLL_INTERVAL_MS', '0');
harness = createClocks();
ctx = createTestAgent(cronServices({
autoStart: false,
clocks: harness.clocks,
pollIntervalMs: null,
}));
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
cron.start();
@ -78,6 +70,7 @@ describe('Cron — session E2E (P1.9)', () => {
} finally {
await ctx.dispose();
vi.unstubAllEnvs();
vi.restoreAllMocks();
}
});

17
packages/agent-core-v2/test/cron/cron.test.ts
View file

@ -1,4 +1,4 @@
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { SyncDescriptor } from '#/_base/di/descriptors';
import { DisposableStore } from '#/_base/di/lifecycle';
@ -9,7 +9,6 @@ import { ConfigRegistry, ConfigService } from '#/config/configService';
import type { ContextMessage } from '#/contextMemory';
import { ICronService } from '#/cron';
import { CronService } from '#/cron/cronService';
import type { ClockSources } from '#/cron/tools/clock';
import { ILogService } from '#/log';
import { IPromptService } from '#/prompt';
import {
@ -59,16 +58,14 @@ describe('CronService', () => {
let steered: ContextMessage[];
beforeEach(() => {
vi.stubEnv('KIMI_CRON_POLL_INTERVAL_MS', '0');
disposables = new DisposableStore();
ix = disposables.add(new TestInstantiationService());
now = 0;
activeTurn = undefined;
steered = [];
vi.spyOn(Date, 'now').mockImplementation(() => now);
const clocks: ClockSources = {
wallNow: () => now,
monoNowMs: () => now,
};
const turnService: ITurnService = {
...stubTurn(),
getActiveTurn: () => activeTurn,
@ -103,10 +100,14 @@ describe('CronService', () => {
ix.set(IConfigService, new SyncDescriptor(ConfigService));
ix.set(
ICronService,
new SyncDescriptor(CronService, [{ autoStart: false, registerTools: false, clocks }]),
new SyncDescriptor(CronService, [{}]),
);
});
afterEach(() => disposables.dispose());
afterEach(() => {
disposables.dispose();
vi.restoreAllMocks();
vi.unstubAllEnvs();
});
it('addTask / list / removeTasks', () => {
const svc = ix.get(ICronService);

53
packages/agent-core-v2/test/cron/manager.test.ts
View file

@ -32,19 +32,13 @@ const WALL_ANCHOR = 1_700_000_000_000;
function createClocks(initial = WALL_ANCHOR) {
let wall = initial;
let mono = 1_000_000;
vi.spyOn(Date, 'now').mockImplementation(() => wall);
return {
clocks: {
wallNow: () => wall,
monoNowMs: () => mono,
},
setNow(v: number) {
wall = v;
mono = v;
},
advance(ms: number) {
wall += ms;
mono += ms;
},
now() {
return wall;
@ -95,6 +89,7 @@ describe('CronManager', () => {
// but setting it here as well shields the construction-path tests
// from any leaked state.
vi.stubEnv('KIMI_CRON_NO_JITTER', '1');
vi.stubEnv('KIMI_CRON_POLL_INTERVAL_MS', '0');
});
afterEach(async () => {
@ -104,6 +99,7 @@ describe('CronManager', () => {
try {
await ctx.dispose();
} finally {
vi.restoreAllMocks();
vi.unstubAllEnvs();
}
}
@ -111,7 +107,7 @@ describe('CronManager', () => {
describe('construction', () => {
beforeEach(() => {
ctx = createTestAgent(cronServices({ autoStart: false, pollIntervalMs: null }));
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
});
@ -144,9 +140,7 @@ describe('CronManager', () => {
beforeEach(() => {
harness = createClocks();
ctx = createTestAgent(
cronServices({ clocks: harness.clocks, autoStart: false, pollIntervalMs: null }),
);
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
telemetry = ctx.get(ITelemetryService);
@ -218,9 +212,7 @@ describe('CronManager', () => {
beforeEach(() => {
harness = createClocks();
ctx = createTestAgent(
cronServices({ clocks: harness.clocks, autoStart: false, pollIntervalMs: null }),
);
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
telemetry = ctx.get(ITelemetryService);
@ -265,9 +257,7 @@ describe('CronManager', () => {
beforeEach(() => {
harness = createClocks();
ctx = createTestAgent(
cronServices({ clocks: harness.clocks, autoStart: false, pollIntervalMs: null }),
);
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
});
@ -322,9 +312,7 @@ describe('CronManager', () => {
beforeEach(() => {
vi.stubEnv('KIMI_CRON_NO_STALE', '1');
harness = createClocks();
ctx = createTestAgent(
cronServices({ clocks: harness.clocks, autoStart: false, pollIntervalMs: null }),
);
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
});
@ -342,12 +330,9 @@ describe('CronManager', () => {
describe('isStale with a broken clock', () => {
beforeEach(() => {
vi.spyOn(Date, 'now').mockReturnValue(Number.NaN);
ctx = createTestAgent(
cronServices({
clocks: { wallNow: () => Number.NaN, monoNowMs: () => 0 },
autoStart: false,
pollIntervalMs: null,
}),
cronServices({}),
);
cron = ctx.get(ICronService);
});
@ -371,9 +356,7 @@ describe('CronManager', () => {
beforeEach(() => {
harness = createClocks();
ctx = createTestAgent(
cronServices({ clocks: harness.clocks, autoStart: false, pollIntervalMs: null }),
);
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
telemetry = ctx.get(ITelemetryService);
@ -441,9 +424,7 @@ describe('CronManager', () => {
beforeEach(() => {
harness = createClocks();
ctx = createTestAgent(
cronServices({ clocks: harness.clocks, autoStart: false, pollIntervalMs: null }),
);
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
telemetry = ctx.get(ITelemetryService);
@ -470,9 +451,7 @@ describe('CronManager', () => {
beforeEach(() => {
harness = createClocks();
ctx = createTestAgent(
cronServices({ clocks: harness.clocks, autoStart: false, pollIntervalMs: null }),
);
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
telemetry = ctx.get(ITelemetryService);
@ -515,9 +494,7 @@ describe('CronManager', () => {
beforeEach(() => {
harness = createClocks();
ctx = createTestAgent(
cronServices({ clocks: harness.clocks, autoStart: false, pollIntervalMs: null }),
);
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
steerCalls = createSteerSpy(prompt);
@ -541,7 +518,7 @@ describe('CronManager', () => {
let telemetryRecords: TelemetryRecord[];
beforeEach(() => {
ctx = createTestAgent(cronServices({ autoStart: false, pollIntervalMs: null }));
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
telemetry = ctx.get(ITelemetryService);

60
packages/agent-core-v2/test/cron/manual-tick.test.ts
View file

@ -13,8 +13,7 @@ import { createTestAgent, cronServices, type TestAgentContext } from '../harness
const WALL_ANCHOR = 1_700_000_000_000;
interface ClockHarness {
readonly clocks: { wallNow(): number; monoNowMs(): number };
/** Advance wall + mono by `ms`. */
/** Advance wall-clock time by `ms`. */
advance(ms: number): void;
/** Current wall-clock value. */
now(): number;
@ -22,15 +21,10 @@ interface ClockHarness {
function createClocks(initial: number = WALL_ANCHOR): ClockHarness {
let wall = initial;
let mono = 1_000_000;
vi.spyOn(Date, 'now').mockImplementation(() => wall);
return {
clocks: {
wallNow: () => wall,
monoNowMs: () => mono,
},
advance: (ms) => {
wall += ms;
mono += ms;
},
now: () => wall,
};
@ -53,6 +47,7 @@ describe('CronService — P1.8 manual tick + SIGUSR1', () => {
afterEach(() => {
vi.unstubAllEnvs();
vi.restoreAllMocks();
vi.useRealTimers();
});
@ -65,21 +60,13 @@ describe('CronService — P1.8 manual tick + SIGUSR1', () => {
beforeEach(() => {
vi.stubEnv('KIMI_CRON_MANUAL_TICK', '1');
harness = createClocks();
ctx = createTestAgent(cronServices({
autoStart: true,
pollIntervalMs: 50,
clocks: harness.clocks,
}));
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
});
afterEach(async () => {
try {
await ctx.expectResumeMatches();
} finally {
await ctx.dispose();
}
await ctx.dispose();
});
it('does not install setInterval; tick() must be called manually', async () => {
@ -111,22 +98,15 @@ describe('CronService — P1.8 manual tick + SIGUSR1', () => {
// Fake timers must be in place BEFORE the manager calls
// setInterval, otherwise the scheduler captures the real one.
vi.useFakeTimers();
vi.stubEnv('KIMI_CRON_POLL_INTERVAL_MS', '50');
harness = createClocks();
ctx = createTestAgent(cronServices({
autoStart: true,
pollIntervalMs: 50,
clocks: harness.clocks,
}));
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
});
afterEach(async () => {
try {
await ctx.expectResumeMatches();
} finally {
await ctx.dispose();
}
await ctx.dispose();
});
it('auto-tick fires when fake timers advance past pollIntervalMs', () => {
@ -155,16 +135,12 @@ describe('CronService — P1.8 manual tick + SIGUSR1', () => {
beforeEach(() => {
vi.stubEnv('KIMI_CRON_MANUAL_TICK', '1');
listenerCountBeforeCreate = process.listenerCount('SIGUSR1');
ctx = createTestAgent(cronServices({ autoStart: true, pollIntervalMs: null }));
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
});
afterEach(async () => {
try {
await ctx.expectResumeMatches();
} finally {
await ctx.dispose();
}
await ctx.dispose();
});
it('triggers tick() once per emit (POSIX only)', () => {
@ -232,16 +208,12 @@ describe('CronService — P1.8 manual tick + SIGUSR1', () => {
beforeEach(() => {
vi.stubEnv('KIMI_CRON_MANUAL_TICK', '1');
vi.stubEnv('KIMI_CRON_DEBUG', '1');
ctx = createTestAgent(cronServices({ autoStart: true, pollIntervalMs: null }));
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
});
afterEach(async () => {
try {
await ctx.expectResumeMatches();
} finally {
await ctx.dispose();
}
await ctx.dispose();
});
it('logs swallowed tick() throws to stderr when KIMI_CRON_DEBUG=1', () => {
@ -272,16 +244,12 @@ describe('CronService — P1.8 manual tick + SIGUSR1', () => {
let cron: ICronService;
beforeEach(() => {
ctx = createTestAgent(cronServices({ autoStart: true, pollIntervalMs: null }));
ctx = createTestAgent(cronServices({}));
cron = ctx.get(ICronService);
});
afterEach(async () => {
try {
await ctx.expectResumeMatches();
} finally {
await ctx.dispose();
}
await ctx.dispose();
});
it('does not bind when KIMI_CRON_MANUAL_TICK is unset', () => {

214
packages/agent-core-v2/test/cron/resume.test.ts
View file

@ -16,17 +16,23 @@ import { join } from 'pathe';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { ContentPart } from '@moonshot-ai/kosong';
import type { ContextMessage, PromptOrigin } from '#/index';
import { IPromptService } from '#/index';
import type { ContextMessage, PromptOrigin } from '#/contextMemory';
import { IPromptService } from '#/prompt';
import { ICronService } from '#/cron';
import { createCronPersistStore } from '#/cron/tools/persist';
import type { ClockSources } from '#/cron/tools/clock';
import { createTestAgent, cronServices, type TestAgentContext } from '../harness';
import type { CronTask } from '#/cron/tools/types';
import { IAtomicDocumentStore } from '#/storage';
import {
createTestAgent,
cronServices,
homeDirServices,
type TestAgentContext,
} from '../harness';
const WALL_ANCHOR = 1_700_000_000_000;
interface ClockHarness {
readonly clocks: ClockSources;
install(): void;
setNow(v: number): void;
advance(ms: number): void;
now(): number;
@ -34,19 +40,15 @@ interface ClockHarness {
function createClocks(initial: number = WALL_ANCHOR): ClockHarness {
let wall = initial;
let mono = 1_000_000;
return {
clocks: {
wallNow: () => wall,
monoNowMs: () => mono,
install: () => {
vi.spyOn(Date, 'now').mockImplementation(() => wall);
},
setNow: (v) => {
wall = v;
mono = v;
},
advance: (ms) => {
wall += ms;
mono += ms;
},
now: () => wall,
};
@ -78,6 +80,24 @@ async function readDiskIds(sessionDir: string): Promise<readonly string[]> {
}
}
function createCronAgent(
sessionDir: string,
cronOverride: ReturnType<typeof cronServices>,
): TestAgentContext {
return createTestAgent(homeDirServices(sessionDir), cronOverride);
}
function cronDocuments(ctx: TestAgentContext): IAtomicDocumentStore {
return ctx.get(IAtomicDocumentStore);
}
async function readPersistedTask(
ctx: TestAgentContext,
id: string,
): Promise<CronTask | undefined> {
return cronDocuments(ctx).get<CronTask>('cron', `${id}.json`);
}
describe('CronManager — persistence and resume', () => {
let sessionDir: string;
let ctx: TestAgentContext;
@ -89,6 +109,7 @@ describe('CronManager — persistence and resume', () => {
beforeEach(async () => {
vi.stubEnv('KIMI_CRON_NO_JITTER', '1');
vi.stubEnv('KIMI_CRON_POLL_INTERVAL_MS', '0');
sessionDir = await mkdtemp(join(tmpdir(), 'kimi-cron-resume-'));
resumedCtx = undefined;
resumedCron = undefined;
@ -97,20 +118,14 @@ describe('CronManager — persistence and resume', () => {
afterEach(async () => {
try {
if (resumedCtx !== undefined) {
await resumedCtx.expectResumeMatches();
}
await ctx.expectResumeMatches();
await resumedCtx?.dispose();
} finally {
try {
await resumedCtx?.dispose();
await ctx.dispose();
} finally {
try {
await ctx.dispose();
} finally {
vi.unstubAllEnvs();
await rm(sessionDir, { recursive: true, force: true });
}
vi.unstubAllEnvs();
vi.restoreAllMocks();
await rm(sessionDir, { recursive: true, force: true });
}
}
});
@ -120,13 +135,10 @@ describe('CronManager — persistence and resume', () => {
beforeEach(() => {
harness = createClocks();
ctx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: harness.clocks,
pollIntervalMs: null,
}),
harness.install();
ctx = createCronAgent(
sessionDir,
cronServices({}),
);
cron = ctx.get(ICronService);
});
@ -138,8 +150,7 @@ describe('CronManager — persistence and resume', () => {
});
await cron.flushPersist();
const store = createCronPersistStore(sessionDir);
const loaded = await store.read(task.id);
const loaded = await readPersistedTask(ctx, task.id);
expect(loaded).toEqual({
id: task.id,
cron: '*/5 * * * *',
@ -168,27 +179,22 @@ describe('CronManager — persistence and resume', () => {
beforeEach(() => {
clockA = createClocks();
clockB = createClocks(clockA.now() + 60_000);
ctx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockA.clocks,
pollIntervalMs: null,
}),
clockA.install();
ctx = createCronAgent(
sessionDir,
cronServices({}),
);
cron = ctx.get(ICronService);
resumedCtx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockB.clocks,
pollIntervalMs: null,
}),
clockB.install();
resumedCtx = createCronAgent(
sessionDir,
cronServices({}),
);
resumedCron = resumedCtx.get(ICronService);
});
it('re-adopts tasks with original id and createdAt', async () => {
clockA.install();
const t1 = cron.addTask({ cron: '*/5 * * * *', prompt: 'a' });
const t2 = cron.addTask({
cron: '0 9 * * *',
@ -198,6 +204,7 @@ describe('CronManager — persistence and resume', () => {
await cron.flushPersist();
expect(resumedCron!.store.list()).toEqual([]);
clockB.install();
await resumedCron!.loadFromDisk();
const loaded = resumedCron!.store.list().slice().toSorted((a, b) => a.id.localeCompare(b.id));
@ -215,34 +222,31 @@ describe('CronManager — persistence and resume', () => {
describe('recurring resume fire', () => {
let clockA: ClockHarness;
let clockB: ClockHarness;
beforeEach(() => {
clockA = createClocks();
const clockB = createClocks(clockA.now() + 23 * 60_000);
ctx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockA.clocks,
pollIntervalMs: null,
}),
clockB = createClocks(clockA.now() + 23 * 60_000);
clockA.install();
ctx = createCronAgent(
sessionDir,
cronServices({}),
);
cron = ctx.get(ICronService);
resumedCtx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockB.clocks,
pollIntervalMs: null,
}),
clockB.install();
resumedCtx = createCronAgent(
sessionDir,
cronServices({}),
);
resumedCron = resumedCtx.get(ICronService);
resumedPrompt = resumedCtx.get(IPromptService);
});
it('recurring task missed during downtime fires once with coalescedCount > 1', async () => {
clockA.install();
cron.addTask({ cron: '*/5 * * * *', prompt: 'check' });
await cron.flushPersist();
clockB.install();
await resumedCron!.loadFromDisk();
const steerCalls = captureSteer(resumedPrompt!);
@ -259,32 +263,28 @@ describe('CronManager — persistence and resume', () => {
describe('one-shot resume fire', () => {
let clockA: ClockHarness;
let clockB: ClockHarness;
beforeEach(() => {
clockA = createClocks(WALL_ANCHOR);
const clockB = createClocks(clockA.now() + 10 * 60_000);
ctx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockA.clocks,
pollIntervalMs: null,
}),
clockB = createClocks(clockA.now() + 10 * 60_000);
clockA.install();
ctx = createCronAgent(
sessionDir,
cronServices({}),
);
cron = ctx.get(ICronService);
resumedCtx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockB.clocks,
pollIntervalMs: null,
}),
clockB.install();
resumedCtx = createCronAgent(
sessionDir,
cronServices({}),
);
resumedCron = resumedCtx.get(ICronService);
resumedPrompt = resumedCtx.get(IPromptService);
});
it('one-shot scheduled in the past fires once on resume and the file is removed', async () => {
clockA.install();
const oneShot = cron.addTask({
cron: '*/5 * * * *',
prompt: 'remind once',
@ -292,6 +292,7 @@ describe('CronManager — persistence and resume', () => {
});
await cron.flushPersist();
expect(await readDiskIds(sessionDir)).toEqual([oneShot.id]);
clockB.install();
await resumedCron!.loadFromDisk();
const steerCalls = captureSteer(resumedPrompt!);
@ -311,33 +312,29 @@ describe('CronManager — persistence and resume', () => {
describe('recurring task already fired before shutdown', () => {
let clockA: ClockHarness;
let clockB: ClockHarness;
beforeEach(() => {
clockA = createClocks(WALL_ANCHOR);
const clockB = createClocks(WALL_ANCHOR + 23 * 60_000);
ctx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockA.clocks,
pollIntervalMs: null,
}),
clockB = createClocks(WALL_ANCHOR + 23 * 60_000);
clockA.install();
ctx = createCronAgent(
sessionDir,
cronServices({}),
);
cron = ctx.get(ICronService);
prompt = ctx.get(IPromptService);
resumedCtx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockB.clocks,
pollIntervalMs: null,
}),
clockB.install();
resumedCtx = createCronAgent(
sessionDir,
cronServices({}),
);
resumedCron = resumedCtx.get(ICronService);
resumedPrompt = resumedCtx.get(IPromptService);
});
it('does NOT replay on resume', async () => {
clockA.install();
const task = cron.addTask({ cron: '*/5 * * * *', prompt: 'check' });
await cron.flushPersist();
@ -348,10 +345,11 @@ describe('CronManager — persistence and resume', () => {
await cron.flushPersist();
const onDisk = await createCronPersistStore(sessionDir).read(task.id);
const onDisk = await readPersistedTask(ctx, task.id);
expect(typeof onDisk?.lastFiredAt).toBe('number');
expect(onDisk!.lastFiredAt!).toBeLessThanOrEqual(clockA.now());
clockB.install();
await resumedCron!.loadFromDisk();
const steerCallsB = captureSteer(resumedPrompt!);
@ -367,43 +365,40 @@ describe('CronManager — persistence and resume', () => {
describe('corrupt lastFiredAt', () => {
let clockA: ClockHarness;
let clockB: ClockHarness;
beforeEach(() => {
clockA = createClocks();
const clockB = createClocks(clockA.now() + 23 * 60_000);
ctx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockA.clocks,
pollIntervalMs: null,
}),
clockB = createClocks(clockA.now() + 23 * 60_000);
clockA.install();
ctx = createCronAgent(
sessionDir,
cronServices({}),
);
cron = ctx.get(ICronService);
resumedCtx = createTestAgent(
cronServices({
homedir: sessionDir,
autoStart: false,
clocks: clockB.clocks,
pollIntervalMs: null,
}),
clockB.install();
resumedCtx = createCronAgent(
sessionDir,
cronServices({}),
);
resumedCron = resumedCtx.get(ICronService);
resumedPrompt = resumedCtx.get(IPromptService);
});
it('treats a future lastFiredAt as corrupt and falls back to createdAt', async () => {
clockA.install();
const task = cron.addTask({ cron: '*/5 * * * *', prompt: 'check' });
await cron.flushPersist();
const store = createCronPersistStore(sessionDir);
const original = await store.read(task.id);
const store = createCronPersistStore(cronDocuments(ctx));
const original = await readPersistedTask(ctx, task.id);
if (original === undefined) throw new Error('expected persisted task');
await store.write(task.id, {
...original,
lastFiredAt: clockA.now() + 365 * 24 * 60 * 60 * 1000,
});
clockB.install();
await resumedCron!.loadFromDisk();
const steerCalls = captureSteer(resumedPrompt!);
@ -419,8 +414,9 @@ describe('CronManager — persistence and resume', () => {
describe('in-memory mode', () => {
beforeEach(() => {
const harness = createClocks();
harness.install();
ctx = createTestAgent(
cronServices({ autoStart: false, clocks: harness.clocks, pollIntervalMs: null }),
cronServices({}),
);
cron = ctx.get(ICronService);
});

4
packages/agent-core-v2/test/toolExecutor/tool-executor.test.ts
View file

@ -5,9 +5,9 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { DisposableStore } from '#/_base/di/lifecycle';
import { createServices, type TestInstantiationService } from '#/_base/di/test';
import type { LoopEvent } from '#/loop';
import { ToolAccesses, type ExecutableTool, type ExecutableToolContext, type ExecutableToolResult, type ToolExecution, type ToolUpdate } from '#/tool';
import { ToolAccesses, type ExecutableTool, type ExecutableToolContext, type ExecutableToolResult, type ToolExecution, type ToolResult, type ToolUpdate } from '#/tool';
import { IToolExecutor, ToolExecutorService } from '#/toolExecutor';
import { IToolRegistry, ToolRegistryService, type ToolResult } from '#/toolRegistry';
import { IToolRegistry, ToolRegistryService } from '#/toolRegistry';
let disposables: DisposableStore;
let ix: TestInstantiationService;