diff --git a/packages/agent-core-v2/docs/state-manifest.d.ts b/packages/agent-core-v2/docs/state-manifest.d.ts index 81d2a4674..bd4c2ea4a 100644 --- a/packages/agent-core-v2/docs/state-manifest.d.ts +++ b/packages/agent-core-v2/docs/state-manifest.d.ts @@ -27,7 +27,7 @@ // references become '(circular)', and class instances collapse to a '(ClassName)' // marker — the wire shape of an entry is the JSON projection of the type here. // -// Index (App: 0 keys · Workspace: 6 keys · Session: 18 keys · Agent: 98 keys) +// Index (App: 0 keys · Workspace: 6 keys · Session: 17 keys · Agent: 98 keys) // App // Workspace // workspaceDirs.ephemeralDirs src/workspace/workspaceDirs/workspaceDirsService.ts @@ -42,7 +42,6 @@ // cron.parsedCache src/session/cron/sessionCronServiceImpl.ts // cron.seededFromStore src/session/cron/sessionCronServiceImpl.ts // cron.started src/session/cron/sessionCronServiceImpl.ts -// cron.tasks src/session/cron/sessionCronServiceImpl.ts // interaction.nextId src/session/interaction/interactionService.ts // interaction.pending src/session/interaction/interactionService.ts // interaction.recentlyResolved src/session/interaction/interactionService.ts @@ -443,15 +442,6 @@ export interface SessionStateSnapshot { }>; 'cron.seededFromStore': Set; 'cron.started': boolean; - 'cron.tasks': Map>; - }>; // src/session/interaction/interactionService.ts 'interaction.nextId': number; 'interaction.pending': Map; + }; +} + +/** + * states: cron + * owner: src/session/cron/cronOps.ts + */ +interface CronCursorPayload { + _name: 'cron.cursor'; + id: string; + lastFiredAt: number; +} + +/** + * states: cron + * owner: src/session/cron/cronOps.ts + */ +interface CronDeletePayload { + _name: 'cron.delete'; + ids: string[]; +} + /** * states: goal, goalForkNotice * owner: src/agent/goal/goalOps.ts @@ -715,6 +754,9 @@ interface WirePayloadMap { "context.apply_compaction": ContextApplyCompactionPayload; "context.clear": ContextClearPayload; "context.undo": ContextUndoPayload; + "cron.add": CronAddPayload; + "cron.cursor": CronCursorPayload; + "cron.delete": CronDeletePayload; "forked": ForkedPayload; "full_compaction.begin": FullCompactionBeginPayload; "full_compaction.cancel": FullCompactionCancelPayload; diff --git a/packages/agent-core-v2/src/app/bootstrap/bootstrap.ts b/packages/agent-core-v2/src/app/bootstrap/bootstrap.ts index d44387689..7513c9bd9 100644 --- a/packages/agent-core-v2/src/app/bootstrap/bootstrap.ts +++ b/packages/agent-core-v2/src/app/bootstrap/bootstrap.ts @@ -63,8 +63,7 @@ export type PersistenceScopeName = | 'store' | 'logs' | 'cache' - | 'credentials' - | 'cron'; + | 'credentials'; export interface IBootstrapService { readonly _serviceBrand: undefined; diff --git a/packages/agent-core-v2/src/app/bootstrap/bootstrapService.ts b/packages/agent-core-v2/src/app/bootstrap/bootstrapService.ts index 0029ca3ec..440a087cb 100644 --- a/packages/agent-core-v2/src/app/bootstrap/bootstrapService.ts +++ b/packages/agent-core-v2/src/app/bootstrap/bootstrapService.ts @@ -56,7 +56,6 @@ export class BootstrapService implements IBootstrapService { logs: relative(options.homeDir, this.logsDir), cache: relative(options.homeDir, this.cacheDir), credentials: 'credentials', - cron: 'cron', }; } diff --git a/packages/agent-core-v2/src/app/cron/cronTask.ts b/packages/agent-core-v2/src/app/cron/cronTask.ts index 0e177ae0b..134fa66e8 100644 --- a/packages/agent-core-v2/src/app/cron/cronTask.ts +++ b/packages/agent-core-v2/src/app/cron/cronTask.ts @@ -9,5 +9,3 @@ export interface CronTask { } export type CronTaskInit = Omit; - -export const CRON_SESSION_TAG = 'sessionId'; diff --git a/packages/agent-core-v2/src/app/cron/cronTaskPersistence.ts b/packages/agent-core-v2/src/app/cron/cronTaskPersistence.ts deleted file mode 100644 index 5002012df..000000000 --- a/packages/agent-core-v2/src/app/cron/cronTaskPersistence.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { createDecorator } from '#/_base/di/instantiation'; - -import type { CronTask } from './cronTask'; - -export interface CronTaskQuery { - readonly workspaceId: string; -} - -export interface ICronTaskPersistence { - readonly _serviceBrand: undefined; - - get(workspaceId: string, taskId: string): Promise; - list(query: CronTaskQuery): Promise; - save(workspaceId: string, task: CronTask): Promise; - delete(workspaceId: string, taskId: string): Promise; -} - -export const ICronTaskPersistence = createDecorator('cronTaskPersistence'); diff --git a/packages/agent-core-v2/src/app/cron/cronTaskPersistenceService.ts b/packages/agent-core-v2/src/app/cron/cronTaskPersistenceService.ts deleted file mode 100644 index 19d91ebab..000000000 --- a/packages/agent-core-v2/src/app/cron/cronTaskPersistenceService.ts +++ /dev/null @@ -1,92 +0,0 @@ -import { Disposable } from '#/_base/di/lifecycle'; -import { LifecycleScope } from '#/app/scopes'; -import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; -import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; -import { IBootstrapService } from '#/app/bootstrap/bootstrap'; - -import { ICronTaskPersistence, type CronTaskQuery } from './cronTaskPersistence'; -import type { CronTask } from './cronTask'; - -export const CRON_ID_REGEX: RegExp = /^(?:[0-9a-f]{8}|[0-9A-HJKMNP-TV-Z]{26})$/i; -const JSON_SUFFIX = '.json'; - -export function isValidCronTask(obj: unknown): obj is CronTask { - if (typeof obj !== 'object' || obj === null) return false; - const o = obj as Record; - if (typeof o['id'] !== 'string' || !CRON_ID_REGEX.test(o['id'])) return false; - if (typeof o['cron'] !== 'string') return false; - if (typeof o['prompt'] !== 'string') return false; - if (typeof o['createdAt'] !== 'number') return false; - if (o['recurring'] !== undefined && typeof o['recurring'] !== 'boolean') return false; - if ( - o['lastFiredAt'] !== undefined && - (typeof o['lastFiredAt'] !== 'number' || !Number.isFinite(o['lastFiredAt'])) - ) { - return false; - } - if (o['tags'] !== undefined) { - if (typeof o['tags'] !== 'object' || o['tags'] === null) return false; - for (const v of Object.values(o['tags'] as Record)) { - if (typeof v !== 'string') return false; - } - } - return true; -} - -export class CronTaskPersistenceService extends Disposable implements ICronTaskPersistence { - declare readonly _serviceBrand: undefined; - - private readonly cronScope: string; - - constructor( - @IBootstrapService private readonly bootstrap: IBootstrapService, - @IAtomicDocumentStore private readonly atomicDocs: IAtomicDocumentStore, - ) { - super(); - this.cronScope = this.bootstrap.scope('cron'); - } - - private workspaceScope(workspaceId: string): string { - return `${this.cronScope}/${workspaceId}`; - } - - async get(workspaceId: string, taskId: string): Promise { - const scope = this.workspaceScope(workspaceId); - const value = await this.atomicDocs.get(scope, `${taskId}${JSON_SUFFIX}`); - if (value === undefined || !isValidCronTask(value)) return undefined; - return value; - } - - async list(query: CronTaskQuery): Promise { - const scope = this.workspaceScope(query.workspaceId); - const keys = await this.atomicDocs.list(scope); - const tasks: CronTask[] = []; - for (const key of keys) { - if (!key.endsWith(JSON_SUFFIX)) continue; - const id = key.slice(0, -JSON_SUFFIX.length); - if (!CRON_ID_REGEX.test(id)) continue; - const value = await this.atomicDocs.get(scope, key); - if (value === undefined || !isValidCronTask(value)) continue; - tasks.push(value); - } - return tasks; - } - - async save(workspaceId: string, task: CronTask): Promise { - const scope = this.workspaceScope(workspaceId); - await this.atomicDocs.set(scope, `${task.id}${JSON_SUFFIX}`, task); - } - - async delete(workspaceId: string, taskId: string): Promise { - const scope = this.workspaceScope(workspaceId); - await this.atomicDocs.delete(scope, `${taskId}${JSON_SUFFIX}`); - } -} - -registerScopedService( - LifecycleScope.App, - ICronTaskPersistence, - CronTaskPersistenceService, - ScopeActivation.OnScopeCreated, - 'cron', -); diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index 31db38ea9..1d3f9823a 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -413,8 +413,6 @@ export * from '#/agent/task/taskOps'; export * from '#/agent/task/taskService'; import '#/app/cron/configSection'; export * from '#/app/cron/cronTask'; -export * from '#/app/cron/cronTaskPersistence'; -export * from '#/app/cron/cronTaskPersistenceService'; export * from '#/app/cron/cron-expr'; export * from '#/app/cron/format'; export * from '#/app/cron/jitter'; diff --git a/packages/agent-core-v2/src/session/cron/cronOps.ts b/packages/agent-core-v2/src/session/cron/cronOps.ts index 3d5b3f06d..12ab2ec27 100644 --- a/packages/agent-core-v2/src/session/cron/cronOps.ts +++ b/packages/agent-core-v2/src/session/cron/cronOps.ts @@ -8,12 +8,28 @@ import { defineState } from '#/state/state'; export type CronModelState = Map; +const cronTaskSchema = z.object({ + id: z.string(), + cron: z.string(), + prompt: z.string(), + createdAt: z.number(), + recurring: z.boolean().optional(), + lastFiredAt: z.number().optional(), + tags: z.record(z.string(), z.string()).optional(), +}); + +const cronAddSchema = z.object({ task: cronTaskSchema }); +const cronDeleteSchema = z.object({ ids: z.array(z.string()) }); +const cronCursorSchema = z.object({ id: z.string(), lastFiredAt: z.number() }); + export interface CronAddPayload { readonly task: CronTask; } export class CronAdd extends Event2 { static override readonly type = 'cron.add'; + static override readonly durable = true; + static override readonly schema = cronAddSchema; } export interface CronAdd extends CronAddPayload {} @@ -23,6 +39,8 @@ export interface CronDeletePayload { export class CronDelete extends Event2 { static override readonly type = 'cron.delete'; + static override readonly durable = true; + static override readonly schema = cronDeleteSchema; } export interface CronDelete extends CronDeletePayload {} @@ -33,6 +51,8 @@ export interface CronCursorPayload { export class CronCursor extends Event2 { static override readonly type = 'cron.cursor'; + static override readonly durable = true; + static override readonly schema = cronCursorSchema; } export interface CronCursor extends CronCursorPayload {} @@ -49,7 +69,6 @@ export interface CronFired extends CronFiredPayload {} export const cronKey = defineState('cron', (): CronModelState => new Map()).replayable({ schema: z.custom(), - durable: false, }) .on(CronAdd, (s, e) => { s.set(e.task.id, e.task); diff --git a/packages/agent-core-v2/src/session/cron/sessionCronService.ts b/packages/agent-core-v2/src/session/cron/sessionCronService.ts index 2417bc2f8..a6b914349 100644 --- a/packages/agent-core-v2/src/session/cron/sessionCronService.ts +++ b/packages/agent-core-v2/src/session/cron/sessionCronService.ts @@ -5,10 +5,6 @@ import type { Turn } from '#/agent/loop/loop'; import type { CronTask, CronTaskInit } from '#/app/cron/cronTask'; import type { ParsedCronExpression } from '#/app/cron/cron-expr'; -export interface CronLoadOptions { - readonly replace?: boolean; -} - export interface ISessionCronService { readonly _serviceBrand: undefined; @@ -27,11 +23,9 @@ export interface ISessionCronService { parsed: ParsedCronExpression, idealMs: number, ): number | null; - loadFromStore(options?: CronLoadOptions): Promise; start(): Promise; stop(): Promise; tick(): Promise; - flushPersist(): Promise; handleMissed( tasks: readonly CronTask[], renderMissedNotification: (tasks: readonly CronTask[]) => readonly ContentPart[], diff --git a/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts b/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts index bf30f4d5e..41b7a3558 100644 --- a/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts +++ b/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts @@ -15,11 +15,9 @@ import { ITelemetryService } from '#/app/telemetry/telemetry'; import { type ClockSources, resolveClockSources, SYSTEM_CLOCKS } from '#/app/cron/clock'; import { type CronConfig, CRON_SECTION } from '#/app/cron/configSection'; import { computeNextCronRun, parseCronExpression, type ParsedCronExpression } from '#/app/cron/cron-expr'; -import { CRON_SESSION_TAG, type CronTask, type CronTaskInit } from '#/app/cron/cronTask'; -import { ICronTaskPersistence } from '#/app/cron/cronTaskPersistence'; +import { type CronTask, type CronTaskInit } from '#/app/cron/cronTask'; import { renderCronFireXml } from '#/app/cron/format'; import { jitteredNextCronRunMs, oneShotJitteredNextCronRunMs } from '#/app/cron/jitter'; -import { ISessionContext } from '#/session/sessionContext/sessionContext'; import { ISessionStateService } from '#/session/state/sessionState'; import { IAgentLifecycleService } from '#/session/agentLifecycle/agentLifecycle'; import type { ContextMessage } from '#/agent/contextMemory/types'; @@ -35,14 +33,13 @@ import { ICronListTool } from '#/agent/tools/cron/cron-list/cron-list'; import { ICronDeleteTool } from '#/agent/tools/cron/cron-delete/cron-delete'; import { CronAdd, CronDelete, CronCursor, CronFired, cronKey } from './cronOps'; -import { ISessionCronService, type CronLoadOptions } from './sessionCronService'; +import { ISessionCronService } from './sessionCronService'; export const CRON_SCHEDULED = 'cron_scheduled' as const; export const CRON_FIRED = 'cron_fired' as const; export const CRON_MISSED = 'cron_missed' as const; export const CRON_DELETED = 'cron_deleted' as const; -export const cronTasksKey = defineState>('cron.tasks', () => new Map()); export const cronParsedCacheKey = defineState>( 'cron.parsedCache', () => new Map(), @@ -62,7 +59,6 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe declare readonly _serviceBrand: undefined; private readonly timer = this._register(new IntervalTimer({ unref: true })); - private readonly persistQueues = new Map>(); private clocks: ClockSources = SYSTEM_CLOCKS; readonly isEnabled: boolean = true; @@ -71,15 +67,12 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe constructor( @ISessionStateService private readonly states: ISessionStateService, - @ISessionContext private readonly ctx: ISessionContext, - @ICronTaskPersistence private readonly store: ICronTaskPersistence, @IAgentLifecycleService private readonly agentLifecycle: IAgentLifecycleService, @ITelemetryService private readonly telemetry: ITelemetryService, @IConfigService private readonly config: IConfigService, ) { super(); - this.states.contributeState(cronTasksKey); this.states.contributeState(cronParsedCacheKey); this.states.contributeState(cronLastSeenAtKey); this.states.contributeState(cronSeededFromStoreKey); @@ -109,8 +102,10 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe ); } - private get tasks(): Map { - return this.states.get(cronTasksKey); + private get tasks(): ReadonlyMap { + const main = this.agentLifecycle.get('main'); + if (main === undefined) return new Map(); + return main.accessor.get(IAgentStateService).get(cronKey); } private get parsedCache(): Map { @@ -139,16 +134,10 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe private bindMainAgent(handle: IAgentScopeHandle): void { const dispatcher = handle.accessor.get(IEventDispatcher); - const agentState = handle.accessor.get(IAgentStateService); this._register( dispatcher.hooks.onDidRestore.register('cron', async (_ctx, next) => { await this.config.ready; this.resolveClocks(); - this.tasks.clear(); - for (const [id, task] of agentState.get(cronKey)) { - this.tasks.set(id, task as CronTask); - } - await this.loadFromStore({ replace: false }); await this.start(); await next(); }), @@ -191,26 +180,15 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe ...init, id: this.generateUniqueId(), createdAt: this.clocks.wallNow(), - tags: { ...init.tags, [CRON_SESSION_TAG]: this.ctx.sessionId }, }; - this.tasks.set(task.id, task); this.dispatchCron(new CronAdd({ task })); - this.persistEnqueue(task.id, () => - this.store.save(this.ctx.workspaceId, task), - ); return task; } removeTasks(ids: readonly string[]): readonly string[] { - const removed = this.removeByIds(ids); + const removed = ids.filter((id) => this.tasks.has(id)); if (removed.length === 0) return removed; - this.dispatchCron(new CronDelete({ ids: removed })); - for (const id of removed) { - this.persistEnqueue(id, () => - this.store.delete(this.ctx.workspaceId, id), - ); - } return removed; } @@ -243,29 +221,6 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe return this.nextFireFor(task); } - async loadFromStore(options: CronLoadOptions = {}): Promise { - if (options.replace !== false) { - this.tasks.clear(); - } - const allTasks = await this.store.list({ workspaceId: this.ctx.workspaceId }); - for (const task of allTasks) { - const owner = task.tags?.[CRON_SESSION_TAG]; - if (owner !== undefined && owner !== this.ctx.sessionId) continue; - if (owner === undefined) { - const claimed: CronTask = { - ...task, - tags: { ...task.tags, [CRON_SESSION_TAG]: this.ctx.sessionId }, - }; - this.adopt(claimed); - this.persistEnqueue(claimed.id, () => - this.store.save(this.ctx.workspaceId, claimed), - ); - continue; - } - this.adopt(task); - } - } - async start(): Promise { if (this.started) return; this.started = true; @@ -287,7 +242,6 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe this.lastSeenAt.clear(); this.seededFromStore.clear(); this.parsedCache.clear(); - await this.flushPersist(); this.started = false; } @@ -380,11 +334,6 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe } } - async flushPersist(): Promise { - const inFlight = Array.from(this.persistQueues.values()); - await Promise.allSettled(inFlight); - } - handleMissed( tasks: readonly CronTask[], renderMissedNotification: (tasks: readonly CronTask[]) => readonly ContentPart[], @@ -500,13 +449,8 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe } private advanceCursor(id: string, lastFiredAt: number): void { - const updated = this.markFired(id, lastFiredAt); - if (updated === undefined) return; - + if (!this.tasks.has(id)) return; this.dispatchCron(new CronCursor({ id, lastFiredAt })); - this.persistEnqueue(id, () => - this.store.save(this.ctx.workspaceId, updated), - ); } private dispatchCron(event: CronAdd | CronDelete | CronCursor): void { @@ -614,28 +558,6 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe } } - private adopt(task: CronTask): void { - this.tasks.set(task.id, task); - } - - private markFired(id: string, lastFiredAt: number): CronTask | undefined { - const existing = this.tasks.get(id); - if (existing === undefined) return undefined; - const updated: CronTask = { ...existing, lastFiredAt }; - this.tasks.set(id, updated); - return updated; - } - - private removeByIds(ids: readonly string[]): readonly string[] { - const removed: string[] = []; - for (const id of ids) { - if (this.tasks.delete(id)) { - removed.push(id); - } - } - return removed; - } - private generateUniqueId(): string { for (let attempt = 0; attempt < MAX_ID_ATTEMPTS; attempt++) { const candidate = ulid(); @@ -654,20 +576,6 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe return Number.isFinite(age) && age >= STALE_THRESHOLD_MS; } - private persistEnqueue(id: string, work: () => Promise): void { - const prev = this.persistQueues.get(id) ?? Promise.resolve(); - const next = prev - .catch(() => {}) - .then(() => work()) - .catch(() => {}) - .finally(() => { - if (this.persistQueues.get(id) === next) { - this.persistQueues.delete(id); - } - }); - this.persistQueues.set(id, next); - } - private bindSigusr1(): void { if (process.platform === 'win32') return; if (!this.getCronConfig().manualTick) return; diff --git a/packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts b/packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts index be88aa77f..229facc06 100644 --- a/packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts +++ b/packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts @@ -1,7 +1,6 @@ import { randomUUID } from 'node:crypto'; import { join } from 'pathe'; -import { ulid } from 'ulid'; import type { IInstantiationService } from '#/_base/di/instantiation'; import { Disposable } from '#/_base/di/lifecycle'; @@ -15,8 +14,6 @@ import { DEFAULT_PLAN_MODE_SECTION } from '#/features/plan/configSection'; import { IAgentPlanService } from '#/features/plan/plan'; import { LifecycleScope } from '#/app/scopes'; import { IBootstrapService } from '#/app/bootstrap/bootstrap'; -import { CRON_SESSION_TAG, type CronTask } from '#/app/cron/cronTask'; -import { ICronTaskPersistence } from '#/app/cron/cronTaskPersistence'; import { IConfigService } from '#/app/config/config'; import { IEventService } from '#/app/event/event'; import { @@ -145,7 +142,6 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec @IAppendLogStore private readonly appendLogStore: IAppendLogStore, @IAtomicDocumentStore private readonly docs: IAtomicDocumentStore, @IHostFileSystem private readonly hostFs: IHostFileSystem, - @ICronTaskPersistence private readonly cronStore: ICronTaskPersistence, @IEventService private readonly event: IEventService, @ITelemetryService private readonly telemetry: ITelemetryService, @IWorkspaceAgentProfileLoader @@ -559,10 +555,6 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec custom: forkCustomMetadata(sourceMeta?.custom, opts.metadata), }); - if (turnSlice === undefined) { - await this.duplicateCronTasks(sourceId, targetId); - } - await this.appendSessionIndexEntry(targetId, this.workspaceContext.cwd); this._onDidForkSession.fire({ sourceSessionId: sourceId, @@ -719,19 +711,6 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec } } - private async duplicateCronTasks(sourceId: string, targetId: string): Promise { - const tasks = await this.cronStore.list({ workspaceId: this.workspaceId }); - for (const task of tasks) { - if (task.tags?.[CRON_SESSION_TAG] !== sourceId) continue; - const clone: CronTask = { - ...task, - id: ulid(), - tags: { ...task.tags, [CRON_SESSION_TAG]: targetId }, - }; - await this.cronStore.save(this.workspaceId, clone); - } - } - private async readMetaFromDisk(sessionId: string): Promise { return this.docs.get(sessionScopeOf(this.handlerScope, sessionId), 'state.json'); } diff --git a/packages/agent-core-v2/src/workspace/workspaceInstance/workspaceInstanceManagerService.ts b/packages/agent-core-v2/src/workspace/workspaceInstance/workspaceInstanceManagerService.ts index ba2197622..e9d6617b9 100644 --- a/packages/agent-core-v2/src/workspace/workspaceInstance/workspaceInstanceManagerService.ts +++ b/packages/agent-core-v2/src/workspace/workspaceInstance/workspaceInstanceManagerService.ts @@ -7,7 +7,6 @@ import { IBuiltinAgentProfileLoader } from '#/app/agentProfileCatalog/builtinAge import { IAgentProfileRegistry } from '#/app/agentProfileCatalog/agentProfileRegistry'; import { IBootstrapService } from '#/app/bootstrap/bootstrap'; import { IConfigService } from '#/app/config/config'; -import { ICronTaskPersistence } from '#/app/cron/cronTaskPersistence'; import { IEventService } from '#/app/event/event'; import { IFlagService } from '#/app/flag/flag'; import { IGitService } from '#/app/git/git'; @@ -54,7 +53,6 @@ export class WorkspaceInstanceManager implements IWorkspaceInstanceManager { @IHostEnvironment private readonly environment: IHostEnvironment, @IAppStateService private readonly appState: IAppStateService, @IConfigService private readonly config: IConfigService, - @ICronTaskPersistence private readonly cronStore: ICronTaskPersistence, @IEventService private readonly event: IEventService, @IFlagService private readonly flags: IFlagService, @ref(IGitService) private readonly git: LiveRef, @@ -209,7 +207,6 @@ export class WorkspaceInstanceManager implements IWorkspaceInstanceManager { this.appendLogStore, this.docs, input.fs, - this.cronStore, this.event, this.telemetry, input.workspaceAgentProfiles, diff --git a/packages/agent-core-v2/test/app/bootstrap/stubs.ts b/packages/agent-core-v2/test/app/bootstrap/stubs.ts index 5236d8f96..773a5d7b0 100644 --- a/packages/agent-core-v2/test/app/bootstrap/stubs.ts +++ b/packages/agent-core-v2/test/app/bootstrap/stubs.ts @@ -25,7 +25,6 @@ export function stubBootstrap( logs: 'logs', cache: 'cache', credentials: 'credentials', - cron: 'cron', }; return { _serviceBrand: undefined, diff --git a/packages/agent-core-v2/test/harness/agent.ts b/packages/agent-core-v2/test/harness/agent.ts index b277d82b8..11e6f6e68 100644 --- a/packages/agent-core-v2/test/harness/agent.ts +++ b/packages/agent-core-v2/test/harness/agent.ts @@ -24,8 +24,6 @@ import type { ContextMessage } from '#/agent/contextMemory/types'; import { ISessionCronService } from '#/session/cron/sessionCronService'; import { SessionCronServiceImpl } from '#/session/cron/sessionCronServiceImpl'; import { IAgentIdentity } from '#/app/agentIdentity/agentIdentity'; -import { ICronTaskPersistence } from '#/app/cron/cronTaskPersistence'; -import { CronTaskPersistenceService } from '#/app/cron/cronTaskPersistenceService'; import { IAgentGoalService } from '#/agent/goal/goal'; import { AgentGoalService } from '#/agent/goal/goalService'; import { ISessionMcpHandle } from '#/session/mcp/sessionMcpHandle'; @@ -1176,7 +1174,6 @@ export class AgentTestContext { ready: Promise.resolve(), } satisfies IHostEnvironment, ); - reg.defineDescriptor(ICronTaskPersistence, new SyncDescriptor(CronTaskPersistenceService)); }, ], this.serviceOverrides, diff --git a/packages/agent-core-v2/test/index.test.ts b/packages/agent-core-v2/test/index.test.ts index 60bd0dc61..75faa6fbd 100644 --- a/packages/agent-core-v2/test/index.test.ts +++ b/packages/agent-core-v2/test/index.test.ts @@ -94,6 +94,9 @@ const V2_RECORD_TYPES: ReadonlySet = new Set([ 'token_counting.measured', 'token_counting.truncated', 'token_counting.rebased', + 'cron.add', + 'cron.delete', + 'cron.cursor', ]); describe('v1 wire vocabulary', () => { diff --git a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts index d204da09a..21e777b11 100644 --- a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts +++ b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts @@ -47,7 +47,6 @@ import '#/state/eventDispatcherService'; import { IAgentTaskService } from '#/agent/task/task'; import { ISessionCronService } from '#/session/cron/sessionCronService'; import { SessionCronServiceImpl } from '#/session/cron/sessionCronServiceImpl'; -import { ICronTaskPersistence } from '#/app/cron/cronTaskPersistence'; import { CRON_SECTION } from '#/app/cron/configSection'; import { ISessionInteractionService } from '#/session/interaction/interaction'; import { SessionInteractionService } from '#/session/interaction/interactionService'; @@ -716,13 +715,6 @@ describe('AgentLifecycleService', () => { }, { type: 'interaction.request', id: 'i1', kind: 'question', request: { q: 1 }, time: 3 }, ]).store); - ix.stub(ICronTaskPersistence, { - _serviceBrand: undefined, - get: async () => undefined, - list: async () => [], - save: async () => {}, - delete: async () => {}, - } as ICronTaskPersistence); ix.stub(IConfigService, { ready: Promise.resolve(), get: ((section: unknown) => diff --git a/packages/agent-core-v2/test/session/cron/cron-tools.test.ts b/packages/agent-core-v2/test/session/cron/cron-tools.test.ts index db85bb577..670d512b2 100644 --- a/packages/agent-core-v2/test/session/cron/cron-tools.test.ts +++ b/packages/agent-core-v2/test/session/cron/cron-tools.test.ts @@ -127,11 +127,9 @@ function createToolHarness(options: { deleted.push(id); deletedAgentIds.push(agentId); }, - loadFromStore: async () => {}, start: () => Promise.resolve(), stop: async () => {}, tick: () => Promise.resolve(), - flushPersist: async () => {}, handleMissed: () => undefined, }; diff --git a/packages/agent-core-v2/test/session/cron/sessionCron.test.ts b/packages/agent-core-v2/test/session/cron/sessionCron.test.ts new file mode 100644 index 000000000..07777f6c4 --- /dev/null +++ b/packages/agent-core-v2/test/session/cron/sessionCron.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from 'vitest'; + +import { Emitter, Event } from '#/_base/event'; +import type { ServiceIdentifier } from '#/_base/di/instantiation'; +import { LifecycleScope } from '#/app/scopes'; +import { type IAgentScopeHandle } from '#/_base/di/scope'; +import { IAgentLifecycleService } from '#/session/agentLifecycle/agentLifecycle'; +import { CronCursor } from '#/session/cron/cronOps'; +import { ISessionCronService } from '#/session/cron/sessionCronService'; + +import { + createTestAgent, + InMemoryWireRecordPersistence, + sessionService, + type TestAgentContext, + type TestAgentOptions, +} from '../../harness'; + +interface CronHarness { + readonly ctx: TestAgentContext; + readonly onDidCreate: Emitter; +} + +async function bootCronContext(options: TestAgentOptions = {}): Promise { + const onDidCreate = new Emitter(); + let mainHandle: IAgentScopeHandle | undefined; + const lifecycleStub: IAgentLifecycleService = { + _serviceBrand: undefined, + onDidCreate: onDidCreate.event, + onDidDispose: Event.None as Event, + create: () => Promise.reject(new Error('not supported in this test')), + fork: () => Promise.reject(new Error('not supported in this test')), + get: (agentId) => (agentId === 'main' ? mainHandle : undefined), + list: () => (mainHandle === undefined ? [] : [mainHandle]), + broadcastPermissionMode: () => {}, + remove: () => Promise.resolve(), + }; + const ctx = createTestAgent(options, sessionService(IAgentLifecycleService, lifecycleStub)); + ctx.kimiConfig = { + ...ctx.kimiConfig, + cron: { debug: false, noJitter: true, noStale: false, disabled: false, manualTick: true }, + }; + const accessor = { + get: (id: ServiceIdentifier): T => ctx.get(id), + }; + mainHandle = { id: 'main', kind: LifecycleScope.Agent, accessor, dispose: () => {} }; + onDidCreate.fire(mainHandle); + return { ctx, onDidCreate }; +} + +describe('session cron wire persistence', () => { + it('writes cron ops as durable wire records and rebuilds the task table on replay', async () => { + const persistence = new InMemoryWireRecordPersistence(); + const first = await bootCronContext({ persistence }); + try { + await first.ctx.restorePersisted(); + + const cron = first.ctx.get(ISessionCronService); + const task = cron.addTask({ cron: '0 9 * * *', prompt: 'wire me', recurring: true }); + await first.ctx.dispatcher.dispatch(new CronCursor({ id: task.id, lastFiredAt: 1234 })); + await first.ctx.dispatcher.flush(); + + const types = persistence.records.map((record) => record.type); + expect(types).toContain('cron.add'); + expect(types).toContain('cron.cursor'); + } finally { + await first.ctx.dispose(); + first.onDidCreate.dispose(); + } + + const second = await bootCronContext({ + persistence: new InMemoryWireRecordPersistence(persistence.records), + }); + try { + await second.ctx.restorePersisted(); + + const resumed = second.ctx.get(ISessionCronService); + const rebuilt = resumed.list(); + expect(rebuilt).toHaveLength(1); + expect(rebuilt[0]).toMatchObject({ + cron: '0 9 * * *', + prompt: 'wire me', + recurring: true, + lastFiredAt: 1234, + }); + } finally { + await second.ctx.dispose(); + second.onDidCreate.dispose(); + } + }); + + it('drops deleted tasks on replay', async () => { + const persistence = new InMemoryWireRecordPersistence(); + const first = await bootCronContext({ persistence }); + try { + await first.ctx.restorePersisted(); + + const cron = first.ctx.get(ISessionCronService); + const kept = cron.addTask({ cron: '0 9 * * *', prompt: 'keep', recurring: true }); + const dropped = cron.addTask({ cron: '0 10 * * *', prompt: 'drop', recurring: true }); + cron.removeTasks([dropped.id]); + await first.ctx.dispatcher.flush(); + + const types = persistence.records.map((record) => record.type); + expect(types).toContain('cron.delete'); + expect(kept.id).not.toBe(dropped.id); + } finally { + await first.ctx.dispose(); + first.onDidCreate.dispose(); + } + + const second = await bootCronContext({ + persistence: new InMemoryWireRecordPersistence(persistence.records), + }); + try { + await second.ctx.restorePersisted(); + + const resumed = second.ctx.get(ISessionCronService); + expect(resumed.list().map((task) => task.prompt)).toEqual(['keep']); + } finally { + await second.ctx.dispose(); + second.onDidCreate.dispose(); + } + }); +}); diff --git a/packages/agent-core-v2/test/workspace/sessionLifecycle/forkTurnSlice.test.ts b/packages/agent-core-v2/test/workspace/sessionLifecycle/forkTurnSlice.test.ts new file mode 100644 index 000000000..582c73891 --- /dev/null +++ b/packages/agent-core-v2/test/workspace/sessionLifecycle/forkTurnSlice.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from 'vitest'; + +import { sliceMainRecordsAtTurn } from '#/workspace/sessionLifecycle/internal/forkTurnSlice'; +import type { WireRecord } from '#/wire/record'; + +function userTurnRecord(text: string, time: number): WireRecord { + return { + type: 'context.append_message', + message: { + role: 'user', + content: [{ type: 'text', text }], + origin: { kind: 'user' }, + }, + time, + }; +} + +describe('sliceMainRecordsAtTurn', () => { + it('keeps cron records that fall inside a truncated fork slice', () => { + const records: WireRecord[] = [ + { type: 'metadata', protocol_version: '1.5', created_at: 1 }, + { + type: 'cron.add', + task: { id: 'aa11bb22', cron: '0 9 * * *', prompt: 'legacy', createdAt: 2 }, + time: 2, + }, + userTurnRecord('hello', 3), + { type: 'cron.cursor', id: 'aa11bb22', lastFiredAt: 4, time: 4 }, + userTurnRecord('second turn', 5), + { type: 'cron.add', task: { id: 'bb22cc33', cron: '0 10 * * *', prompt: 'late', createdAt: 6 }, time: 6 }, + ]; + + const slice = sliceMainRecordsAtTurn(records, 'ses_source', 0); + + const types = slice.records.map((record) => record.type); + expect(types).toContain('cron.add'); + expect(types).toContain('cron.cursor'); + expect( + slice.records.filter((record) => record.type === 'cron.add'), + ).toHaveLength(1); + expect(types).toContain('metadata'); + expect(types).toContain('context.append_message'); + }); +}); diff --git a/packages/agent-core-v2/test/workspace/workspaceInstance/workspaceInstanceManager.test.ts b/packages/agent-core-v2/test/workspace/workspaceInstance/workspaceInstanceManager.test.ts index 700c1f00d..0baf601c1 100644 --- a/packages/agent-core-v2/test/workspace/workspaceInstance/workspaceInstanceManager.test.ts +++ b/packages/agent-core-v2/test/workspace/workspaceInstance/workspaceInstanceManager.test.ts @@ -139,10 +139,10 @@ function manager( { scope: () => 'sessions' }, workspaces, { ready }, - ...Array.from({ length: 22 }, () => undefined), + ...Array.from({ length: 21 }, () => undefined), new TestRuntimeUnitHostFactory(), ]; - args[20] = { entries: () => [] }; + args[19] = { entries: () => [] }; const value = Reflect.construct(WorkspaceInstanceManager, args) as WorkspaceInstanceManager; const providers = (value as unknown as { providers: Map }).providers; providers.clear(); diff --git a/packages/kap-server/test/sessions.test.ts b/packages/kap-server/test/sessions.test.ts index 496a6794c..edec7cd58 100644 --- a/packages/kap-server/test/sessions.test.ts +++ b/packages/kap-server/test/sessions.test.ts @@ -18,6 +18,8 @@ import { IAgentLifecycleService, IEventBus, IEventService, + ISessionCronService, + ISessionManager, MAIN_AGENT_ID, closeSessionById, getLiveSessionById, @@ -913,6 +915,62 @@ describe('server-v2 /api/v1/sessions', () => { expect(children.body.data.items.some((s) => s.id === forked.body.data.id)).toBe(false); }); + it('fork inherits cron tasks through the copied wire', async () => { + const cwd = home as string; + const parent = await postJson('/api/v1/sessions', { metadata: { cwd } }); + const parentId = parent.body.data.id; + const session = getLiveSessionById((server as RunningServer).core.accessor, parentId); + expect(session).toBeDefined(); + await session!.accessor.get(IAgentLifecycleService).create({ agentId: MAIN_AGENT_ID }); + const cron = session!.accessor.get(ISessionCronService); + const task = cron.addTask({ cron: '0 9 * * *', prompt: 'fork me', recurring: true }); + + const forked = await postJson(`/api/v1/sessions/${parentId}:fork`, {}); + expect(forked.body.code).toBe(0); + + const forkedSession = getLiveSessionById( + (server as RunningServer).core.accessor, + forked.body.data.id, + ); + expect(forkedSession).toBeDefined(); + const forkedCron = forkedSession!.accessor.get(ISessionCronService); + expect(forkedCron.list().map((t) => ({ id: t.id, prompt: t.prompt }))).toEqual([ + { id: task.id, prompt: 'fork me' }, + ]); + }); + + it('keeps cron tasks across a server restart through the wire', async () => { + const cwd = home as string; + const parent = await postJson('/api/v1/sessions', { metadata: { cwd } }); + const parentId = parent.body.data.id; + const session = getLiveSessionById((server as RunningServer).core.accessor, parentId); + expect(session).toBeDefined(); + await session!.accessor.get(IAgentLifecycleService).create({ agentId: MAIN_AGENT_ID }); + const task = session!.accessor + .get(ISessionCronService) + .addTask({ cron: '0 9 * * *', prompt: 'restart me', recurring: true }); + + await (server as RunningServer).close(); + server = await startServer({ + hostIdentity: TEST_HOST_IDENTITY, + host: '127.0.0.1', + port: 0, + homeDir: home, + logLevel: 'silent', + debugEndpoints: true, + }); + base = `http://127.0.0.1:${server.port}`; + + const resumed = await (server as RunningServer).core.accessor + .get(ISessionManager) + .resume(parentId); + expect(resumed).toBeDefined(); + const cron = resumed!.accessor.get(ISessionCronService); + expect(cron.list().map((t) => ({ id: t.id, prompt: t.prompt }))).toEqual([ + { id: task.id, prompt: 'restart me' }, + ]); + }); + it('returns 40401 when listing children of a missing parent', async () => { const { body } = await getJson('/api/v1/sessions/sess_missing_parent/children'); expect(body.code).toBe(40401);