diff --git a/.changeset/fix-fork-session-files.md b/.changeset/fix-fork-session-files.md new file mode 100644 index 0000000000..2e976747d5 --- /dev/null +++ b/.changeset/fix-fork-session-files.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/agent-core-v2": patch +"@moonshot-ai/kimi-code": patch +--- + +Fix session fork losing everything except the conversation log: forked sessions now carry over media attachments, plan files, background task output, and cron tasks, and a failed fork no longer leaves a broken half-copy behind. diff --git a/.changeset/fix-tool-dedupe-ignition.md b/.changeset/fix-tool-dedupe-ignition.md new file mode 100644 index 0000000000..037750fbc0 --- /dev/null +++ b/.changeset/fix-tool-dedupe-ignition.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/agent-core-v2": patch +"@moonshot-ai/kimi-code": patch +--- + +Fix the v2 engine never activating tool-call deduplication: identical tool calls issued in the same step no longer execute multiple times, and repeated identical calls across steps receive escalating reminders again. diff --git a/packages/agent-core-v2/src/app/cron/cronTask.ts b/packages/agent-core-v2/src/app/cron/cronTask.ts index 954fec7671..7a2569dbaa 100644 --- a/packages/agent-core-v2/src/app/cron/cronTask.ts +++ b/packages/agent-core-v2/src/app/cron/cronTask.ts @@ -19,3 +19,10 @@ export interface CronTask { } export type CronTaskInit = Omit; + +/** + * `tags` key carrying the id of the session a task belongs to. The Session + * projection (`ISessionCronService`) filters the workspace-level store on + * this tag, and session fork rewrites it when cloning tasks for the fork. + */ +export const CRON_SESSION_TAG = 'sessionId'; diff --git a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts index 79f178ee30..dcc9cd14c3 100644 --- a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts +++ b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts @@ -17,6 +17,9 @@ import { randomUUID } from 'node:crypto'; +import { join } from 'pathe'; +import { ulid } from 'ulid'; + import { InstantiationType } from '#/_base/di/extensions'; import { IInstantiationService } from '#/_base/di/instantiation'; import { Disposable } from '#/_base/di/lifecycle'; @@ -26,6 +29,7 @@ import { LifecycleScope, registerScopedService, } from '#/_base/di/scope'; +import { unwrapErrorCause } from '#/_base/errors/errors'; import { Emitter, type Event } from '#/_base/event'; import { encodeWorkDirKey } from '#/_base/utils/workdir-slug'; import { ISessionActivityKernel } from '#/activity/activity'; @@ -39,6 +43,8 @@ import { } from '#/agent/wireRecord/wireRecord'; import { WIRE_RECORD_FILENAME, wireRecordScope } from '#/agent/wireRecord/wireRecordService'; 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 { @@ -53,6 +59,7 @@ import { ITelemetryService } from '#/app/telemetry/telemetry'; import { ErrorCodes, Error2, isError2 } from '#/errors'; import { createHooks } from '#/hooks'; import { IHostEnvironment } from '#/os/interface/hostEnvironment'; +import { IHostFileSystem, type HostDirEntry } from '#/os/interface/hostFileSystem'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; import { IAgentLifecycleService } from '#/session/agentLifecycle/agentLifecycle'; @@ -115,6 +122,8 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec @ISessionIndex private readonly index: ISessionIndex, @IAppendLogStore private readonly appendLogStore: IAppendLogStore, @IAtomicDocumentStore private readonly docs: IAtomicDocumentStore, + @IHostFileSystem private readonly hostFs: IHostFileSystem, + @ICronTaskPersistence private readonly cronStore: ICronTaskPersistence, @IWorkspaceRegistry private readonly workspaceRegistry: IWorkspaceRegistry, @IWorkspaceLocalConfigService private readonly workspaceLocalConfig: IWorkspaceLocalConfigService, @@ -363,6 +372,9 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec sourceHandle !== undefined ? await sourceHandle.accessor.get(ISessionActivityKernel).quiesce('fork') : undefined; + let targetId: string | undefined; + let target: ISessionScopeHandle | undefined; + let targetSessionDir: string | undefined; try { // 3. Resolve the work dir the fork inherits (same workspace as the source). const workspace = await this.workspaceRegistry.get(workspaceId); @@ -377,7 +389,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec : await this.readMetaFromDisk(workspaceId, sourceId); // 5. Mint the target id and reject collisions. - const targetId = opts.newSessionId ?? createSessionId(); + targetId = opts.newSessionId ?? createSessionId(); if (this.sessions.has(targetId) || (await this.index.get(targetId)) !== undefined) { throw new Error2( ErrorCodes.SESSION_ALREADY_EXISTS, @@ -386,16 +398,27 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec } // 6. Materialize the target session scope (fresh metadata + storage). - const target = await this.materializeSession({ + target = await this.materializeSession({ sessionId: targetId, workDir: workspace.root, }); const targetCtx = target.accessor.get(ISessionContext); + targetSessionDir = targetCtx.sessionDir; const targetMeta = target.accessor.get(ISessionMetadata); - // 7. Copy every source agent's wire log into the target's per-agent log + // 7. Copy the source session's on-disk state into the target — per-agent + // `blobs/` and `plans/`, background-task output, and media originals. + // v1 achieved this with `cp -r` of the whole session dir; the wire logs + // (step 8) and `state.json` (step 9) are rewritten by the fork flow + // itself, and `logs/` is the source's debug log, so those are excluded. + await this.copySessionFiles( + this.bootstrap.sessionDir(workspaceId, sourceId), + targetCtx.sessionDir, + ); + + // 8. Copy every source agent's wire log into the target's per-agent log // (BEFORE the target agents are created, so the logs are in place when - // their AgentWireRecordService restores them in step 9). + // their AgentWireRecordService restores them in step 11). const sourceAgents = sourceMeta?.agents ?? {}; const agentIds = Object.keys(sourceAgents); for (const agentId of agentIds) { @@ -409,7 +432,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec }); } - // 8. Rewrite the target metadata to reflect fork provenance. + // 9. Rewrite the target metadata to reflect fork provenance. const title = opts.title ?? `Fork: ${sourceMeta?.title || sourceId}`; await targetMeta.update({ title, @@ -420,7 +443,12 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec custom: forkCustomMetadata(sourceMeta?.custom, opts.metadata), }); - // 9. Create the target agents (same ids) and restore each from its copied + // 10. Clone the source session's cron tasks for the target. v1 kept cron + // records inside the session dir so `cp -r` carried them; v2 persists + // them at workspace level tagged with the owning session id. + await this.duplicateCronTasks(workspaceId, sourceId, targetId); + + // 11. Create the target agents (same ids) and restore each from its copied // log. Creating them registers fresh agent entries with TARGET homedirs. for (const agentId of agentIds) { const sourceAgent = sourceAgents[agentId]!; @@ -443,6 +471,25 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec }); await this.announceCreated({ sessionId: targetId, handle: target, source: 'fork' }); return target; + } catch (error) { + // Roll back the half-fork, mirroring v1's `rm -rf` of the target dir: + // drop the materialized handle from the live registry (otherwise a + // retry with the same id trips SESSION_ALREADY_EXISTS on the in-memory + // check) and delete whatever was copied to disk. + if (targetId !== undefined) { + this.sessions.delete(targetId); + } + if (target !== undefined) { + try { + target.dispose(); + } catch { + // best effort — the session dir is removed below regardless + } + } + if (targetSessionDir !== undefined) { + await this.hostFs.remove(targetSessionDir).catch(() => {}); + } + throw error; } finally { quiesce?.dispose(); } @@ -529,6 +576,90 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec ); } + /** + * Copy the source session's on-disk state into the target session dir — + * everything the fork flow does not regenerate itself: per-agent `blobs/` + * and `plans/`, background-task output, and session media originals. v1 + * achieved this with `cp -r` of the whole session dir; these live under + * the v2 session dir too (per-agent scopes and hostFs paths), so the fork + * must carry them explicitly or the target's blob refs resolve to + * `[media missing]` and its active plan file is gone. + * + * A missing source dir means there is nothing on disk to carry over (the + * wire logs are read through the append-log store, not this walk). + */ + private async copySessionFiles(sourceDir: string, targetDir: string): Promise { + let entries: readonly HostDirEntry[]; + try { + entries = await this.hostFs.readdir(sourceDir); + } catch (error) { + if (isMissingFileError(error)) return; + throw error; + } + await this.copySessionDirEntries(sourceDir, targetDir, entries, ''); + } + + private async copySessionDirEntries( + sourceDir: string, + targetDir: string, + entries: readonly HostDirEntry[], + relBase: string, + ): Promise { + for (const entry of entries) { + const rel = relBase === '' ? entry.name : `${relBase}/${entry.name}`; + // `state.json` is rewritten with fork provenance, the per-agent wire + // logs are copied by `copyAgentWire` (with a fork boundary record), and + // `logs/` is the source's debug log — the fork writes its own. + if (rel === 'state.json' || rel === 'logs' || entry.name === WIRE_RECORD_FILENAME) { + continue; + } + // Never follow symlinks out of the session dir. + if (entry.isSymbolicLink === true) continue; + const sourcePath = join(sourceDir, entry.name); + const targetPath = join(targetDir, entry.name); + if (entry.isDirectory) { + let children: readonly HostDirEntry[]; + try { + children = await this.hostFs.readdir(sourcePath); + } catch (error) { + if (isMissingFileError(error)) continue; + throw error; + } + await this.hostFs.mkdir(targetPath, { recursive: true }); + await this.copySessionDirEntries(sourcePath, targetPath, children, rel); + } else if (entry.isFile) { + const data = await this.hostFs.readBytes(sourcePath); + await this.hostFs.mkdir(targetDir, { recursive: true }); + await this.hostFs.writeBytes(targetPath, data); + } + } + } + + /** + * Clone the source session's cron tasks for the fork. v1 kept cron records + * inside the session dir, so its `cp -r` fork carried them; v2 persists + * cron at workspace level keyed by a session-id tag, so fork duplicates + * the source's tasks with fresh ids pointing at the target. Fired + * one-shot tasks are already gone from the store (removed on delivery), + * so everything cloned here is still live. + */ + private async duplicateCronTasks( + workspaceId: string, + sourceId: string, + targetId: string, + ): Promise { + const tasks = await this.cronStore.list({ 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(workspaceId, clone); + } + } + private async readMetaFromDisk( workspaceId: string, sessionId: string, @@ -554,6 +685,14 @@ async function collect(iterable: AsyncIterable): Promise { return items; } +/** hostFs wraps raw errnos in `HostFsError`; classify the unwrapped cause. */ +function isMissingFileError(error: unknown): boolean { + const unwrapped = unwrapErrorCause(error); + if (unwrapped === null || typeof unwrapped !== 'object') return false; + const code = (unwrapped as { readonly code?: unknown }).code; + return code === 'ENOENT'; +} + /** * Mint a session id in the canonical `session_` form, matching * v1's `createSessionId` (`packages/agent-core/src/rpc/core-impl.ts`). diff --git a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts index e6fecbbbe5..5cecd3c336 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts @@ -74,6 +74,7 @@ import { WireService } from '#/wire/wireServiceImpl'; import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { AgentBlobServiceImpl } from '#/agent/blob/agentBlobServiceImpl'; import { IAgentExternalHooksService } from '#/agent/externalHooks/externalHooks'; +import { IAgentToolDedupeService } from '#/agent/toolDedupe/toolDedupe'; import { ISessionInteractionService } from '#/session/interaction/interaction'; import { createHooks } from '#/hooks'; @@ -324,6 +325,14 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle // ReadMediaFile / MCP / prompt ingestion honor `[image] max_edge_px` and // `read_byte_budget` (and their env overrides) through the implicit default. handle.accessor.get(IImageConfigBridge); + // Tool-call dedupe plugin: self-wiring — its constructor registers the + // loop step hooks and the executor's onBefore/onDidExecuteTool handlers, + // and no other service injects it. Must be ignited BEFORE the external + // hooks service below: that get transitively constructs the permission + // gate, and `toolDedupe` has to sit ahead of `permission` on + // `onBeforeExecuteTool` so same-step duplicates are suppressed before + // authorization runs (v1 ran dedup in prepare, before authorize). + handle.accessor.get(IAgentToolDedupeService); // External hook adapter: registers listeners on the agent's domain hooks // before the first turn. No business service injects it directly; it // observes their hooks instead. diff --git a/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts b/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts index 85d1270bc5..630e472756 100644 --- a/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts +++ b/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts @@ -30,7 +30,7 @@ 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 { type CronTask, type CronTaskInit } from '#/app/cron/cronTask'; +import { CRON_SESSION_TAG, type CronTask, type CronTaskInit } from '#/app/cron/cronTask'; import { ICronTaskPersistence } from '#/app/cron/cronTaskPersistence'; import { renderCronFireXml } from '#/app/cron/format'; import { jitteredNextCronRunMs, oneShotJitteredNextCronRunMs } from '#/app/cron/jitter'; @@ -61,7 +61,6 @@ const DEFAULT_POLL_INTERVAL_MS = 1_000; const MAX_COALESCE_ITERATIONS = 10_000; const CRON_ID_REGEX: RegExp = /^(?:[0-9a-f]{8}|[0-9A-HJKMNP-TV-Z]{26})$/i; const MAX_ID_ATTEMPTS = 8; -const SESSION_TAG = 'sessionId'; export class SessionCronServiceImpl extends Disposable implements ISessionCronService { declare readonly _serviceBrand: undefined; @@ -178,7 +177,7 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe ...init, id: this.generateUniqueId(), createdAt: this.clocks.wallNow(), - tags: { ...init.tags, [SESSION_TAG]: this.ctx.sessionId }, + tags: { ...init.tags, [CRON_SESSION_TAG]: this.ctx.sessionId }, }; this.tasks.set(task.id, task); this.dispatchCron(cronAdd({ task })); @@ -240,7 +239,7 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe } const allTasks = await this.store.list({ workspaceId: this.ctx.workspaceId }); for (const task of allTasks) { - const owner = task.tags?.[SESSION_TAG]; + const owner = task.tags?.[CRON_SESSION_TAG]; if (owner !== undefined && owner !== this.ctx.sessionId) continue; if (owner === undefined) { // Legacy / hand-edited task whose shape is valid but which carries no @@ -250,7 +249,7 @@ export class SessionCronServiceImpl extends Disposable implements ISessionCronSe // so future resumes filter by tag as usual). const claimed: CronTask = { ...task, - tags: { ...task.tags, [SESSION_TAG]: this.ctx.sessionId }, + tags: { ...task.tags, [CRON_SESSION_TAG]: this.ctx.sessionId }, }; this.adopt(claimed); this.persistEnqueue(claimed.id, () => diff --git a/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts b/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts index bc73c81073..8eb745c793 100644 --- a/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts +++ b/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts @@ -1,6 +1,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { isAbsolute, resolve } from 'node:path'; +import { mkdtemp, mkdir, readFile, rm, stat, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { isAbsolute, join, resolve } from 'node:path'; import { InstantiationType } from '#/_base/di/extensions'; import { Disposable } from '#/_base/di/lifecycle'; @@ -15,6 +17,8 @@ import { Event } from '#/_base/event'; import { IBootstrapService } from '#/app/bootstrap/bootstrap'; import { IConfigService } from '#/app/config/config'; import { IHostEnvironment } from '#/os/interface/hostEnvironment'; +import { IHostFileSystem } from '#/os/interface/hostFileSystem'; +import { HostFileSystem } from '#/os/backends/node-local/hostFsService'; import { IEventService } from '#/app/event/event'; import { type AgentTaskHooks, @@ -24,6 +28,8 @@ import { import { MAIN_AGENT_ID } from '#/session/agentLifecycle/mainAgent'; import { IAgentPlanService } from '#/agent/plan/plan'; import { ISessionCronService } from '#/session/cron/sessionCronService'; +import { ICronTaskPersistence } from '#/app/cron/cronTaskPersistence'; +import { CRON_SESSION_TAG, type CronTask } from '#/app/cron/cronTask'; import { ISessionLifecycleService } from '#/app/sessionLifecycle/sessionLifecycle'; import { SessionLifecycleService } from '#/app/sessionLifecycle/sessionLifecycleService'; import { ISessionActivityKernel } from '#/activity/activity'; @@ -58,6 +64,39 @@ function bootstrapStub(): IBootstrapService { } as IBootstrapService; } +function tmpBootstrapStub(root: string): IBootstrapService { + return { + sessionsDir: join(root, 'sessions'), + homeDir: root, + sessionScope: (workspaceId: string, sessionId: string) => + `sessions/${workspaceId}/${sessionId}`, + sessionDir: (workspaceId: string, sessionId: string) => + join(root, 'sessions', workspaceId, sessionId), + agentHomedir: (workspaceId: string, sessionId: string, agentId: string) => + join(root, 'sessions', workspaceId, sessionId, 'agents', agentId), + } as IBootstrapService; +} + +function cronStoreStub( + initial: readonly CronTask[] = [], +): ICronTaskPersistence & { readonly docs: Map } { + const docs = new Map(initial.map((task) => [task.id, task])); + return { + _serviceBrand: undefined, + docs, + get: (_workspaceId, taskId) => Promise.resolve(docs.get(taskId)), + list: () => Promise.resolve([...docs.values()]), + save: (_workspaceId, task) => { + docs.set(task.id, task); + return Promise.resolve(); + }, + delete: (_workspaceId, taskId) => { + docs.delete(taskId); + return Promise.resolve(); + }, + }; +} + function metadataStub(): ISessionMetadata { return { _serviceBrand: undefined, @@ -355,10 +394,12 @@ class RecordingSessionExternalHooksService describe('SessionLifecycleService', () => { let host: ScopedTestHost | undefined; let telemetryRecords: TelemetryRecord[]; + let tmpRoots: string[]; beforeEach(() => { recordedSessionHookEvents = []; telemetryRecords = []; + tmpRoots = []; _clearScopedRegistryForTests(); registerScopedService( LifecycleScope.App, @@ -381,11 +422,22 @@ describe('SessionLifecycleService', () => { InstantiationType.Delayed, 'activity', ); + // The unit under test copies session files through hostFs on fork; the + // real backend has no dependencies and operates on the tmp paths the + // fork tests seed, so register it instead of stubbing. + registerScopedService( + LifecycleScope.App, + IHostFileSystem, + HostFileSystem, + InstantiationType.Delayed, + 'hostFs', + ); }); - afterEach(() => { + afterEach(async () => { host?.dispose(); host = undefined; + await Promise.all(tmpRoots.map((root) => rm(root, { recursive: true, force: true }))); }); function build(extra: ReturnType[] = []): ISessionLifecycleService { @@ -405,11 +457,18 @@ describe('SessionLifecycleService', () => { stubPair(ISessionActivityKernel, stubSessionActivityKernel()), stubPair(IWorkspaceLocalConfigService, workspaceLocalConfigStub()), stubPair(ITelemetryService, recordingTelemetry(telemetryRecords)), + stubPair(ICronTaskPersistence, cronStoreStub()), ...extra, ]); return host.app.accessor.get(ISessionLifecycleService); } + async function makeTmpRoot(): Promise { + const root = await mkdtemp(join(tmpdir(), 'kimi-fork-test-')); + tmpRoots.push(root); + return root; + } + it('create / get / list / close', async () => { const svc = build(); const h = await svc.create({ sessionId: 's1', workDir: '/tmp/proj' }); @@ -928,6 +987,143 @@ describe('SessionLifecycleService', () => { }); }); + describe('fork session state', () => { + function workspaceGetStub(): ReturnType { + return stubPair(IWorkspaceRegistry, { + ...workspaceRegistryStub(), + get: () => + Promise.resolve({ + id: 'wd_stub', + root: '/tmp/proj', + name: 'stub', + createdAt: 0, + lastOpenedAt: 0, + }), + }); + } + + it('copies blobs, plans, background tasks, and media originals into the fork', async () => { + const root = await makeTmpRoot(); + const svc = build([ + stubPair(IBootstrapService, tmpBootstrapStub(root)), + workspaceGetStub(), + ]); + await svc.create({ sessionId: 'src', workDir: '/tmp/proj' }); + + const srcDir = join(root, 'sessions', 'wd_stub', 'src'); + await mkdir(join(srcDir, 'agents', 'main', 'blobs'), { recursive: true }); + await writeFile(join(srcDir, 'agents', 'main', 'blobs', 'ab12cd'), 'blob-bytes'); + await mkdir(join(srcDir, 'agents', 'main', 'plans'), { recursive: true }); + await writeFile(join(srcDir, 'agents', 'main', 'plans', 'p1.md'), '# plan'); + await mkdir(join(srcDir, 'agents', 'main', 'tasks', 'bash-1'), { recursive: true }); + await writeFile(join(srcDir, 'agents', 'main', 'tasks', 'bash-1.json'), '{}'); + await writeFile(join(srcDir, 'agents', 'main', 'tasks', 'bash-1', 'output.log'), 'out'); + await mkdir(join(srcDir, 'media-originals'), { recursive: true }); + await writeFile(join(srcDir, 'media-originals', 'x.png'), 'png'); + // Excluded from the copy: state.json (rewritten with fork provenance), + // the wire logs (copied with a fork boundary record), and the source's + // debug log. + await writeFile(join(srcDir, 'state.json'), '{"source":true}'); + await writeFile(join(srcDir, 'agents', 'main', 'wire.jsonl'), '{"type":"metadata"}\n'); + await mkdir(join(srcDir, 'logs'), { recursive: true }); + await writeFile(join(srcDir, 'logs', 'kimi-code.log'), 'log'); + + await svc.fork({ sourceSessionId: 'src', newSessionId: 'dst' }); + + const dstDir = join(root, 'sessions', 'wd_stub', 'dst'); + await expect( + readFile(join(dstDir, 'agents', 'main', 'blobs', 'ab12cd'), 'utf8'), + ).resolves.toBe('blob-bytes'); + await expect( + readFile(join(dstDir, 'agents', 'main', 'plans', 'p1.md'), 'utf8'), + ).resolves.toBe('# plan'); + await expect( + readFile(join(dstDir, 'agents', 'main', 'tasks', 'bash-1.json'), 'utf8'), + ).resolves.toBe('{}'); + await expect( + readFile(join(dstDir, 'agents', 'main', 'tasks', 'bash-1', 'output.log'), 'utf8'), + ).resolves.toBe('out'); + await expect(readFile(join(dstDir, 'media-originals', 'x.png'), 'utf8')).resolves.toBe( + 'png', + ); + // The materialize path is stubbed to write nothing, so any of these in + // the target could only have come from the copy. + await expect(stat(join(dstDir, 'state.json'))).rejects.toThrow(); + await expect(stat(join(dstDir, 'agents', 'main', 'wire.jsonl'))).rejects.toThrow(); + await expect(stat(join(dstDir, 'logs'))).rejects.toThrow(); + }); + + it('rolls back the target session when fork fails after materializing', async () => { + const root = await makeTmpRoot(); + const srcDir = join(root, 'sessions', 'wd_stub', 'src'); + const svc = build([ + stubPair(IBootstrapService, tmpBootstrapStub(root)), + workspaceGetStub(), + stubPair(ISessionMetadata, { + ...metadataStub(), + read: () => + Promise.resolve({ + agents: { main: { homedir: join(srcDir, 'agents', 'main') } }, + } as never), + }), + ]); + await svc.create({ sessionId: 'src', workDir: '/tmp/proj' }); + // Seed one file so the copy materializes the target dir before the + // (stubbed) agent creation rejects. + await mkdir(join(srcDir, 'agents', 'main', 'plans'), { recursive: true }); + await writeFile(join(srcDir, 'agents', 'main', 'plans', 'p1.md'), '# plan'); + const dstDir = join(root, 'sessions', 'wd_stub', 'dst'); + + await expect(svc.fork({ sourceSessionId: 'src', newSessionId: 'dst' })).rejects.toThrow( + 'not implemented', + ); + + expect(svc.get('dst')).toBeUndefined(); + await expect(stat(dstDir)).rejects.toThrow(); + // The registry rollback unblocks a retry with the same ids: it fails + // again at agent creation, not with SESSION_ALREADY_EXISTS. + await expect(svc.fork({ sourceSessionId: 'src', newSessionId: 'dst' })).rejects.toThrow( + 'not implemented', + ); + }); + + it('duplicates the source session cron tasks for the fork', async () => { + const root = await makeTmpRoot(); + const cron = cronStoreStub([ + { + id: 'task-src', + cron: '0 9 * * *', + prompt: 'standup', + createdAt: 1, + tags: { [CRON_SESSION_TAG]: 'src' }, + }, + { + id: 'task-other', + cron: '0 9 * * *', + prompt: 'other', + createdAt: 1, + tags: { [CRON_SESSION_TAG]: 'other' }, + }, + { id: 'task-untagged', cron: '* * * * *', prompt: 'x', createdAt: 1 }, + ]); + const svc = build([ + stubPair(IBootstrapService, tmpBootstrapStub(root)), + workspaceGetStub(), + stubPair(ICronTaskPersistence, cron), + ]); + await svc.create({ sessionId: 'src', workDir: '/tmp/proj' }); + + await svc.fork({ sourceSessionId: 'src', newSessionId: 'dst' }); + + const all = [...cron.docs.values()]; + expect(all).toHaveLength(4); + const clone = all.find((task) => task.tags?.[CRON_SESSION_TAG] === 'dst'); + expect(clone).toMatchObject({ cron: '0 9 * * *', prompt: 'standup', createdAt: 1 }); + expect(clone!.id).not.toBe('task-src'); + expect(cron.docs.get('task-src')!.tags![CRON_SESSION_TAG]).toBe('src'); + }); + }); + describe('defaultPlanMode bootstrap', () => { it('enters plan mode on a fresh session when config.defaultPlanMode is true', async () => { const { lifecycle, enter, create } = agentLifecycleCapturingPlanSpy(); diff --git a/packages/agent-core-v2/test/harness/agent.ts b/packages/agent-core-v2/test/harness/agent.ts index e5443a4ad4..d88b1feb9c 100644 --- a/packages/agent-core-v2/test/harness/agent.ts +++ b/packages/agent-core-v2/test/harness/agent.ts @@ -30,6 +30,7 @@ import { IAgentPromptService } from '#/agent/prompt/prompt'; import type { AgentAPI } from '#/agent/rpc/core-api'; import { IAgentSkillService } from '#/agent/skill/skill'; import { AgentSkillService } from '#/agent/skill/skillService'; +import { IAgentToolDedupeService } from '#/agent/toolDedupe/toolDedupe'; import type { ExecutableToolOutput as ToolOutput, ExecutableToolResult, @@ -1202,6 +1203,11 @@ export class AgentTestContext { // `Bash`/etc. land in the per-agent registry the same way they would // under a real Agent scope (see `AgentLifecycleService.create`). this.get(IAgentBuiltinToolsRegistrar); + // The tool-call dedupe plugin is self-wiring too and nothing injects it. + // Ignite it BEFORE external hooks (whose construction transitively builds + // the permission gate) so `toolDedupe` stays ahead of `permission` on + // `onBeforeExecuteTool`, matching `AgentLifecycleService`. + this.get(IAgentToolDedupeService); this.get(IAgentExternalHooksService); // The step-retry plugin registers its loop error handler at construction; // nothing pulls it lazily, so ignite it the way `AgentLifecycleService` 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 ea868c4021..b244709b5b 100644 --- a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts +++ b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts @@ -23,6 +23,7 @@ import { AgentLifecycleService } from '#/session/agentLifecycle/agentLifecycleSe import { ensureMainAgent } from '#/session/agentLifecycle/mainAgent'; import { ISessionCronService } from '#/session/cron/sessionCronService'; import '#/activity/agentActivityService'; +import '#/agent/toolDedupe/toolDedupeService'; import { IAgentActivityService, ISessionActivityKernel } from '#/activity/activity'; import { IBootstrapService } from '#/app/bootstrap/bootstrap'; import { IConfigService } from '#/app/config/config'; @@ -36,6 +37,8 @@ import { ISessionContext } from '#/session/sessionContext/sessionContext'; import { ISessionMetadata } from '#/session/sessionMetadata/sessionMetadata'; import { AGENT_WIRE_PROTOCOL_VERSION, type PersistedWireRecord } from '#/agent/wireRecord/wireRecord'; import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor'; +import { IAgentLoopService } from '#/agent/loop/loop'; +import { ITelemetryService } from '#/app/telemetry/telemetry'; import { _clearToolContributionsForTests } from '#/agent/toolRegistry/toolContribution'; import { IAgentToolRegistryService } from '#/agent/toolRegistry/toolRegistry'; import { IAgentMediaToolsRegistrar } from '#/agent/media/mediaTools'; @@ -131,6 +134,8 @@ describe('AgentLifecycleService', () => { let registerAgent: ReturnType>; let atomicDocs: Map; let permissionModeSetMode: ReturnType; + let beforeExecuteHookIds: string[]; + let didExecuteHookIds: string[]; beforeEach(() => { // The unit under test force-instantiates the builtin-tools registrar per @@ -213,13 +218,37 @@ describe('AgentLifecycleService', () => { ix.stub(IAgentMediaToolsRegistrar, { _serviceBrand: undefined, } as IAgentMediaToolsRegistrar); + beforeExecuteHookIds = []; + didExecuteHookIds = []; ix.stub(IAgentToolExecutorService, { _serviceBrand: undefined, hooks: { - onBeforeExecuteTool: { register: () => ({ dispose: () => {} }) }, - onDidExecuteTool: { register: () => ({ dispose: () => {} }) }, + onBeforeExecuteTool: { + register: (id: string) => { + beforeExecuteHookIds.push(id); + return { dispose: () => {} }; + }, + }, + onDidExecuteTool: { + register: (id: string) => { + didExecuteHookIds.push(id); + return { dispose: () => {} }; + }, + }, }, } as unknown as IAgentToolExecutorService); + ix.stub(IAgentLoopService, { + _serviceBrand: undefined, + hooks: { + onWillBeginStep: { register: () => ({ dispose: () => {} }) }, + onDidFinishStep: { register: () => ({ dispose: () => {} }) }, + }, + registerLoopErrorHandler: () => ({ dispose: () => {} }), + } as unknown as IAgentLoopService); + ix.stub(ITelemetryService, { + _serviceBrand: undefined, + track2: () => {}, + } as unknown as ITelemetryService); permissionModeSetMode = vi.fn(); ix.stub(IAgentPermissionModeService, { _serviceBrand: undefined, @@ -244,6 +273,17 @@ describe('AgentLifecycleService', () => { expect(svc.getHandle('main')).toBeUndefined(); }); + it('ignites the self-wiring toolDedupe plugin so its hooks exist before the first turn', async () => { + // `AgentToolDedupeService` only acts through the loop/executor hooks its + // constructor registers; nothing injects it, so agent creation must ignite + // it explicitly (its ordering ahead of `permission` is enforced by the + // ignition order in `igniteEagerServices`). + const svc = ix.get(IAgentLifecycleService); + await svc.create({ agentId: 'main' }); + expect(beforeExecuteHookIds).toContain('toolDedupe'); + expect(didExecuteHookIds).toContain('toolDedupe'); + }); + it('seeds metadata into an empty agent wire before the first business op', async () => { const log = recordingAppendLog(); ix.stub(IAppendLogStore, log.store);