From f76fa4a25b39d01db9f4c7d9d0a395564266cf1e Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Mon, 17 Aug 2026 10:27:35 +0800 Subject: [PATCH 01/19] feat(kap-server): accept bundled skill activations on the prompt submission route MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bundled-submission capability was only reachable through the in-process klient transports; the App talks to kap-server over /api/v1. The submit-prompt route now accepts an optional non-empty skills field and delegates to IAgentSkillService.promptWithSkills — same validation, events, and single bundled user message as the TUI path — skipping its own prompt-metadata update (the engine owns it there) and mapping skill.not_found / skill.type_unsupported onto the skills route's codes. To return the submission's queue identity, the engine's promptWithSkills now resolves with prompt_id / user_message_id / created_at / state (plus turn_id once launched), mirrored through the klient contract. --- .../agent-core-v2/src/agent/skill/skill.ts | 16 +++++-- .../src/agent/skill/skillService.ts | 20 +++++++-- .../test/agent/skill/activateSkill.test.ts | 5 ++- packages/agent-core-v2/test/harness/agent.ts | 4 +- .../kap-server/src/protocol/rest-prompt.ts | 18 +++++++- packages/kap-server/src/routes/prompts.ts | 39 +++++++++++++++- packages/kap-server/test/prompts.test.ts | 45 +++++++++++++++++++ packages/klient/src/contract/agent/schemas.ts | 20 +++++++++ .../klient/src/contract/agent/services.ts | 3 +- packages/klient/src/core/facade/agent.ts | 11 +++-- packages/klient/src/index.ts | 1 + packages/klient/test/contract-parity.ts | 6 +++ packages/klient/test/facade.test.ts | 16 ++++++- 13 files changed, 186 insertions(+), 18 deletions(-) diff --git a/packages/agent-core-v2/src/agent/skill/skill.ts b/packages/agent-core-v2/src/agent/skill/skill.ts index 3d0076d19e9..a6558d4fb24 100644 --- a/packages/agent-core-v2/src/agent/skill/skill.ts +++ b/packages/agent-core-v2/src/agent/skill/skill.ts @@ -10,12 +10,14 @@ * the rendered skill blocks precede the caller's parts in the content and the * activation metadata rides the prompt's origin, so the bundle is a single * turn and a single undo unit), and records model-tool activations without a - * turn (`recordModelToolActivation`). Bound at Agent scope. + * turn (`recordModelToolActivation`). `promptWithSkills` resolves with the + * queue identity of the submitted bundle (`prompt_id` / `user_message_id` / + * `created_at` / `state`, plus `turn_id` once launched). Bound at Agent scope. */ import { createDecorator } from "#/_base/di/instantiation"; import type { SkillActivationOrigin } from '#/agent/contextMemory/types'; -import type { PromptLaunchResult } from '#/agent/prompt/prompt'; +import type { PromptLaunchResult, PromptState } from '#/agent/prompt/prompt'; import type { ContentPart } from '#/kosong/contract/message'; export interface SkillActivationInput { @@ -34,11 +36,19 @@ export interface PromptWithSkillsInput { readonly skills: readonly PromptSkillActivation[]; } +export interface PromptWithSkillsResult { + readonly turn_id?: number; + readonly prompt_id: string; + readonly user_message_id: string; + readonly created_at: string; + readonly state: PromptState; +} + export interface IAgentSkillService { readonly _serviceBrand: undefined; activate(input: SkillActivationInput): Promise; - promptWithSkills(input: PromptWithSkillsInput): Promise; + promptWithSkills(input: PromptWithSkillsInput): Promise; recordModelToolActivation(origin: SkillActivationOrigin): void; } diff --git a/packages/agent-core-v2/src/agent/skill/skillService.ts b/packages/agent-core-v2/src/agent/skill/skillService.ts index dea30d29073..cc7c1440736 100644 --- a/packages/agent-core-v2/src/agent/skill/skillService.ts +++ b/packages/agent-core-v2/src/agent/skill/skillService.ts @@ -48,6 +48,7 @@ import { IAgentSkillService, type PromptSkillActivation, type PromptWithSkillsInput, + type PromptWithSkillsResult, type SkillActivationInput, } from './skill'; import { skillActivate } from './skillOps'; @@ -136,7 +137,7 @@ export class AgentSkillService extends Service implements IAgentSkillService { return { turn_id: turn.id }; } - async promptWithSkills(input: PromptWithSkillsInput): Promise { + async promptWithSkills(input: PromptWithSkillsInput): Promise { if (input.input.length === 0) { throw new Error2(ErrorCodes.REQUEST_INVALID, 'promptWithSkills requires a non-empty prompt'); } @@ -172,9 +173,22 @@ export class AgentSkillService extends Service implements IAgentSkillService { }, }, }); - if (handle.state === 'pending') return undefined; + if (handle.state === 'pending') { + return { + prompt_id: handle.id, + user_message_id: handle.userMessageId, + created_at: handle.createdAt, + state: handle.state, + }; + } const turn = await handle.launched; - return turn === undefined ? undefined : { turn_id: turn.id }; + return { + turn_id: turn?.id, + prompt_id: handle.id, + user_message_id: handle.userMessageId, + created_at: handle.createdAt, + state: handle.state, + }; } recordModelToolActivation(origin: SkillActivationOrigin): void { diff --git a/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts b/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts index 33bed108bb8..115399e4c5f 100644 --- a/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts +++ b/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts @@ -89,7 +89,10 @@ describe('promptWithSkills', () => { input: [{ type: 'text', text: 'Review this change.' }], skills: [{ name: 'review' }, { name: 'security' }], }); - expect(launched?.turn_id).toBe(0); + expect(launched.turn_id).toBe(0); + expect(launched.prompt_id).toBeTruthy(); + expect(launched.user_message_id).toBeTruthy(); + expect(launched.state).toBe('running'); await ctx.untilTurnEnd(); expect(ctx.llmCalls).toHaveLength(1); diff --git a/packages/agent-core-v2/test/harness/agent.ts b/packages/agent-core-v2/test/harness/agent.ts index f03f37b63b7..27e16d716db 100644 --- a/packages/agent-core-v2/test/harness/agent.ts +++ b/packages/agent-core-v2/test/harness/agent.ts @@ -87,7 +87,7 @@ interface StopTaskPayload { readonly taskId: string; readonly reason?: string } interface UndoHistoryPayload { readonly count: number } interface UnregisterToolPayload { readonly name: string } import { type UsageStatus } from '#/agent/usage/usage'; -import { IAgentSkillService, type PromptWithSkillsInput, type SkillActivationInput } from '#/agent/skill/skill'; +import { IAgentSkillService, type PromptWithSkillsInput, type PromptWithSkillsResult, type SkillActivationInput } from '#/agent/skill/skill'; import { AgentSkillService } from '#/agent/skill/skillService'; import { IAgentRuntimeBindingSeed } from '#/agent/runtimeBinding/runtimeBinding'; import { IAgentRuntimeService } from '#/agent/runtimeBinding/agentRuntime'; @@ -335,7 +335,7 @@ type RpcPromise = Promise & { interface AgentRpcPassthroughAPI { prompt: (payload: PromptPayload) => Promisable; - promptWithSkills: (payload: PromptWithSkillsInput) => Promisable; + promptWithSkills: (payload: PromptWithSkillsInput) => Promisable; steer: (payload: SteerPayload) => Promisable; cancel: (payload: CancelPayload) => void; undoHistory: (payload: UndoHistoryPayload) => Promisable; diff --git a/packages/kap-server/src/protocol/rest-prompt.ts b/packages/kap-server/src/protocol/rest-prompt.ts index b04a2fe78af..64d077b97e7 100644 --- a/packages/kap-server/src/protocol/rest-prompt.ts +++ b/packages/kap-server/src/protocol/rest-prompt.ts @@ -2,9 +2,16 @@ * POST /v1/sessions/{sid}/prompts * Body: PromptSubmission { content, metadata?, agent_id?, profile?, model?, thinking?, * permission_mode?, plan_mode?, swarm_mode?, goal_objective?, goal_control?, - * disabled_tools? } + * disabled_tools?, skills? } * Reply: PromptSubmitResult { prompt_id, user_message_id, status, content, created_at } * + * `skills` (optional, non-empty) submits a bundled skill prompt through + * `IAgentSkillService.promptWithSkills`: every named skill is validated up + * front (an unknown name rejects the whole submission), one `skill.activated` + * event fires per skill, and the prompt enqueues as a single user message + * with the rendered skill blocks preceding the caller's content — one turn, + * one undo unit. Each entry is `{ name, args? }` (activation by name only). + * * GET /v1/sessions/{sid}/prompts * Reply: { active: PromptItem | null, queued: PromptItem[] } * @@ -29,6 +36,12 @@ import { export { promptPermissionModeSchema, promptThinkingSchema }; export type { PromptPermissionMode, PromptThinking } from '@moonshot-ai/agent-core-v2/app/sessionLegacy/sessionProtocol'; +export const promptSkillActivationSchema = z.object({ + name: z.string().min(1), + args: z.string().optional(), +}); +export type PromptSkillActivation = z.infer; + export const promptSubmissionSchema = z.object({ content: z.array(messageContentSchema).min(1), metadata: z.record(z.string(), z.unknown()).optional(), @@ -47,6 +60,9 @@ export const promptSubmissionSchema = z.object({ // bound profile's own deny always survives. Omit to keep the persisted // value, send `[]` to clear the client portion. disabled_tools: z.array(z.string()).optional(), + // Bundled skill submission: every named skill activates with the prompt as + // a single bundled user message (one turn, one undo unit). + skills: z.array(promptSkillActivationSchema).min(1).optional(), }); export type PromptSubmission = z.infer; diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 4f26880f5d7..550d5a3eefb 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -14,6 +14,7 @@ import { IAgentProfileService, IAgentToolPolicyService, IAgentPromptService, + IAgentSkillService, IAuthSummaryService, IEventService, IFileService, @@ -115,6 +116,7 @@ async function resolvePromptFromSession(session: ISessionScopeHandle, agentId?: } return { prompt: agent.accessor.get(IAgentPromptService), + skill: agent.accessor.get(IAgentSkillService), auth: agent.accessor.get(IAuthSummaryService), profile: agent.accessor.get(IAgentProfileService), toolPolicy: agent.accessor.get(IAgentToolPolicyService), @@ -202,7 +204,7 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { [ErrorCode.SESSION_NOT_FOUND]: {}, [ErrorCode.PROMPT_ALREADY_COMPLETED]: { dataSchema: z.object({ aborted: z.literal(false) }) }, }, - description: 'Submit a prompt to a session', + description: 'Submit a prompt to a session, optionally with bundled skill activations', tags: ['prompts'], operationId: 'submitPrompt', }, @@ -270,6 +272,35 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { } } const parts = contentToCoreParts(resolvedContent); + if (req.body.skills !== undefined) { + // Bundled skill submission: the engine validates every skill up + // front, records one activation event per skill, and enqueues a + // single user message (rendered skill blocks first, then the + // caller's parts). It owns the prompt-metadata update for the main + // agent, so this edge skips its own to avoid a double write. + const result = await resolved.skill.promptWithSkills({ + input: parts, + skills: req.body.skills, + }); + reply.send( + okEnvelope( + { + prompt_id: result.prompt_id, + user_message_id: result.user_message_id, + status: + result.state === 'running' || result.state === 'steered' + ? 'running' + : result.state === 'blocked' + ? 'blocked' + : 'queued', + content: corePartsToProtocol(parts), + created_at: result.created_at, + }, + req.id, + ), + ); + return; + } const session = await resolveSession(core, session_id); await applyPromptMetadataUpdate({ metadata: session.accessor.get(ISessionMetadata), @@ -451,6 +482,12 @@ function sendMappedError( case 'validation.failed': reply.send(errEnvelope(ErrorCode.VALIDATION_FAILED, err.message, requestId, err.stack)); return; + case 'skill.not_found': + reply.send(errEnvelope(ErrorCode.SKILL_NOT_FOUND, err.message, requestId, err.stack)); + return; + case 'skill.type_unsupported': + reply.send(errEnvelope(ErrorCode.SKILL_NOT_ACTIVATABLE, err.message, requestId, err.stack)); + return; case 'auth.provisioning_required': reply.send({ code: ErrorCode.AUTH_PROVISIONING_REQUIRED, diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 10ceb752a39..bea453ca0a4 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -216,6 +216,51 @@ describe('server-v2 /api/v1 prompts', () => { expect(Array.isArray(list.body.data.queued)).toBe(true); }); + it('submits a bundled skill prompt through the skills field', async () => { + const id = await createSession(home as string); + await createMainAgent(id); + + const submitted = await call('POST', `/api/v1/sessions/${id}/prompts`, { + content: [{ type: 'text', text: 'Review this change.' }], + skills: [{ name: 'update-config' }, { name: 'check-kimi-code-docs' }], + }); + expect(submitted.body.code).toBe(0); + expect(submitted.body.data.prompt_id).toMatch(/^msg_/); + expect(['running', 'queued']).toContain(submitted.body.data.status); + expect(submitted.body.data.content).toEqual([{ type: 'text', text: 'Review this change.' }]); + + // The bundled prompt lands as one user message with every activation on + // its origin and the rendered skill blocks ahead of the caller's text. + const session = getLiveSessionById(server!.core.accessor, id); + const agent = session!.accessor.get(IAgentLifecycleService).get('main'); + const history = agent!.accessor.get(IAgentContextMemoryService).get(); + const bundled = history.find((message) => message.origin?.kind === 'user'); + expect(bundled?.origin).toMatchObject({ + kind: 'user', + skillActivations: [{ skillName: 'update-config' }, { skillName: 'check-kimi-code-docs' }], + }); + const texts = bundled?.content + .filter((part) => part.type === 'text') + .map((part) => part.text); + expect(texts?.[texts.length - 1]).toBe('Review this change.'); + }); + + it('rejects a bundled submission with an unknown skill and records nothing', async () => { + const id = await createSession(home as string); + await createMainAgent(id); + + const submitted = await call('POST', `/api/v1/sessions/${id}/prompts`, { + content: [{ type: 'text', text: 'Review this change.' }], + skills: [{ name: 'does-not-exist' }], + }); + expect(submitted.body.code).toBe(40415); + + const session = getLiveSessionById(server!.core.accessor, id); + const agent = session!.accessor.get(IAgentLifecycleService).get('main'); + const history = agent!.accessor.get(IAgentContextMemoryService).get(); + expect(history.filter((message) => message.origin?.kind === 'user')).toHaveLength(0); + }); + it('makes the first three REST prompts available to title generation', async () => { const id = await createSession(home as string); await createMainAgent(id); diff --git a/packages/klient/src/contract/agent/schemas.ts b/packages/klient/src/contract/agent/schemas.ts index fabcfcd5c82..069284556b7 100644 --- a/packages/klient/src/contract/agent/schemas.ts +++ b/packages/klient/src/contract/agent/schemas.ts @@ -52,6 +52,26 @@ export const promptWithSkillsPayloadSchema = promptPayloadSchema.extend({ skills: z.array(promptSkillActivationSchema).min(1), }); +/** Same shape as `PromptState` in the engine. */ +export const promptStateSchema = z.enum([ + 'pending', + 'running', + 'steered', + 'completed', + 'failed', + 'cancelled', + 'blocked', +]); + +/** Same shape as `PromptWithSkillsResult` in the engine. */ +export const promptWithSkillsResultSchema = z.object({ + turn_id: z.number().optional(), + prompt_id: z.string(), + user_message_id: z.string(), + created_at: z.string(), + state: promptStateSchema, +}); + /** Same shape as `SteerPayload` in the engine. */ export const steerPayloadSchema = z.object({ input: z.array(promptPartSchema), diff --git a/packages/klient/src/contract/agent/services.ts b/packages/klient/src/contract/agent/services.ts index 554d94b724a..552be30d555 100644 --- a/packages/klient/src/contract/agent/services.ts +++ b/packages/klient/src/contract/agent/services.ts @@ -19,6 +19,7 @@ import { promptLaunchResultSchema, promptPayloadSchema, promptWithSkillsPayloadSchema, + promptWithSkillsResultSchema, runShellCommandPayloadSchema, runtimeBindingSchema, setModelResultSchema, @@ -42,7 +43,7 @@ export const agentSkillContract = { activate: { input: z.tuple([activateSkillPayloadSchema]), output: promptLaunchResultSchema }, promptWithSkills: { input: z.tuple([promptWithSkillsPayloadSchema]), - output: maybe(promptLaunchResultSchema), + output: maybe(promptWithSkillsResultSchema), }, } satisfies ServiceContract; diff --git a/packages/klient/src/core/facade/agent.ts b/packages/klient/src/core/facade/agent.ts index 3fdb529b69a..be9953e71c8 100644 --- a/packages/klient/src/core/facade/agent.ts +++ b/packages/klient/src/core/facade/agent.ts @@ -30,6 +30,7 @@ import type { ScopedCaller } from './session.js'; // klient free of protocol-package imports). export type PromptLaunchResult = Awaited>; export type PromptWithSkillsInput = Parameters[0]; +export type PromptWithSkillsResult = Awaited>; export type ShellCommandResult = Awaited>; export type SetModelResult = Awaited>; export type ThinkingLevel = ReturnType; @@ -51,10 +52,12 @@ export interface AgentFacade { * same user message: the skills are validated up front (an unknown name or * an empty list rejects the whole submission), rendered ahead of the * caller's parts in the same turn, and the bundle undoes as a single - * anchor. Resolves with the launched turn id, or `undefined` when the - * submission queued behind a running turn. + * anchor. Resolves with the submitted bundle's queue identity (`prompt_id` + * / `user_message_id` / `created_at` / `state`), plus `turn_id` once + * launched — `state` is `pending` when the submission queued behind a + * running turn. */ - promptWithSkills(input: PromptWithSkillsInput): Promise; + promptWithSkills(input: PromptWithSkillsInput): Promise; steer(input: { input: readonly ContentPart[] }): Promise; /** * Activate a skill as a user-slash activation: the engine renders the skill @@ -103,7 +106,7 @@ export function createAgentFacade(call: ScopedCaller, scope: ScopeRef): AgentFac prompt: (input) => call(scope, 'agentPromptService', 'submit', [input]) as Promise, promptWithSkills: (input) => - call(scope, 'agentSkillService', 'promptWithSkills', [input]) as Promise, + call(scope, 'agentSkillService', 'promptWithSkills', [input]) as Promise, steer: (input) => call(scope, 'agentPromptService', 'submitSteer', [input]) as Promise, activateSkill: (input) => diff --git a/packages/klient/src/index.ts b/packages/klient/src/index.ts index 40eb1745e93..3bda190d2e5 100644 --- a/packages/klient/src/index.ts +++ b/packages/klient/src/index.ts @@ -74,6 +74,7 @@ export type { PlanData, PromptLaunchResult, PromptWithSkillsInput, + PromptWithSkillsResult, SetModelResult, ShellCommandResult, ThinkingLevel, diff --git a/packages/klient/test/contract-parity.ts b/packages/klient/test/contract-parity.ts index 1738515e1a4..cb5bb014f54 100644 --- a/packages/klient/test/contract-parity.ts +++ b/packages/klient/test/contract-parity.ts @@ -166,6 +166,7 @@ import { promptPayloadSchema, promptSkillActivationSchema, promptWithSkillsPayloadSchema, + promptWithSkillsResultSchema, runCommandPayloadSchema, runShellCommandPayloadSchema, runtimeBindingSchema, @@ -576,6 +577,11 @@ const _steerPayload: AssertWireToEngine const _activateSkillPayload: AssertWire = true; const _promptLaunchResult: AssertWire = true; +type PromptWithSkillsResult = Awaited>; +const _promptWithSkillsResult: AssertWire< + typeof promptWithSkillsResultSchema, + PromptWithSkillsResult +> = true; const _cancelPayload: AssertWire = true; const _runShellCommandPayload: AssertWire< typeof runShellCommandPayloadSchema, diff --git a/packages/klient/test/facade.test.ts b/packages/klient/test/facade.test.ts index 21b0b21fd8c..ad0b57325c1 100644 --- a/packages/klient/test/facade.test.ts +++ b/packages/klient/test/facade.test.ts @@ -189,13 +189,25 @@ describe('agent skill routing', () => { const klient = createKlientFromChannel(channel); const agent = klient.session('s1').agent('main'); - channel.result = { turn_id: 7 }; + channel.result = { + turn_id: 7, + prompt_id: 'p1', + user_message_id: 'm1', + created_at: '2026-01-01T00:00:00.000Z', + state: 'running', + }; await expect( agent.promptWithSkills({ input: [{ type: 'text', text: 'Review this change.' }], skills: [{ name: 'review' }, { name: 'security', args: 'src/app.ts' }], }), - ).resolves.toEqual({ turn_id: 7 }); + ).resolves.toEqual({ + turn_id: 7, + prompt_id: 'p1', + user_message_id: 'm1', + created_at: '2026-01-01T00:00:00.000Z', + state: 'running', + }); expect(channel.calls[0]).toEqual({ scope: { sessionId: 's1', agentId: 'main' }, service: 'agentSkillService', From 6bce8b31f2f675ddaf69e59fe2d5acc99ce8ffe7 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Mon, 17 Aug 2026 10:45:58 +0800 Subject: [PATCH 02/19] refactor(agent-core-v2): slim the promptWithSkills result contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop the user_message_id field (it is always the same identity as prompt_id — the route duplicates it) and narrow state to the running/queued/blocked vocabulary, mapped at the engine edge instead of exposing the internal seven-state PromptState on the wire. --- packages/agent-core-v2/src/agent/skill/skill.ts | 9 ++++----- .../agent-core-v2/src/agent/skill/skillService.ts | 10 ++-------- .../test/agent/skill/activateSkill.test.ts | 1 - packages/kap-server/src/routes/prompts.ts | 11 ++++------- packages/klient/src/contract/agent/schemas.ts | 14 +------------- packages/klient/src/core/facade/agent.ts | 5 ++--- packages/klient/test/facade.test.ts | 2 -- 7 files changed, 13 insertions(+), 39 deletions(-) diff --git a/packages/agent-core-v2/src/agent/skill/skill.ts b/packages/agent-core-v2/src/agent/skill/skill.ts index a6558d4fb24..5ede80c0b66 100644 --- a/packages/agent-core-v2/src/agent/skill/skill.ts +++ b/packages/agent-core-v2/src/agent/skill/skill.ts @@ -11,13 +11,13 @@ * activation metadata rides the prompt's origin, so the bundle is a single * turn and a single undo unit), and records model-tool activations without a * turn (`recordModelToolActivation`). `promptWithSkills` resolves with the - * queue identity of the submitted bundle (`prompt_id` / `user_message_id` / - * `created_at` / `state`, plus `turn_id` once launched). Bound at Agent scope. + * queue identity of the submitted bundle (`prompt_id` / `created_at`, the + * launch `state`, and `turn_id` once launched). Bound at Agent scope. */ import { createDecorator } from "#/_base/di/instantiation"; import type { SkillActivationOrigin } from '#/agent/contextMemory/types'; -import type { PromptLaunchResult, PromptState } from '#/agent/prompt/prompt'; +import type { PromptLaunchResult } from '#/agent/prompt/prompt'; import type { ContentPart } from '#/kosong/contract/message'; export interface SkillActivationInput { @@ -39,9 +39,8 @@ export interface PromptWithSkillsInput { export interface PromptWithSkillsResult { readonly turn_id?: number; readonly prompt_id: string; - readonly user_message_id: string; readonly created_at: string; - readonly state: PromptState; + readonly state: 'running' | 'queued' | 'blocked'; } export interface IAgentSkillService { diff --git a/packages/agent-core-v2/src/agent/skill/skillService.ts b/packages/agent-core-v2/src/agent/skill/skillService.ts index cc7c1440736..7aea6021e08 100644 --- a/packages/agent-core-v2/src/agent/skill/skillService.ts +++ b/packages/agent-core-v2/src/agent/skill/skillService.ts @@ -174,20 +174,14 @@ export class AgentSkillService extends Service implements IAgentSkillService { }, }); if (handle.state === 'pending') { - return { - prompt_id: handle.id, - user_message_id: handle.userMessageId, - created_at: handle.createdAt, - state: handle.state, - }; + return { prompt_id: handle.id, created_at: handle.createdAt, state: 'queued' }; } const turn = await handle.launched; return { turn_id: turn?.id, prompt_id: handle.id, - user_message_id: handle.userMessageId, created_at: handle.createdAt, - state: handle.state, + state: handle.state === 'blocked' ? 'blocked' : 'running', }; } diff --git a/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts b/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts index 115399e4c5f..924f3c21b10 100644 --- a/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts +++ b/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts @@ -91,7 +91,6 @@ describe('promptWithSkills', () => { }); expect(launched.turn_id).toBe(0); expect(launched.prompt_id).toBeTruthy(); - expect(launched.user_message_id).toBeTruthy(); expect(launched.state).toBe('running'); await ctx.untilTurnEnd(); diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 550d5a3eefb..1e74c73ae8a 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -286,13 +286,10 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { okEnvelope( { prompt_id: result.prompt_id, - user_message_id: result.user_message_id, - status: - result.state === 'running' || result.state === 'steered' - ? 'running' - : result.state === 'blocked' - ? 'blocked' - : 'queued', + // prompt_id IS the user_message_id — one identity for prompt + // and message, same as the plain-prompt path. + user_message_id: result.prompt_id, + status: result.state, content: corePartsToProtocol(parts), created_at: result.created_at, }, diff --git a/packages/klient/src/contract/agent/schemas.ts b/packages/klient/src/contract/agent/schemas.ts index 069284556b7..e37e3a15a90 100644 --- a/packages/klient/src/contract/agent/schemas.ts +++ b/packages/klient/src/contract/agent/schemas.ts @@ -52,24 +52,12 @@ export const promptWithSkillsPayloadSchema = promptPayloadSchema.extend({ skills: z.array(promptSkillActivationSchema).min(1), }); -/** Same shape as `PromptState` in the engine. */ -export const promptStateSchema = z.enum([ - 'pending', - 'running', - 'steered', - 'completed', - 'failed', - 'cancelled', - 'blocked', -]); - /** Same shape as `PromptWithSkillsResult` in the engine. */ export const promptWithSkillsResultSchema = z.object({ turn_id: z.number().optional(), prompt_id: z.string(), - user_message_id: z.string(), created_at: z.string(), - state: promptStateSchema, + state: z.enum(['running', 'queued', 'blocked']), }); /** Same shape as `SteerPayload` in the engine. */ diff --git a/packages/klient/src/core/facade/agent.ts b/packages/klient/src/core/facade/agent.ts index be9953e71c8..6e8d8be22b9 100644 --- a/packages/klient/src/core/facade/agent.ts +++ b/packages/klient/src/core/facade/agent.ts @@ -53,9 +53,8 @@ export interface AgentFacade { * an empty list rejects the whole submission), rendered ahead of the * caller's parts in the same turn, and the bundle undoes as a single * anchor. Resolves with the submitted bundle's queue identity (`prompt_id` - * / `user_message_id` / `created_at` / `state`), plus `turn_id` once - * launched — `state` is `pending` when the submission queued behind a - * running turn. + * / `created_at` / `state`), plus `turn_id` once launched — `state` is + * `queued` when the submission queued behind a running turn. */ promptWithSkills(input: PromptWithSkillsInput): Promise; steer(input: { input: readonly ContentPart[] }): Promise; diff --git a/packages/klient/test/facade.test.ts b/packages/klient/test/facade.test.ts index ad0b57325c1..011a69f66b5 100644 --- a/packages/klient/test/facade.test.ts +++ b/packages/klient/test/facade.test.ts @@ -192,7 +192,6 @@ describe('agent skill routing', () => { channel.result = { turn_id: 7, prompt_id: 'p1', - user_message_id: 'm1', created_at: '2026-01-01T00:00:00.000Z', state: 'running', }; @@ -204,7 +203,6 @@ describe('agent skill routing', () => { ).resolves.toEqual({ turn_id: 7, prompt_id: 'p1', - user_message_id: 'm1', created_at: '2026-01-01T00:00:00.000Z', state: 'running', }); From 09deb908b9330bcfa997013f714c37cec2d5d3ac Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Mon, 17 Aug 2026 11:07:49 +0800 Subject: [PATCH 03/19] fix(kap-server): harden bundled skill submissions against review findings - Validate bundled skill names and types before any media materialization or control override, so a rejected bundle leaves session state untouched (the engine still re-validates authoritatively). - Declare the 40415/40912 outcomes on the submit route so the generated API documentation includes them. - The klient output schema no longer tolerates a missing promptWithSkills result (a transport-level absence now raises instead of resolving undefined), and a failed launch surfaces as an error rather than a successful running result. - Add the changeset for the new public API field. --- .changeset/kap-prompt-skills.md | 5 +++ .../src/agent/skill/skillService.ts | 3 ++ packages/kap-server/src/routes/prompts.ts | 38 +++++++++++++++++++ packages/kap-server/test/prompts.test.ts | 19 ++++++++++ .../klient/src/contract/agent/services.ts | 2 +- 5 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 .changeset/kap-prompt-skills.md diff --git a/.changeset/kap-prompt-skills.md b/.changeset/kap-prompt-skills.md new file mode 100644 index 00000000000..c8ee256c5bc --- /dev/null +++ b/.changeset/kap-prompt-skills.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +The session prompt submission API now accepts an optional `skills` field: one or more named skills activate together with the prompt as a single bundled turn (one undo unit), validated up front with zero side effects on rejection. diff --git a/packages/agent-core-v2/src/agent/skill/skillService.ts b/packages/agent-core-v2/src/agent/skill/skillService.ts index 7aea6021e08..65bba58b50c 100644 --- a/packages/agent-core-v2/src/agent/skill/skillService.ts +++ b/packages/agent-core-v2/src/agent/skill/skillService.ts @@ -177,6 +177,9 @@ export class AgentSkillService extends Service implements IAgentSkillService { return { prompt_id: handle.id, created_at: handle.createdAt, state: 'queued' }; } const turn = await handle.launched; + if (turn === undefined && handle.state !== 'blocked') { + throw new Error2(ErrorCodes.INTERNAL, 'promptWithSkills failed to launch a turn'); + } return { turn_id: turn?.id, prompt_id: handle.id, diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 1e74c73ae8a..127d3796c6f 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -19,6 +19,8 @@ import { IEventService, IFileService, ISessionMetadata, + ISessionSkillCatalog, + isUserActivatableSkillType, parseKimiFileUrl, promptMetadataTextFromContentParts, ProfileError, @@ -44,6 +46,7 @@ import { promptSteerResultSchema, promptSubmissionSchema, promptSubmitResultSchema, + type PromptSkillActivation, type PromptSubmission, } from '../protocol/rest-prompt'; import { z } from 'zod'; @@ -117,6 +120,7 @@ async function resolvePromptFromSession(session: ISessionScopeHandle, agentId?: return { prompt: agent.accessor.get(IAgentPromptService), skill: agent.accessor.get(IAgentSkillService), + skillCatalog: session.accessor.get(ISessionSkillCatalog), auth: agent.accessor.get(IAuthSummaryService), profile: agent.accessor.get(IAgentProfileService), toolPolicy: agent.accessor.get(IAgentToolPolicyService), @@ -124,6 +128,33 @@ async function resolvePromptFromSession(session: ISessionScopeHandle, agentId?: }; } +/** + * Read-only pre-flight for bundled skill submissions: every named skill must + * exist in the session catalog and be user-activatable. Runs before any media + * materialization or control override, so a rejected bundle leaves the + * session untouched. The engine re-validates authoritatively inside + * `promptWithSkills`; this edge check only exists to protect side effects + * that precede it (media copies, persistent overrides). + */ +async function assertActivatableSkills( + catalog: ISessionSkillCatalog, + skills: readonly PromptSkillActivation[], +): Promise { + await catalog.ready; + for (const skill of skills) { + const definition = catalog.catalog.getSkill(skill.name); + if (definition === undefined) { + throw new Error2(ErrorCodes.SKILL_NOT_FOUND, `Skill "${skill.name}" was not found`); + } + if (!isUserActivatableSkillType(definition.metadata.type)) { + throw new Error2( + ErrorCodes.SKILL_TYPE_UNSUPPORTED, + `Skill "${definition.name}" cannot be activated by the user`, + ); + } + } +} + /** * Bind the resolved agent to the profile named by a prompt submission's * `profile` field. First-bind semantics live in the engine: a same-name @@ -197,6 +228,8 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { success: { data: promptSubmitResultSchema }, errors: { [ErrorCode.VALIDATION_FAILED]: { detailsSchema: validationDetailsSchema }, + [ErrorCode.SKILL_NOT_FOUND]: {}, + [ErrorCode.SKILL_NOT_ACTIVATABLE]: {}, [ErrorCode.AUTH_PROVISIONING_REQUIRED]: {}, [ErrorCode.AUTH_TOKEN_MISSING]: { detailsSchema: authProviderDetailsSchema }, [ErrorCode.AUTH_TOKEN_UNAUTHORIZED]: { detailsSchema: authProviderDetailsSchema }, @@ -216,6 +249,11 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { // in session metadata, or touch the session's controls. await assertPromptFileRefs(req.body.content, core.accessor.get(IFileService)); const resolved = await resolvePrompt(core, session_id, req.body.agent_id); + if (req.body.skills !== undefined) { + // Bundled skills validate before any media materialization or + // control override: a rejected bundle leaves the session untouched. + await assertActivatableSkills(resolved.skillCatalog, req.body.skills); + } await resolved.auth.ensureReady(); // Media resolution runs BEFORE any control mutation, so a failed diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index bea453ca0a4..6811545665b 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -7,6 +7,7 @@ import { IAgentTitlePromptSource, IAgentContextMemoryService, IAgentLifecycleService, + IAgentPermissionModeService, IAgentProfileService, IAgentToolPolicyService, closeSessionById, @@ -261,6 +262,24 @@ describe('server-v2 /api/v1 prompts', () => { expect(history.filter((message) => message.origin?.kind === 'user')).toHaveLength(0); }); + it('rejects an unknown bundled skill before any control override binds', async () => { + const id = await createSession(home as string); + await createMainAgent(id); + + const submitted = await call('POST', `/api/v1/sessions/${id}/prompts`, { + content: [{ type: 'text', text: 'Review this change.' }], + permission_mode: 'yolo', + skills: [{ name: 'does-not-exist' }], + }); + expect(submitted.body.code).toBe(40415); + + const session = getLiveSessionById(server!.core.accessor, id); + const agent = session!.accessor.get(IAgentLifecycleService).get('main'); + expect(agent!.accessor.get(IAgentPermissionModeService).mode).toBe('manual'); + const history = agent!.accessor.get(IAgentContextMemoryService).get(); + expect(history.filter((message) => message.origin?.kind === 'user')).toHaveLength(0); + }); + it('makes the first three REST prompts available to title generation', async () => { const id = await createSession(home as string); await createMainAgent(id); diff --git a/packages/klient/src/contract/agent/services.ts b/packages/klient/src/contract/agent/services.ts index 552be30d555..54755a7ab54 100644 --- a/packages/klient/src/contract/agent/services.ts +++ b/packages/klient/src/contract/agent/services.ts @@ -43,7 +43,7 @@ export const agentSkillContract = { activate: { input: z.tuple([activateSkillPayloadSchema]), output: promptLaunchResultSchema }, promptWithSkills: { input: z.tuple([promptWithSkillsPayloadSchema]), - output: maybe(promptWithSkillsResultSchema), + output: promptWithSkillsResultSchema, }, } satisfies ServiceContract; From 90907f277e8f4d6fa9286d7bd246c567eb957fb7 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Mon, 17 Aug 2026 11:37:18 +0800 Subject: [PATCH 04/19] fix(kap-server): preflight bundled skills before agent materialization and stabilize listed content - Skill preflight now runs on the session's catalog before the main agent is resolved, so a rejected bundle cannot mutate session metadata by registering main (regression test on a cold session without an agent). - The prompts list projection strips the stored skill blocks from a bundled prompt, so GET /prompts returns the same caller-only content as the submit response. --- packages/kap-server/src/routes/prompts.ts | 25 ++++++++---- packages/kap-server/test/prompts.test.ts | 49 +++++++++++++++++++++++ 2 files changed, 66 insertions(+), 8 deletions(-) diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 127d3796c6f..5d1d3059aec 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -120,7 +120,6 @@ async function resolvePromptFromSession(session: ISessionScopeHandle, agentId?: return { prompt: agent.accessor.get(IAgentPromptService), skill: agent.accessor.get(IAgentSkillService), - skillCatalog: session.accessor.get(ISessionSkillCatalog), auth: agent.accessor.get(IAuthSummaryService), profile: agent.accessor.get(IAgentProfileService), toolPolicy: agent.accessor.get(IAgentToolPolicyService), @@ -248,12 +247,17 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { // mutated: a bad `file_id` must not create the agent, register `main` // in session metadata, or touch the session's controls. await assertPromptFileRefs(req.body.content, core.accessor.get(IFileService)); - const resolved = await resolvePrompt(core, session_id, req.body.agent_id); + // A cold resume loads the session but does not create the main agent; + // bundled-skill preflight runs at this point precisely so a rejected + // bundle cannot even mutate session metadata by registering `main`. + const session = await resolveSession(core, session_id); if (req.body.skills !== undefined) { - // Bundled skills validate before any media materialization or - // control override: a rejected bundle leaves the session untouched. - await assertActivatableSkills(resolved.skillCatalog, req.body.skills); + await assertActivatableSkills( + session.accessor.get(ISessionSkillCatalog), + req.body.skills, + ); } + const resolved = await resolvePromptFromSession(session, req.body.agent_id); await resolved.auth.ensureReady(); // Media resolution runs BEFORE any control mutation, so a failed @@ -336,7 +340,6 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { ); return; } - const session = await resolveSession(core, session_id); await applyPromptMetadataUpdate({ metadata: session.accessor.get(ISessionMetadata), eventService: core.accessor.get(IEventService), @@ -441,15 +444,21 @@ function projectPromptHandle(handle: PromptHandle) { return projectPromptSnapshot(handle); } -function projectPromptSnapshot(prompt: PromptQueueSnapshot['pending'][number]) { +export function projectPromptSnapshot(prompt: PromptQueueSnapshot['pending'][number]) { const status = prompt.state === 'running' || prompt.state === 'steered' ? 'running' : prompt.state === 'blocked' ? 'blocked' : 'queued'; + const origin = prompt.message.origin; + // A bundled prompt stores the rendered skill blocks ahead of the caller's + // parts; project the caller's parts only, so the listed content matches the + // submit response for the same prompt. + const bundled = origin?.kind === 'user' ? (origin.skillActivations?.length ?? 0) : 0; + const content = bundled === 0 ? prompt.message.content : prompt.message.content.slice(bundled); return { prompt_id: prompt.id, user_message_id: prompt.userMessageId, status, - content: corePartsToProtocol(prompt.message.content), + content: corePartsToProtocol(content), created_at: prompt.createdAt, }; } diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 6811545665b..6ac573879c1 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -16,6 +16,7 @@ import { import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { type RunningServer, startServer } from '../src/start'; +import { projectPromptSnapshot } from '../src/routes/prompts'; import { TEST_HOST_IDENTITY } from './helpers/hostIdentity'; import { authHeaders } from './helpers/auth'; @@ -244,6 +245,41 @@ describe('server-v2 /api/v1 prompts', () => { .filter((part) => part.type === 'text') .map((part) => part.text); expect(texts?.[texts.length - 1]).toBe('Review this change.'); + + // The list projection shows the same caller-only content as the submit + // response (the stored skill blocks are not projected back). + const projected = projectPromptSnapshot({ + id: 'msg_1', + userMessageId: 'msg_1', + createdAt: '2026-01-01T00:00:00.000Z', + state: 'running', + message: { + role: 'user', + content: [ + { type: 'text', text: 'rendered skill block' }, + { type: 'text', text: 'Review this change.' }, + ], + toolCalls: [], + origin: { + kind: 'user', + skillActivations: [{ activationId: 'a1', skillName: 'update-config' }], + }, + }, + }); + expect(projected.content).toEqual([{ type: 'text', text: 'Review this change.' }]); + const plain = projectPromptSnapshot({ + id: 'msg_2', + userMessageId: 'msg_2', + createdAt: '2026-01-01T00:00:00.000Z', + state: 'pending', + message: { + role: 'user', + content: [{ type: 'text', text: 'plain question' }], + toolCalls: [], + origin: { kind: 'user' }, + }, + }); + expect(plain.content).toEqual([{ type: 'text', text: 'plain question' }]); }); it('rejects a bundled submission with an unknown skill and records nothing', async () => { @@ -280,6 +316,19 @@ describe('server-v2 /api/v1 prompts', () => { expect(history.filter((message) => message.origin?.kind === 'user')).toHaveLength(0); }); + it('rejects an unknown bundled skill without materializing the main agent', async () => { + const id = await createSession(home as string); + + const submitted = await call('POST', `/api/v1/sessions/${id}/prompts`, { + content: [{ type: 'text', text: 'Review this change.' }], + skills: [{ name: 'does-not-exist' }], + }); + expect(submitted.body.code).toBe(40415); + + const session = getLiveSessionById(server!.core.accessor, id); + expect(session!.accessor.get(IAgentLifecycleService).get('main')).toBeUndefined(); + }); + it('makes the first three REST prompts available to title generation', async () => { const id = await createSession(home as string); await createMainAgent(id); From 482566517a44bc9294da7df725511bceb4678f66 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Mon, 17 Aug 2026 22:20:07 +0800 Subject: [PATCH 05/19] fix(kap-server): reject bundled prompt_id combos at preflight and clean queued staging - The skills + prompt_id incompatibility rejection now runs at the initial bundled preflight, before the main agent is materialized or any override binds (previously a yolo override could bind before the 40001). - Queued bundles no longer skip staging cleanup forever: the discard is deferred to the bundle's prompt.completed / prompt.aborted lifecycle event, mirroring the plain path's launch-raced cleanup. --- packages/kap-server/src/routes/prompts.ts | 51 +++++++++++++++++------ packages/kap-server/test/prompts.test.ts | 41 +++++++++++++++++- 2 files changed, 78 insertions(+), 14 deletions(-) diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index c30b1d0fb28..e74b8b16965 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -123,6 +123,7 @@ async function resolvePromptFromSession(session: ISessionScopeHandle, agentId?: return { prompt: agent.accessor.get(IAgentPromptService), skill: agent.accessor.get(IAgentSkillService), + events: agent.accessor.get(IEventService), auth: agent.accessor.get(IAuthSummaryService), profile: agent.accessor.get(IAgentProfileService), toolPolicy: agent.accessor.get(IAgentToolPolicyService), @@ -259,6 +260,15 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { // bundle cannot even mutate session metadata by registering `main`. const session = await resolveSession(core, session_id); if (req.body.skills !== undefined) { + // A bundled submission goes through the engine's own enqueue path, + // which assigns the prompt id — reject the combination here, before + // the agent is materialized or any override binds. + if (req.body.prompt_id !== undefined) { + throw new Error2( + ErrorCodes.REQUEST_INVALID, + 'prompt_id cannot be combined with a bundled skill submission', + ); + } await assertActivatableSkills( session.accessor.get(ISessionSkillCatalog), req.body.skills, @@ -329,16 +339,6 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { } const parts = contentToCoreParts(resolvedContent); if (req.body.skills !== undefined) { - // Bundled submissions go through the engine's own enqueue path, - // which assigns the prompt id — a client-chosen prompt_id cannot be - // honored here, so the combination is rejected explicitly instead - // of being silently ignored. - if (req.body.prompt_id !== undefined) { - throw new Error2( - ErrorCodes.REQUEST_INVALID, - 'prompt_id cannot be combined with a bundled skill submission', - ); - } // Bundled skill submission: the engine validates every skill up // front, records one activation event per skill, and enqueues a // single user message (rendered skill blocks first, then the @@ -349,10 +349,11 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { skills: req.body.skills, }); enqueued = true; - // Queued bundles complete media intake only when their turn pops, - // which this edge cannot observe through the result shape; launched - // (or hook-blocked) bundles are safe to release now. + // Queued bundles complete media intake only when their turn pops; + // the result shape carries no handle, so the plain path's deferred + // cleanup is mirrored through the prompt lifecycle events instead. if (result.state !== 'queued') await preparedMedia?.discard(); + else deferDiscardUntilPromptSettles(resolved.events, result.prompt_id, () => preparedMedia?.discard()); reply.send( okEnvelope( { @@ -509,6 +510,30 @@ export function projectPromptSnapshot(prompt: PromptQueueSnapshot['pending'][num }; } +/** + * Deferred media-staging cleanup for a queued bundled submission: the + * bundle's prompt intake only runs when its turn pops (or settles), so the + * staging upload is discarded on the matching `prompt.completed` / + * `prompt.aborted` lifecycle event rather than eagerly at submit time. + * Mirrors the plain path's `launched`/`completion`-raced discard, which the + * `promptWithSkills` result shape cannot express. + */ +export function deferDiscardUntilPromptSettles( + events: IEventService, + promptId: string, + discard: () => void | Promise, +): void { + const subscription = events.subscribe((event) => { + if ( + (event.type === 'prompt.completed' || event.type === 'prompt.aborted') && + (event as { readonly promptId?: unknown }).promptId === promptId + ) { + subscription.dispose(); + void discard(); + } + }); +} + function sendMappedError( diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 556bdb35160..044e9647e5e 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -20,7 +20,7 @@ import { import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { type RunningServer, startServer } from '../src/start'; -import { projectPromptSnapshot } from '../src/routes/prompts'; +import { deferDiscardUntilPromptSettles, projectPromptSnapshot } from '../src/routes/prompts'; import { TEST_HOST_IDENTITY } from './helpers/hostIdentity'; import { authHeaders } from './helpers/auth'; @@ -402,6 +402,45 @@ describe('server-v2 /api/v1 prompts', () => { expect(session!.accessor.get(IAgentLifecycleService).get('main')).toBeUndefined(); }); + it('rejects a bundled prompt_id combination before any override or agent materialization', async () => { + const id = await createSession(home as string); + + const submitted = await call('POST', `/api/v1/sessions/${id}/prompts`, { + content: [{ type: 'text', text: 'Review this change.' }], + permission_mode: 'yolo', + prompt_id: 'submission-1', + skills: [{ name: 'update-config' }], + }); + expect(submitted.body.code).toBe(40001); + + const session = getLiveSessionById(server!.core.accessor, id); + expect(session!.accessor.get(IAgentLifecycleService).get('main')).toBeUndefined(); + }); + + it('discards queued bundle staging on the prompt lifecycle events', async () => { + const handlers: Array<(event: { type: string; promptId?: string }) => void> = []; + const events = { + subscribe(handler: (event: { type: string; promptId?: string }) => void) { + handlers.push(handler); + return { dispose: vi.fn() }; + }, + }; + const discard = vi.fn(); + deferDiscardUntilPromptSettles(events as never, 'msg_1', discard); + + handlers[0]!({ type: 'prompt.completed', promptId: 'msg_other' }); + expect(discard).not.toHaveBeenCalled(); + handlers[0]!({ type: 'turn.started' }); + expect(discard).not.toHaveBeenCalled(); + handlers[0]!({ type: 'prompt.completed', promptId: 'msg_1' }); + expect(discard).toHaveBeenCalledTimes(1); + + const second = vi.fn(); + deferDiscardUntilPromptSettles(events as never, 'msg_2', second); + handlers[1]!({ type: 'prompt.aborted', promptId: 'msg_2' }); + expect(second).toHaveBeenCalledTimes(1); + }); + it('makes the first three REST prompts available to title generation', async () => { const id = await createSession(home as string); await createMainAgent(id); From eb4e6dbd147d4d32fbb1a42b0c82a7a21f46c6ef Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:03:49 +0800 Subject: [PATCH 06/19] fix(kap-server): clean queued bundle staging on the steer path too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A queued bundle steered into the active turn is consumed at steer time, but the engine publishes prompt.completed/aborted only for the parent — the deferred cleanup never fired and its subscription leaked. The prompt.steered event (matching promptIds) now counts as the child's intake-completion signal. --- packages/kap-server/src/routes/prompts.ts | 24 ++++++++++++++++------- packages/kap-server/test/prompts.test.ts | 11 +++++++++-- 2 files changed, 26 insertions(+), 9 deletions(-) diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index e74b8b16965..080c46782f5 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -514,9 +514,13 @@ export function projectPromptSnapshot(prompt: PromptQueueSnapshot['pending'][num * Deferred media-staging cleanup for a queued bundled submission: the * bundle's prompt intake only runs when its turn pops (or settles), so the * staging upload is discarded on the matching `prompt.completed` / - * `prompt.aborted` lifecycle event rather than eagerly at submit time. - * Mirrors the plain path's `launched`/`completion`-raced discard, which the - * `promptWithSkills` result shape cannot express. + * `prompt.aborted` lifecycle event rather than eagerly at submit time. A + * bundle steered into the active turn is consumed at steer time and the + * engine publishes completed/aborted only for the parent, so the + * `prompt.steered` event (matching `promptIds`) counts as its + * intake-completion signal too. Mirrors the plain path's + * `launched`/`completion`-raced discard, which the `promptWithSkills` + * result shape cannot express. */ export function deferDiscardUntilPromptSettles( events: IEventService, @@ -524,10 +528,16 @@ export function deferDiscardUntilPromptSettles( discard: () => void | Promise, ): void { const subscription = events.subscribe((event) => { - if ( - (event.type === 'prompt.completed' || event.type === 'prompt.aborted') && - (event as { readonly promptId?: unknown }).promptId === promptId - ) { + const settles = + ((event.type === 'prompt.completed' || event.type === 'prompt.aborted') && + (event as { readonly promptId?: unknown }).promptId === promptId) || + (event.type === 'prompt.steered' && + ( + (event as { readonly promptIds?: unknown }).promptIds as + | readonly string[] + | undefined + )?.includes(promptId) === true); + if (settles) { subscription.dispose(); void discard(); } diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 044e9647e5e..3b995a2ce60 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -418,9 +418,9 @@ describe('server-v2 /api/v1 prompts', () => { }); it('discards queued bundle staging on the prompt lifecycle events', async () => { - const handlers: Array<(event: { type: string; promptId?: string }) => void> = []; + const handlers: Array<(event: { type: string; promptId?: string; promptIds?: string[] }) => void> = []; const events = { - subscribe(handler: (event: { type: string; promptId?: string }) => void) { + subscribe(handler: (event: { type: string; promptId?: string; promptIds?: string[] }) => void) { handlers.push(handler); return { dispose: vi.fn() }; }, @@ -439,6 +439,13 @@ describe('server-v2 /api/v1 prompts', () => { deferDiscardUntilPromptSettles(events as never, 'msg_2', second); handlers[1]!({ type: 'prompt.aborted', promptId: 'msg_2' }); expect(second).toHaveBeenCalledTimes(1); + + const steered = vi.fn(); + deferDiscardUntilPromptSettles(events as never, 'msg_3', steered); + handlers[2]!({ type: 'prompt.steered', promptIds: ['msg_other'] }); + expect(steered).not.toHaveBeenCalled(); + handlers[2]!({ type: 'prompt.steered', promptIds: ['msg_parent', 'msg_3'] }); + expect(steered).toHaveBeenCalledTimes(1); }); it('makes the first three REST prompts available to title generation', async () => { From e03a946d094741160eaf58f81568f097ecba7451 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:28:37 +0800 Subject: [PATCH 07/19] fix(agent-core-v2): materialize daemon-ref media on the steer and inject paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit startNext materializes daemon file references into the session media store before a prompt's turn, but steer() and inject() enqueued the same references without that intake, leaving the staging upload as the only copy — any staging cleanup at steer time would delete the media the turn is about to consume. Both paths now run the same intake before the SteerStepRequest is created, so prompt.steered is a truthful intake-complete signal. --- .../src/agent/prompt/promptService.ts | 15 ++++--- .../test/agent/prompt/promptService.test.ts | 43 ++++++++++++++++++- 2 files changed, 52 insertions(+), 6 deletions(-) diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 77944af2f9f..2de9dcdc219 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -322,6 +322,7 @@ export class AgentPromptService implements IAgentPromptService { role: 'user', content: selected.flatMap((item) => item.message.content), toolCalls: [], origin: USER_PROMPT_ORIGIN, }; const { message: rerouted, captions } = this.extractCompressionCaptions(message); + await this.materializeDaemonRefs(rerouted); const request = new SteerStepRequest(rerouted, captions, this.reminders, (materialized) => { void this.dispatcher.dispatch( new TurnSteer({ input: materialized.content, origin: materialized.origin ?? USER_PROMPT_ORIGIN }), @@ -355,6 +356,7 @@ export class AgentPromptService implements IAgentPromptService { async inject(message: ContextMessage): Promise { const { message: rerouted, captions } = this.extractCompressionCaptions(message); + await this.materializeDaemonRefs(rerouted); const request = new SteerStepRequest(rerouted, captions, this.reminders, (materialized) => { void this.dispatcher.dispatch( new TurnSteer({ input: materialized.content, origin: materialized.origin ?? USER_PROMPT_ORIGIN }), @@ -378,11 +380,7 @@ export class AgentPromptService implements IAgentPromptService { try { if (this.fullCompaction.compacting !== null && this.loop.status().state !== 'running') { this.pending.unshift(item); return; } const { message, captions } = this.extractCompressionCaptions(item.message); - if (message.content.some((part) => daemonFileRefFromPart(part) !== undefined)) { - const files = this.instantiation.invokeFunction((accessor) => accessor.get(IFileService)); - const mediaStore = this.instantiation.invokeFunction((accessor) => accessor.get(ISessionMediaStore)); - await materializePromptDaemonRefs(message.content, { files, mediaStore }); - } + await this.materializeDaemonRefs(message); if (await this.blockedByHook(message, false)) { this.appendPrompt(message, captions); item.state = 'blocked'; item.launchedDeferred.resolve(undefined); item.completionDeferred.resolve({ promptId: item.id, result: undefined, state: 'blocked' }); @@ -414,6 +412,13 @@ export class AgentPromptService implements IAgentPromptService { void this.startNext(); } + private async materializeDaemonRefs(message: ContextMessage): Promise { + if (!message.content.some((part) => daemonFileRefFromPart(part) !== undefined)) return; + const files = this.instantiation.invokeFunction((accessor) => accessor.get(IFileService)); + const mediaStore = this.instantiation.invokeFunction((accessor) => accessor.get(ISessionMediaStore)); + await materializePromptDaemonRefs(message.content, { files, mediaStore }); + } + private async blockedByHook(promptMessage: ContextMessage, isSteer: boolean): Promise { const ctx = { promptMessage, isSteer, block: false }; await this.hooks.onBeforeSubmitPrompt.run(ctx); return ctx.block; } diff --git a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts index 688be554d73..bdf6c320c13 100644 --- a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts +++ b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts @@ -8,6 +8,8 @@ import { describe, expect, it, onTestFinished, vi } from 'vitest'; +import { Readable } from 'node:stream'; + import { DisposableStore } from '#/_base/di/lifecycle'; import { createServices } from '#/_base/di/test'; import { Event } from '#/_base/event'; @@ -34,6 +36,8 @@ import { ISessionMetadata } from '#/session/sessionMetadata/sessionMetadata'; import { IEventDispatcher } from '#/state/eventDispatcher'; import { EventDispatcherService } from '#/state/eventDispatcherService'; import { IWireService } from '#/wire/wire'; +import { IFileService } from '#/app/file/fileService'; +import { ISessionMediaStore } from '#/agent/media/sessionMediaStore'; import { stubContextMemory } from '../contextMemory/stubs'; import { stubLoopWithHooks, stubToolExecutor, stubWire } from '../loop/stubs'; @@ -62,6 +66,19 @@ function harness() { hooks: createHooks(['onWillCompact']), onDidFinishCompaction: Event.None, } as unknown as IAgentFullCompactionService; + const intake = { + get: vi.fn(async () => ({ + meta: { + id: 'file_1', + size: 3, + name: 'pic.png', + media_type: 'image/png', + created_at: '2026-01-01T00:00:00.000Z', + }, + stream: () => Readable.from([new Uint8Array([1, 2, 3])]), + })), + materialize: vi.fn(async (): Promise => undefined), + }; const ix = createServices(disposables, { strict: true, additionalServices: (reg) => { registerStateServices(reg); @@ -84,9 +101,11 @@ function harness() { reg.definePartialInstance(IEventService, { publish: () => {} }); reg.definePartialInstance(ISessionContext, { sessionId: 'test-session' }); reg.defineInstance(IAgentScopeContext, makeAgentScopeContext({ agentId: 'main', agentScope: '' })); + reg.definePartialInstance(IFileService, { get: intake.get }); + reg.definePartialInstance(ISessionMediaStore, { materialize: intake.materialize }); } }); - return { prompt: ix.get(IAgentPromptService), loop, context, fullCompaction, eventBus: ix.get(IEventBus) }; + return { prompt: ix.get(IAgentPromptService), loop, context, fullCompaction, eventBus: ix.get(IEventBus), intake }; } describe('AgentPromptService', () => { @@ -247,4 +266,26 @@ describe('AgentPromptService', () => { parts.some((part) => part.type === 'text' && part.text.includes('image/avif')), ).toBe(true); }); + + it('materializes daemon-ref media at steer intake', async () => { + const { prompt, intake } = harness(); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const queued = await prompt.enqueue({ + id: 'prompt-steer-daemon', + message: { + role: 'user', + content: [{ type: 'image_url', imageUrl: { url: 'kimi-file://file_1' } }], + toolCalls: [], + origin: { kind: 'user' }, + }, + }); + + await prompt.steer([queued.id]); + + expect(intake.get).toHaveBeenCalledWith('file_1'); + expect(intake.materialize).toHaveBeenCalledWith( + expect.objectContaining({ fileId: 'file_1', name: 'pic.png' }), + ); + }); }); From 0124751b78523f8f7bf5b1509941d2f7f014503c Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:50:45 +0800 Subject: [PATCH 08/19] fix(kap-server): defer staging cleanup to turn settlement, never to steer time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prompt-intake materialization is best-effort: when it degrades, the daemon upload is the request-time resolver's fallback source. Discarding staging at prompt.steered could therefore delete the only readable copy before the parent's request ran. Cleanup is now uniformly event-driven — the bundle's own prompt.completed/aborted, or the steer parent's — so the upload always outlives the request it feeds. --- packages/kap-server/src/routes/prompts.ts | 60 +++++++++++++---------- packages/kap-server/test/prompts.test.ts | 18 +++++-- 2 files changed, 49 insertions(+), 29 deletions(-) diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 080c46782f5..65d949b52b5 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -349,11 +349,12 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { skills: req.body.skills, }); enqueued = true; - // Queued bundles complete media intake only when their turn pops; - // the result shape carries no handle, so the plain path's deferred - // cleanup is mirrored through the prompt lifecycle events instead. - if (result.state !== 'queued') await preparedMedia?.discard(); - else deferDiscardUntilPromptSettles(resolved.events, result.prompt_id, () => preparedMedia?.discard()); + // Staging releases only when the bundle's turn lifecycle settles + // (or its steer parent's): the daemon upload is the resolver's + // fallback when intake degraded, so it must outlive the request. + deferDiscardUntilPromptSettles(resolved.events, result.prompt_id, () => + preparedMedia?.discard(), + ); reply.send( okEnvelope( { @@ -511,33 +512,42 @@ export function projectPromptSnapshot(prompt: PromptQueueSnapshot['pending'][num } /** - * Deferred media-staging cleanup for a queued bundled submission: the - * bundle's prompt intake only runs when its turn pops (or settles), so the - * staging upload is discarded on the matching `prompt.completed` / - * `prompt.aborted` lifecycle event rather than eagerly at submit time. A - * bundle steered into the active turn is consumed at steer time and the - * engine publishes completed/aborted only for the parent, so the - * `prompt.steered` event (matching `promptIds`) counts as its - * intake-completion signal too. Mirrors the plain path's - * `launched`/`completion`-raced discard, which the `promptWithSkills` - * result shape cannot express. + * Deferred media-staging cleanup for a bundled submission: the daemon + * upload is the request-time resolver's fallback whenever intake degraded + * (materialization is best-effort by design), so the staging blob must + * outlive the bundle's request. Cleanup therefore fires only when the + * bundle's turn lifecycle settles (`prompt.completed` / `prompt.aborted` + * with the bundle's id). A bundle steered into an active turn resolves its + * refs during the parent's request, and the engine publishes + * completed/aborted only for the parent — so `prompt.steered` (matching + * `promptIds`) re-targets the cleanup at the parent (`activePromptId`) + * instead of discarding at steer time. */ export function deferDiscardUntilPromptSettles( events: IEventService, promptId: string, discard: () => void | Promise, ): void { + const watched = new Set([promptId]); const subscription = events.subscribe((event) => { - const settles = - ((event.type === 'prompt.completed' || event.type === 'prompt.aborted') && - (event as { readonly promptId?: unknown }).promptId === promptId) || - (event.type === 'prompt.steered' && - ( - (event as { readonly promptIds?: unknown }).promptIds as - | readonly string[] - | undefined - )?.includes(promptId) === true); - if (settles) { + if (event.type === 'prompt.steered') { + const steered = event as { + readonly promptIds?: unknown; + readonly activePromptId?: unknown; + }; + if ( + Array.isArray(steered.promptIds) && + steered.promptIds.includes(promptId) && + typeof steered.activePromptId === 'string' + ) { + watched.add(steered.activePromptId); + } + return; + } + if ( + (event.type === 'prompt.completed' || event.type === 'prompt.aborted') && + watched.has((event as { readonly promptId?: unknown }).promptId as string) + ) { subscription.dispose(); void discard(); } diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 3b995a2ce60..62b0b2f5b94 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -418,9 +418,11 @@ describe('server-v2 /api/v1 prompts', () => { }); it('discards queued bundle staging on the prompt lifecycle events', async () => { - const handlers: Array<(event: { type: string; promptId?: string; promptIds?: string[] }) => void> = []; + const handlers: Array<(event: { type: string; promptId?: string; promptIds?: string[]; activePromptId?: string }) => void> = []; const events = { - subscribe(handler: (event: { type: string; promptId?: string; promptIds?: string[] }) => void) { + subscribe( + handler: (event: { type: string; promptId?: string; promptIds?: string[]; activePromptId?: string }) => void, + ) { handlers.push(handler); return { dispose: vi.fn() }; }, @@ -442,10 +444,18 @@ describe('server-v2 /api/v1 prompts', () => { const steered = vi.fn(); deferDiscardUntilPromptSettles(events as never, 'msg_3', steered); - handlers[2]!({ type: 'prompt.steered', promptIds: ['msg_other'] }); + handlers[2]!({ type: 'prompt.steered', promptIds: ['msg_3'], activePromptId: 'msg_parent' }); expect(steered).not.toHaveBeenCalled(); - handlers[2]!({ type: 'prompt.steered', promptIds: ['msg_parent', 'msg_3'] }); + handlers[2]!({ type: 'prompt.completed', promptId: 'msg_other' }); + expect(steered).not.toHaveBeenCalled(); + handlers[2]!({ type: 'prompt.completed', promptId: 'msg_parent' }); expect(steered).toHaveBeenCalledTimes(1); + + const unrelatedSteer = vi.fn(); + deferDiscardUntilPromptSettles(events as never, 'msg_4', unrelatedSteer); + handlers[3]!({ type: 'prompt.steered', promptIds: ['msg_other'], activePromptId: 'msg_parent' }); + handlers[3]!({ type: 'prompt.completed', promptId: 'msg_parent' }); + expect(unrelatedSteer).not.toHaveBeenCalled(); }); it('makes the first three REST prompts available to title generation', async () => { From ae982a50d3107544e0dccea9447095192826bb26 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:02:08 +0800 Subject: [PATCH 09/19] fix(kap-server): install settlement tracking before bundled enqueue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A hook-blocked bundle completes synchronously inside the submission call, and an exceptionally fast launch can settle just as early — a post-call subscription misses the only settlement event and leaks both the staging blob and the listener. The tracker now subscribes before enqueueing, buffers lifecycle events, and settles against the returned prompt id (or its steer parent's). --- packages/kap-server/src/routes/prompts.ts | 81 ++++++++++++++--------- packages/kap-server/test/prompts.test.ts | 34 +++++----- 2 files changed, 69 insertions(+), 46 deletions(-) diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 65d949b52b5..8946a7660f9 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -339,6 +339,12 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { } const parts = contentToCoreParts(resolvedContent); if (req.body.skills !== undefined) { + // Settlement tracking installs before enqueue: a hook-blocked + // bundle completes synchronously inside the call, and a fast + // launch can settle just as early. The daemon upload is the + // resolver's fallback when intake degraded, so cleanup fires only + // once the bundle's lifecycle (or its steer parent's) settles. + const settlement = watchPromptSettlements(resolved.events); // Bundled skill submission: the engine validates every skill up // front, records one activation event per skill, and enqueues a // single user message (rendered skill blocks first, then the @@ -349,12 +355,7 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { skills: req.body.skills, }); enqueued = true; - // Staging releases only when the bundle's turn lifecycle settles - // (or its steer parent's): the daemon upload is the resolver's - // fallback when intake degraded, so it must outlive the request. - deferDiscardUntilPromptSettles(resolved.events, result.prompt_id, () => - preparedMedia?.discard(), - ); + settlement.settle(result.prompt_id, () => preparedMedia?.discard()); reply.send( okEnvelope( { @@ -512,46 +513,64 @@ export function projectPromptSnapshot(prompt: PromptQueueSnapshot['pending'][num } /** - * Deferred media-staging cleanup for a bundled submission: the daemon + * Settlement tracker for bundled submissions' media staging. The daemon * upload is the request-time resolver's fallback whenever intake degraded - * (materialization is best-effort by design), so the staging blob must - * outlive the bundle's request. Cleanup therefore fires only when the - * bundle's turn lifecycle settles (`prompt.completed` / `prompt.aborted` - * with the bundle's id). A bundle steered into an active turn resolves its - * refs during the parent's request, and the engine publishes - * completed/aborted only for the parent — so `prompt.steered` (matching - * `promptIds`) re-targets the cleanup at the parent (`activePromptId`) - * instead of discarding at steer time. + * (materialization is best-effort by design), so it must outlive the + * bundle's request: cleanup fires only when the bundle's turn lifecycle + * settles (`prompt.completed` / `prompt.aborted`). A bundle steered into + * an active turn resolves its refs during the parent's request, and the + * engine publishes completed/aborted only for the parent — so + * `prompt.steered` (matching `promptIds`) re-targets the cleanup at the + * parent (`activePromptId`). + * + * The tracker installs BEFORE the submission is enqueued: a hook-blocked + * prompt completes synchronously inside the submission call, and an + * exceptionally fast launch can settle just as early — both land in the + * buffered `settledIds` instead of racing past a late subscription. */ -export function deferDiscardUntilPromptSettles( - events: IEventService, - promptId: string, - discard: () => void | Promise, -): void { - const watched = new Set([promptId]); +export function watchPromptSettlements(events: IEventService): { + settle(promptId: string, discard: () => void | Promise): void; +} { + const settledIds = new Set(); + const parentOf = new Map(); + let armed: { id: string; discard: () => void | Promise } | undefined; const subscription = events.subscribe((event) => { if (event.type === 'prompt.steered') { const steered = event as { readonly promptIds?: unknown; readonly activePromptId?: unknown; }; - if ( - Array.isArray(steered.promptIds) && - steered.promptIds.includes(promptId) && - typeof steered.activePromptId === 'string' - ) { - watched.add(steered.activePromptId); + if (Array.isArray(steered.promptIds) && typeof steered.activePromptId === 'string') { + for (const childId of steered.promptIds) { + if (typeof childId === 'string') parentOf.set(childId, steered.activePromptId); + } + if (armed !== undefined && steered.promptIds.includes(armed.id)) { + armed = { id: steered.activePromptId, discard: armed.discard }; + } } return; } - if ( - (event.type === 'prompt.completed' || event.type === 'prompt.aborted') && - watched.has((event as { readonly promptId?: unknown }).promptId as string) - ) { + if (event.type !== 'prompt.completed' && event.type !== 'prompt.aborted') return; + const id = (event as { readonly promptId?: unknown }).promptId; + if (typeof id !== 'string') return; + settledIds.add(id); + if (armed !== undefined && armed.id === id) { + const { discard } = armed; + armed = undefined; subscription.dispose(); void discard(); } }); + return { + settle(promptId: string, discard: () => void | Promise): void { + if (settledIds.has(promptId) || settledIds.has(parentOf.get(promptId) ?? '')) { + subscription.dispose(); + void discard(); + return; + } + armed = { id: promptId, discard }; + }, + }; } diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 62b0b2f5b94..138f2b1a3c3 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -20,7 +20,7 @@ import { import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { type RunningServer, startServer } from '../src/start'; -import { deferDiscardUntilPromptSettles, projectPromptSnapshot } from '../src/routes/prompts'; +import { projectPromptSnapshot, watchPromptSettlements } from '../src/routes/prompts'; import { TEST_HOST_IDENTITY } from './helpers/hostIdentity'; import { authHeaders } from './helpers/auth'; @@ -417,7 +417,7 @@ describe('server-v2 /api/v1 prompts', () => { expect(session!.accessor.get(IAgentLifecycleService).get('main')).toBeUndefined(); }); - it('discards queued bundle staging on the prompt lifecycle events', async () => { + it('cleans bundled staging through the settlement tracker', async () => { const handlers: Array<(event: { type: string; promptId?: string; promptIds?: string[]; activePromptId?: string }) => void> = []; const events = { subscribe( @@ -427,23 +427,27 @@ describe('server-v2 /api/v1 prompts', () => { return { dispose: vi.fn() }; }, }; - const discard = vi.fn(); - deferDiscardUntilPromptSettles(events as never, 'msg_1', discard); + const discard = vi.fn(); + const tracker = watchPromptSettlements(events as never); + tracker.settle('msg_1', discard); handlers[0]!({ type: 'prompt.completed', promptId: 'msg_other' }); - expect(discard).not.toHaveBeenCalled(); handlers[0]!({ type: 'turn.started' }); expect(discard).not.toHaveBeenCalled(); handlers[0]!({ type: 'prompt.completed', promptId: 'msg_1' }); expect(discard).toHaveBeenCalledTimes(1); - const second = vi.fn(); - deferDiscardUntilPromptSettles(events as never, 'msg_2', second); - handlers[1]!({ type: 'prompt.aborted', promptId: 'msg_2' }); - expect(second).toHaveBeenCalledTimes(1); + // A hook-blocked bundle completes synchronously inside the submission + // call, before `settle` runs: the buffered event must still clean up. + const blockedDiscard = vi.fn(); + const blockedTracker = watchPromptSettlements(events as never); + handlers[1]!({ type: 'prompt.completed', promptId: 'msg_blocked' }); + blockedTracker.settle('msg_blocked', blockedDiscard); + expect(blockedDiscard).toHaveBeenCalledTimes(1); const steered = vi.fn(); - deferDiscardUntilPromptSettles(events as never, 'msg_3', steered); + const steeredTracker = watchPromptSettlements(events as never); + steeredTracker.settle('msg_3', steered); handlers[2]!({ type: 'prompt.steered', promptIds: ['msg_3'], activePromptId: 'msg_parent' }); expect(steered).not.toHaveBeenCalled(); handlers[2]!({ type: 'prompt.completed', promptId: 'msg_other' }); @@ -451,11 +455,11 @@ describe('server-v2 /api/v1 prompts', () => { handlers[2]!({ type: 'prompt.completed', promptId: 'msg_parent' }); expect(steered).toHaveBeenCalledTimes(1); - const unrelatedSteer = vi.fn(); - deferDiscardUntilPromptSettles(events as never, 'msg_4', unrelatedSteer); - handlers[3]!({ type: 'prompt.steered', promptIds: ['msg_other'], activePromptId: 'msg_parent' }); - handlers[3]!({ type: 'prompt.completed', promptId: 'msg_parent' }); - expect(unrelatedSteer).not.toHaveBeenCalled(); + const aborted = vi.fn(); + const abortedTracker = watchPromptSettlements(events as never); + abortedTracker.settle('msg_4', aborted); + handlers[3]!({ type: 'prompt.aborted', promptId: 'msg_4' }); + expect(aborted).toHaveBeenCalledTimes(1); }); it('makes the first three REST prompts available to title generation', async () => { From 8305710b6f4ff4c9b92adba26c5c844beced75d8 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:14:31 +0800 Subject: [PATCH 10/19] fix(kap-server): scope settlement tracking to the owning agent and dispose on rejection - The tracker now subscribes through the agent-scoped IEventBus instead of the App-scoped IEventService: prompt lifecycle events from other sessions never reach it, so a colliding client-chosen prompt id cannot trigger a foreign settlement (and the steer re-target only follows this agent's parent). - A bundled submission that rejects after the tracker was installed now disposes it on the error path instead of leaking a permanent listener. --- packages/kap-server/src/routes/prompts.ts | 30 ++++++++++++++++++----- packages/kap-server/test/prompts.test.ts | 9 +++++++ 2 files changed, 33 insertions(+), 6 deletions(-) diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 8946a7660f9..842b92e881e 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -16,6 +16,7 @@ import { IAgentPromptService, IAgentSkillService, IAuthSummaryService, + IEventBus, IEventService, IFileService, ISessionMediaStore, @@ -37,6 +38,7 @@ import { ErrorCodes, sessionMediaOriginalsDir, type ISessionScopeHandle, + type PromptWithSkillsResult, type Scope, } from '@moonshot-ai/agent-core-v2'; import { ErrorCode } from '../protocol/error-codes'; @@ -123,7 +125,10 @@ async function resolvePromptFromSession(session: ISessionScopeHandle, agentId?: return { prompt: agent.accessor.get(IAgentPromptService), skill: agent.accessor.get(IAgentSkillService), - events: agent.accessor.get(IEventService), + // Agent-scoped bus: prompt lifecycle events of OTHER sessions/agents do + // not arrive here (the App-scoped IEventService would leak them, and + // client-chosen prompt ids can collide across sessions). + events: agent.accessor.get(IEventBus), auth: agent.accessor.get(IAuthSummaryService), profile: agent.accessor.get(IAgentProfileService), toolPolicy: agent.accessor.get(IAgentToolPolicyService), @@ -350,10 +355,16 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { // single user message (rendered skill blocks first, then the // caller's parts). It owns the prompt-metadata update for the main // agent, so this edge skips its own to avoid a double write. - const result = await resolved.skill.promptWithSkills({ - input: parts, - skills: req.body.skills, - }); + let result: PromptWithSkillsResult; + try { + result = await resolved.skill.promptWithSkills({ + input: parts, + skills: req.body.skills, + }); + } catch (error) { + settlement.dispose(); + throw error; + } enqueued = true; settlement.settle(result.prompt_id, () => preparedMedia?.discard()); reply.send( @@ -528,8 +539,9 @@ export function projectPromptSnapshot(prompt: PromptQueueSnapshot['pending'][num * exceptionally fast launch can settle just as early — both land in the * buffered `settledIds` instead of racing past a late subscription. */ -export function watchPromptSettlements(events: IEventService): { +export function watchPromptSettlements(events: IEventBus): { settle(promptId: string, discard: () => void | Promise): void; + dispose(): void; } { const settledIds = new Set(); const parentOf = new Map(); @@ -570,6 +582,12 @@ export function watchPromptSettlements(events: IEventService): { } armed = { id: promptId, discard }; }, + // A submission that rejects after the tracker was installed never calls + // `settle`; the route's error path disposes the tracker here. + dispose(): void { + armed = undefined; + subscription.dispose(); + }, }; } diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 138f2b1a3c3..910f4a993af 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -460,6 +460,15 @@ describe('server-v2 /api/v1 prompts', () => { abortedTracker.settle('msg_4', aborted); handlers[3]!({ type: 'prompt.aborted', promptId: 'msg_4' }); expect(aborted).toHaveBeenCalledTimes(1); + + // A submission that rejects never calls settle: the route disposes the + // tracker, and later events must not fire the discard. + const rejected = vi.fn(); + const rejectedTracker = watchPromptSettlements(events as never); + rejectedTracker.settle('msg_5', rejected); + rejectedTracker.dispose(); + handlers[4]!({ type: 'prompt.completed', promptId: 'msg_5' }); + expect(rejected).not.toHaveBeenCalled(); }); it('makes the first three REST prompts available to title generation', async () => { From bbbaa60faf0362c22c453a3653995b1e68ba97b4 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:31:35 +0800 Subject: [PATCH 11/19] fix(agent-core-v2): keep steered prompts queued until their media intake finishes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Materializing a steered prompt's daemon-ref media awaits a file copy during which the active turn may finish. Records are now spliced out of the queue only after that copy completes, and when the turn is gone by enqueue time they are restored to pending so startNext can launch them as fresh prompts — their handles always launch or settle. --- .../agent-core-v2/src/agent/prompt/promptService.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 2de9dcdc219..6053ff26dc9 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -317,19 +317,28 @@ export class AgentPromptService implements IAgentPromptService { throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'one or more prompts are not pending'); } const selected = this.pending.filter((item) => ids.has(item.id)); - for (const item of selected) this.pending.splice(this.pending.indexOf(item), 1); const message: ContextMessage = { role: 'user', content: selected.flatMap((item) => item.message.content), toolCalls: [], origin: USER_PROMPT_ORIGIN, }; const { message: rerouted, captions } = this.extractCompressionCaptions(message); + // Materialize BEFORE the records leave the queue: the file copy awaits, + // and the active turn may finish meanwhile — records removed earlier + // would never launch or settle in that case. await this.materializeDaemonRefs(rerouted); + for (const item of selected) this.pending.splice(this.pending.indexOf(item), 1); const request = new SteerStepRequest(rerouted, captions, this.reminders, (materialized) => { void this.dispatcher.dispatch( new TurnSteer({ input: materialized.content, origin: materialized.origin ?? USER_PROMPT_ORIGIN }), ); }, () => {}); const turn = (await this.loop.enqueue(request).assigned).turn; - if (turn === undefined) throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); + if (turn === undefined) { + // The active turn finished while the media copy ran: restore the + // records to the queue so startNext can launch them as fresh prompts + // instead of dropping their handles unsettled. + this.pending.unshift(...selected); + throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); + } for (const item of selected) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...selected]); void this.dispatcher.dispatch( From 98309f4b65b15dab5822339cee70d7e93c65bff6 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:40:36 +0800 Subject: [PATCH 12/19] fix(agent-core-v2): revalidate the queue and active turn after steer media intake The daemon-ref copy yields, so settle/abort can consume selected records and the active turn can rotate meanwhile. Only records still pending are steered, and only into the turn that was active at entry; records that vanish from the queue are left to their own launch path, and a missing turn restores them to pending instead of splicing an unrelated tail prompt. The intake/queue-preservation contract is documented in the module header. --- .../src/agent/prompt/promptService.ts | 33 +++++++++++-------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 6053ff26dc9..fcc1c1ad79c 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -4,7 +4,12 @@ * Assigns prompt and message identities, serializes user prompts through an * active slot and FIFO, converts selected pending prompts into active-turn * steers, settles lifecycle handles, and keeps system input outside the prompt - * resource model. `submit` / `submitSteer` are the wire-facing user entry + * resource model. Daemon-backed media in a prompt (normal or steered) is + * materialized into the session media store before the request is queued; a + * steered record leaves the queue only after that copy, revalidated as still + * pending and still facing the same active turn, and is restored when the + * turn vanished meanwhile, so every handle always launches or settles. + * `submit` / `submitSteer` are the wire-facing user entry * points: they track `input_steer` through `telemetry`, persist the derived * title/lastPrompt through `sessionMetadata` for the main agent only * (publishing the live update through `event`), enqueue, and settle @@ -317,15 +322,20 @@ export class AgentPromptService implements IAgentPromptService { throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'one or more prompts are not pending'); } const selected = this.pending.filter((item) => ids.has(item.id)); + const activeAtEntry = this.active; const message: ContextMessage = { role: 'user', content: selected.flatMap((item) => item.message.content), toolCalls: [], origin: USER_PROMPT_ORIGIN, }; const { message: rerouted, captions } = this.extractCompressionCaptions(message); - // Materialize BEFORE the records leave the queue: the file copy awaits, - // and the active turn may finish meanwhile — records removed earlier - // would never launch or settle in that case. await this.materializeDaemonRefs(rerouted); - for (const item of selected) this.pending.splice(this.pending.indexOf(item), 1); + // The copy yields: settle/abort may have consumed some records and the + // active turn may have rotated. Only records still pending steer, and + // only into the turn that was active at entry. + const stillPending = selected.filter((item) => this.pending.includes(item)); + if (stillPending.length === 0 || this.active === undefined || this.active !== activeAtEntry) { + throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); + } + for (const item of stillPending) this.pending.splice(this.pending.indexOf(item), 1); const request = new SteerStepRequest(rerouted, captions, this.reminders, (materialized) => { void this.dispatcher.dispatch( new TurnSteer({ input: materialized.content, origin: materialized.origin ?? USER_PROMPT_ORIGIN }), @@ -333,18 +343,15 @@ export class AgentPromptService implements IAgentPromptService { }, () => {}); const turn = (await this.loop.enqueue(request).assigned).turn; if (turn === undefined) { - // The active turn finished while the media copy ran: restore the - // records to the queue so startNext can launch them as fresh prompts - // instead of dropping their handles unsettled. - this.pending.unshift(...selected); + this.pending.unshift(...stillPending); throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); } - for (const item of selected) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } - this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...selected]); + for (const item of stillPending) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } + this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...stillPending]); void this.dispatcher.dispatch( - new PromptSteered({ activePromptId: this.active.id, promptIds: selected.map((x) => x.id), content: rerouted.content as ContentPart[], steeredAt: new Date().toISOString() }), + new PromptSteered({ activePromptId: this.active.id, promptIds: stillPending.map((x) => x.id), content: rerouted.content as ContentPart[], steeredAt: new Date().toISOString() }), ); - return selected.map((item) => item.handle); + return stillPending.map((item) => item.handle); } abort(promptId: string, reason: Error = userCancellationReason()): boolean { From c5164a66c6d6cbedcc8e32be4ad91b504eee6ef7 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 01:15:04 +0800 Subject: [PATCH 13/19] fix(agent-core-v2): steer only the surviving records and keep their media truthful - The steered content is rebuilt from the records that are still pending after the media intake, so an aborted or concurrently consumed record's text is never injected (or injected twice) alongside the surviving handles. - The enqueue is wrapped so an activeTurnOnly rejection restores the records to pending (the loop throws instead of resolving a missing turn, which made the previous rollback unreachable). - The merged origin now carries the union of every record's bundled skillActivations, and prompt.steered publishes the caller-only content, so the skill instructions reach the model with their metadata intact while the event projection stops leaking internal skill markdown. --- .../src/agent/prompt/promptService.ts | 36 +++++++++++++++---- 1 file changed, 29 insertions(+), 7 deletions(-) diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 69836a07870..83555de5d1f 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -109,6 +109,23 @@ interface Record extends PromptSnapshot { handle: PromptHandle; } +function mergeSteerMessages(records: readonly Record[]): ContextMessage { + const skillActivations = records.flatMap((item) => + item.message.origin?.kind === 'user' ? (item.message.origin.skillActivations ?? []) : [], + ); + return { + role: 'user', + content: records.flatMap((item) => item.message.content), + toolCalls: [], + origin: skillActivations.length === 0 ? USER_PROMPT_ORIGIN : { kind: 'user', skillActivations }, + }; +} + +function stripBundledSkillBlocks(message: ContextMessage): readonly ContentPart[] { + const bundled = message.origin?.kind === 'user' ? (message.origin.skillActivations?.length ?? 0) : 0; + return bundled === 0 ? message.content : message.content.slice(bundled); +} + export const promptLaunchingKey = defineState('prompt.launching', () => false); export class AgentPromptService implements IAgentPromptService { @@ -285,22 +302,27 @@ export class AgentPromptService implements IAgentPromptService { } const selected = this.pending.filter((item) => ids.has(item.id)); const activeAtEntry = this.active; - const message: ContextMessage = { - role: 'user', content: selected.flatMap((item) => item.message.content), toolCalls: [], origin: USER_PROMPT_ORIGIN, - }; - const { message: rerouted, captions } = this.extractCompressionCaptions(message); + const { message: rerouted, captions } = this.extractCompressionCaptions(mergeSteerMessages(selected)); await this.materializeDaemonRefs(rerouted); const stillPending = selected.filter((item) => this.pending.includes(item)); if (stillPending.length === 0 || this.active === undefined || this.active !== activeAtEntry) { throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); } + const steerInput = stillPending.length === selected.length + ? { message: rerouted, captions } + : this.extractCompressionCaptions(mergeSteerMessages(stillPending)); for (const item of stillPending) this.pending.splice(this.pending.indexOf(item), 1); - const request = new SteerStepRequest(rerouted, captions, this.reminders, (materialized) => { + const request = new SteerStepRequest(steerInput.message, steerInput.captions, this.reminders, (materialized) => { void this.dispatcher.dispatch( new TurnSteer({ input: materialized.content, origin: materialized.origin ?? USER_PROMPT_ORIGIN }), ); }, () => {}); - const turn = (await this.loop.enqueue(request).assigned).turn; + let turn: Turn | undefined; + try { + turn = (await this.loop.enqueue(request).assigned).turn; + } catch { + turn = undefined; + } if (turn === undefined) { this.pending.unshift(...stillPending); throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); @@ -308,7 +330,7 @@ export class AgentPromptService implements IAgentPromptService { for (const item of stillPending) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...stillPending]); void this.dispatcher.dispatch( - new PromptSteered({ activePromptId: this.active.id, promptIds: stillPending.map((x) => x.id), content: rerouted.content as ContentPart[], steeredAt: new Date().toISOString() }), + new PromptSteered({ activePromptId: this.active.id, promptIds: stillPending.map((x) => x.id), content: stripBundledSkillBlocks(steerInput.message) as ContentPart[], steeredAt: new Date().toISOString() }), ); return stillPending.map((item) => item.handle); } From a27a8575b755c854eb4f2ff3cb7516a4025332de Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 08:55:57 +0800 Subject: [PATCH 14/19] fix(agent-core-v2): harden steer rollback and register bundled prompt ids --- .../src/agent/prompt/promptService.ts | 13 ++++-- .../src/agent/skill/skillService.ts | 37 ++++++++------- .../test/agent/prompt/promptService.test.ts | 46 ++++++++++++++++++- .../test/agent/skill/activateSkill.test.ts | 17 +++++++ 4 files changed, 91 insertions(+), 22 deletions(-) diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 83555de5d1f..a3e167ee917 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -311,7 +311,12 @@ export class AgentPromptService implements IAgentPromptService { const steerInput = stillPending.length === selected.length ? { message: rerouted, captions } : this.extractCompressionCaptions(mergeSteerMessages(stillPending)); - for (const item of stillPending) this.pending.splice(this.pending.indexOf(item), 1); + const removed: { readonly item: Record; readonly index: number }[] = []; + for (const item of stillPending) { + const index = this.pending.indexOf(item); + removed.push({ item, index }); + this.pending.splice(index, 1); + } const request = new SteerStepRequest(steerInput.message, steerInput.captions, this.reminders, (materialized) => { void this.dispatcher.dispatch( new TurnSteer({ input: materialized.content, origin: materialized.origin ?? USER_PROMPT_ORIGIN }), @@ -323,14 +328,14 @@ export class AgentPromptService implements IAgentPromptService { } catch { turn = undefined; } - if (turn === undefined) { - this.pending.unshift(...stillPending); + if (turn === undefined || this.active !== activeAtEntry) { + for (const { item, index } of removed.reverse()) this.pending.splice(index, 0, item); throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); } for (const item of stillPending) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...stillPending]); void this.dispatcher.dispatch( - new PromptSteered({ activePromptId: this.active.id, promptIds: stillPending.map((x) => x.id), content: stripBundledSkillBlocks(steerInput.message) as ContentPart[], steeredAt: new Date().toISOString() }), + new PromptSteered({ activePromptId: this.active.id, promptIds: stillPending.map((x) => x.id), content: stillPending.flatMap((item) => stripBundledSkillBlocks(item.message)), steeredAt: new Date().toISOString() }), ); return stillPending.map((item) => item.handle); } diff --git a/packages/agent-core-v2/src/agent/skill/skillService.ts b/packages/agent-core-v2/src/agent/skill/skillService.ts index 2474ab9324e..aa25fe55c86 100644 --- a/packages/agent-core-v2/src/agent/skill/skillService.ts +++ b/packages/agent-core-v2/src/agent/skill/skillService.ts @@ -15,7 +15,7 @@ import { ISessionContext } from '#/session/sessionContext/sessionContext'; import { Service } from '#/_base/di/service'; import { ErrorCodes, Error2 } from '#/errors'; import { isUserActivatableSkillType, type SkillDefinition } from '#/app/skillCatalog/types'; -import { IAgentPromptService, type PromptLaunchResult } from '#/agent/prompt/prompt'; +import { IAgentPromptService, reservePrompt, type PromptLaunchResult } from '#/agent/prompt/prompt'; import { ITelemetryService } from '#/app/telemetry/telemetry'; import { IAgentLoopService, type Turn } from '#/agent/loop/loop'; import { IAgentStateService } from '#/agent/state/agentState'; @@ -140,8 +140,9 @@ export class AgentSkillService extends Service implements IAgentSkillService { for (const activation of prepared) { void this.recordActivation(activation.origin); } - const handle = await this.prompt.enqueue({ - message: { + const reservation = reservePrompt(this.prompt); + try { + const handle = await reservation.submit({ role: 'user', content: [...prepared.map((activation) => activation.part), ...input.input], toolCalls: [], @@ -149,21 +150,23 @@ export class AgentSkillService extends Service implements IAgentSkillService { kind: 'user', skillActivations: prepared.map((activation) => activation.entry), }, - }, - }); - if (handle.state === 'pending') { - return { prompt_id: handle.id, created_at: handle.createdAt, state: 'queued' }; - } - const turn = await handle.launched; - if (turn === undefined && handle.state !== 'blocked') { - throw new Error2(ErrorCodes.INTERNAL, 'promptWithSkills failed to launch a turn'); + }); + if (handle.state === 'pending') { + return { prompt_id: handle.id, created_at: handle.createdAt, state: 'queued' }; + } + const turn = await handle.launched; + if (turn === undefined && handle.state !== 'blocked') { + throw new Error2(ErrorCodes.INTERNAL, 'promptWithSkills failed to launch a turn'); + } + return { + turn_id: turn?.id, + prompt_id: handle.id, + created_at: handle.createdAt, + state: handle.state === 'blocked' ? 'blocked' : 'running', + }; + } finally { + reservation.dispose(); } - return { - turn_id: turn?.id, - prompt_id: handle.id, - created_at: handle.createdAt, - state: handle.state === 'blocked' ? 'blocked' : 'running', - }; } recordModelToolActivation(origin: SkillActivationOrigin): void { diff --git a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts index 35aab9a299d..fcc6d902745 100644 --- a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts +++ b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts @@ -8,10 +8,11 @@ import { Event } from '#/_base/event'; import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; import type { ContextMessage } from '#/agent/contextMemory/types'; +import type { ContentPart } from '#/kosong/contract/message'; import { IAgentFullCompactionService } from '#/agent/fullCompaction/fullCompaction'; import { IAgentLoopService } from '#/agent/loop/loop'; import { IAgentPromptService } from '#/agent/prompt/prompt'; -import { AgentPromptService, PromptQueued } from '#/agent/prompt/promptService'; +import { AgentPromptService, PromptQueued, PromptSteered } from '#/agent/prompt/promptService'; import { IAgentScopeContext, makeAgentScopeContext } from '#/agent/scopeContext/scopeContext'; import { IAgentSystemReminderService } from '#/agent/systemReminder/systemReminder'; import { AgentSystemReminderService } from '#/agent/systemReminder/systemReminderService'; @@ -280,4 +281,47 @@ describe('AgentPromptService', () => { expect.objectContaining({ fileId: 'file_1', name: 'pic.png' }), ); }); + + it('publishes each record’s user parts when steering bundled prompts', async () => { + const { prompt, eventBus } = harness(); + const steered: ContentPart[][] = []; + eventBus.subscribe(PromptSteered, (event) => steered.push(event.content)); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const bundled = (skillName: string, user: string): ContextMessage => ({ + role: 'user', + content: [ + { type: 'text', text: `${skillName}` }, + { type: 'text', text: user }, + ], + toolCalls: [], + origin: { kind: 'user', skillActivations: [{ activationId: `act-${skillName}`, skillName }] }, + }); + const one = await prompt.enqueue({ message: bundled('review', 'first user text') }); + const two = await prompt.enqueue({ message: bundled('security', 'second user text') }); + + await prompt.steer([one.id, two.id]); + + expect(steered).toHaveLength(1); + expect(steered[0]).toEqual([ + { type: 'text', text: 'first user text' }, + { type: 'text', text: 'second user text' }, + ]); + }); + + it('restores failed steers to their original queue positions', async () => { + const { prompt, loop } = harness(); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + await prompt.enqueue({ id: 'a', message: message('a') }); + await prompt.enqueue({ id: 'b', message: message('b') }); + await prompt.enqueue({ id: 'c', message: message('c') }); + vi.spyOn(loop, 'enqueue').mockImplementation(() => { + throw new Error('boom'); + }); + + await expect(prompt.steer(['b'])).rejects.toMatchObject({ code: 'prompt.not_found' }); + + expect(prompt.list().pending.map((item) => item.id)).toEqual(['a', 'b', 'c']); + }); }); diff --git a/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts b/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts index 474da5223df..a1b1b9c4d5a 100644 --- a/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts +++ b/packages/agent-core-v2/test/agent/skill/activateSkill.test.ts @@ -180,4 +180,21 @@ describe('promptWithSkills', () => { expect(undone).toBe(1); expect(ctx.context.get()).toHaveLength(0); }); + + it('reserves the bundled prompt id against later prompt_id reuse', async () => { + ctx = agentWithSkills(); + ctx.mockNextResponse({ type: 'text', text: 'done' }); + const launched = await ctx.rpc.promptWithSkills({ + input: [{ type: 'text', text: 'Review this change.' }], + skills: [{ name: 'review' }], + }); + await ctx.untilTurnEnd(); + + await expect( + ctx.rpc.prompt({ + input: [{ type: 'text', text: 'again' }], + promptId: launched.prompt_id, + }), + ).rejects.toThrow(/already in use/i); + }); }); From 95d36799972cb3c0ac1b4c7fafe0ec13ce158a96 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 09:14:27 +0800 Subject: [PATCH 15/19] fix(agent-core-v2): strip bundled blocks from prompt.queued and reject partial steers --- .../src/agent/prompt/promptService.ts | 24 +++--- .../test/agent/prompt/promptService.test.ts | 75 ++++++++++++++++--- 2 files changed, 74 insertions(+), 25 deletions(-) diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index a3e167ee917..0627e7a2989 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -121,7 +121,7 @@ function mergeSteerMessages(records: readonly Record[]): ContextMessage { }; } -function stripBundledSkillBlocks(message: ContextMessage): readonly ContentPart[] { +function stripBundledSkillBlocks(message: ContextMessage): ContentPart[] { const bundled = message.origin?.kind === 'user' ? (message.origin.skillActivations?.length ?? 0) : 0; return bundled === 0 ? message.content : message.content.slice(bundled); } @@ -304,20 +304,16 @@ export class AgentPromptService implements IAgentPromptService { const activeAtEntry = this.active; const { message: rerouted, captions } = this.extractCompressionCaptions(mergeSteerMessages(selected)); await this.materializeDaemonRefs(rerouted); - const stillPending = selected.filter((item) => this.pending.includes(item)); - if (stillPending.length === 0 || this.active === undefined || this.active !== activeAtEntry) { - throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); + if (selected.some((item) => !this.pending.includes(item)) || this.active !== activeAtEntry) { + throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'one or more prompts are no longer pending'); } - const steerInput = stillPending.length === selected.length - ? { message: rerouted, captions } - : this.extractCompressionCaptions(mergeSteerMessages(stillPending)); const removed: { readonly item: Record; readonly index: number }[] = []; - for (const item of stillPending) { + for (const item of selected) { const index = this.pending.indexOf(item); removed.push({ item, index }); this.pending.splice(index, 1); } - const request = new SteerStepRequest(steerInput.message, steerInput.captions, this.reminders, (materialized) => { + const request = new SteerStepRequest(rerouted, captions, this.reminders, (materialized) => { void this.dispatcher.dispatch( new TurnSteer({ input: materialized.content, origin: materialized.origin ?? USER_PROMPT_ORIGIN }), ); @@ -332,12 +328,12 @@ export class AgentPromptService implements IAgentPromptService { for (const { item, index } of removed.reverse()) this.pending.splice(index, 0, item); throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); } - for (const item of stillPending) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } - this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...stillPending]); + for (const item of selected) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } + this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...selected]); void this.dispatcher.dispatch( - new PromptSteered({ activePromptId: this.active.id, promptIds: stillPending.map((x) => x.id), content: stillPending.flatMap((item) => stripBundledSkillBlocks(item.message)), steeredAt: new Date().toISOString() }), + new PromptSteered({ activePromptId: this.active.id, promptIds: selected.map((x) => x.id), content: selected.flatMap((item) => stripBundledSkillBlocks(item.message)), steeredAt: new Date().toISOString() }), ); - return stillPending.map((item) => item.handle); + return selected.map((item) => item.handle); } abort(promptId: string, reason: Error = userCancellationReason()): boolean { @@ -460,7 +456,7 @@ export class AgentPromptService implements IAgentPromptService { private publishCompleted(promptId: string, reason: 'completed' | 'failed' | 'blocked'): void { void this.dispatcher.dispatch(new PromptCompleted({ promptId, finishedAt: new Date().toISOString(), reason })); } private publishQueued(record: Record): void { if ((record.message.origin ?? USER_PROMPT_ORIGIN).kind !== 'user') return; - void this.dispatcher.dispatch(new PromptQueued({ promptId: record.id, content: record.message.content, queueLength: this.pending.length })); + void this.dispatcher.dispatch(new PromptQueued({ promptId: record.id, content: stripBundledSkillBlocks(record.message), queueLength: this.pending.length })); } private publishAborted(promptId: string): void { void this.dispatcher.dispatch(new PromptAborted({ promptId, abortedAt: new Date().toISOString() })); } } diff --git a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts index fcc6d902745..aa5bc2685a0 100644 --- a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts +++ b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts @@ -40,6 +40,15 @@ function message(text: string): ContextMessage { return { role: 'user', content: [{ type: 'text', text }], toolCalls: [], origin: { kind: 'user' } }; } +function bundledMessage(skillName: string, user: string, extra: readonly ContentPart[] = []): ContextMessage { + return { + role: 'user', + content: [{ type: 'text', text: `${skillName}` }, { type: 'text', text: user }, ...extra], + toolCalls: [], + origin: { kind: 'user', skillActivations: [{ activationId: `act-${skillName}`, skillName }] }, + }; +} + const noopBlob: IAgentBlobService = { _serviceBrand: undefined, offloadParts: async (parts) => parts, @@ -288,17 +297,8 @@ describe('AgentPromptService', () => { eventBus.subscribe(PromptSteered, (event) => steered.push(event.content)); const active = await prompt.enqueue({ message: message('active') }); await active.launched; - const bundled = (skillName: string, user: string): ContextMessage => ({ - role: 'user', - content: [ - { type: 'text', text: `${skillName}` }, - { type: 'text', text: user }, - ], - toolCalls: [], - origin: { kind: 'user', skillActivations: [{ activationId: `act-${skillName}`, skillName }] }, - }); - const one = await prompt.enqueue({ message: bundled('review', 'first user text') }); - const two = await prompt.enqueue({ message: bundled('security', 'second user text') }); + const one = await prompt.enqueue({ message: bundledMessage('review', 'first user text') }); + const two = await prompt.enqueue({ message: bundledMessage('security', 'second user text') }); await prompt.steer([one.id, two.id]); @@ -324,4 +324,57 @@ describe('AgentPromptService', () => { expect(prompt.list().pending.map((item) => item.id)).toEqual(['a', 'b', 'c']); }); + + it('publishes only caller parts when a bundled prompt queues', async () => { + const { prompt, eventBus } = harness(); + const queued: Array<{ promptId: string; content: ContentPart[] }> = []; + eventBus.subscribe(PromptQueued, (event) => { + queued.push({ promptId: event.promptId, content: event.content }); + }); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + + await prompt.enqueue({ id: 'bundled', message: bundledMessage('review', 'user text') }); + + expect(queued).toEqual([ + { promptId: 'bundled', content: [{ type: 'text', text: 'user text' }] }, + ]); + }); + + it('rejects the whole steer when a selected prompt is aborted during intake', async () => { + const { prompt, intake } = harness(); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + let releaseIntake!: () => void; + intake.get.mockImplementationOnce( + () => + new Promise((resolve) => { + releaseIntake = () => + resolve({ + meta: { + id: 'file_1', + size: 3, + name: 'pic.png', + media_type: 'image/png', + created_at: '2026-01-01T00:00:00.000Z', + }, + stream: () => Readable.from([new Uint8Array([1, 2, 3])]), + }); + }), + ); + await prompt.enqueue({ + id: 'a', + message: bundledMessage('review', 'a text', [ + { type: 'image_url', imageUrl: { url: 'kimi-file://file_1' } }, + ]), + }); + await prompt.enqueue({ id: 'b', message: message('b') }); + + const steerPromise = prompt.steer(['a', 'b']); + prompt.abort('a'); + releaseIntake(); + + await expect(steerPromise).rejects.toMatchObject({ code: 'prompt.not_found' }); + expect(prompt.list().pending.map((item) => item.id)).toEqual(['b']); + }); }); From ece08d44e0ac09ab28397adaaa70e636cc946470 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 09:33:07 +0800 Subject: [PATCH 16/19] fix(kap-server): update session metadata for bundled prompts routed to subagents --- packages/kap-server/src/routes/prompts.ts | 7 +++++++ packages/kap-server/test/prompts.test.ts | 20 ++++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 61c15a9824d..644e7332cb9 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -282,6 +282,13 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { } const parts = contentToCoreParts(resolvedContent); if (req.body.skills !== undefined) { + if (req.body.agent_id !== undefined && req.body.agent_id !== MAIN_AGENT_ID) { + await applyPromptMetadataUpdate({ + metadata: session.accessor.get(ISessionMetadata), + eventService: core.accessor.get(IEventService), + sessionId: session_id, + }, promptMetadataTextFromContentParts(parts)); + } const settlement = watchPromptSettlements(resolved.events); let result: PromptWithSkillsResult; try { diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 50722fd6dd1..28391f03bb6 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -312,6 +312,26 @@ describe('server-v2 /api/v1 prompts', () => { expect(submitted.body.data.user_message_id).toBe('submission-1'); }); + it('updates session metadata for a bundled prompt routed to a non-main agent', async () => { + const id = await createSession(home as string); + await createMainAgent(id); + + const session = getLiveSessionById(server!.core.accessor, id); + if (session === undefined) throw new Error(`session ${id} not found`); + const child = await session.accessor.get(IAgentLifecycleService).fork('main'); + + const submitted = await call('POST', `/api/v1/sessions/${id}/prompts`, { + content: [{ type: 'text', text: 'bundled side question' }], + agent_id: child.id, + skills: [{ name: 'update-config' }], + }); + expect(submitted.body.code).toBe(0); + + expect((await session.accessor.get(ISessionMetadata).read()).lastPrompt).toBe( + 'bundled side question', + ); + }); + it('rejects a reused prompt_id live and after cold resume without changing metadata', async () => { const id = await createSession(home as string); await createMainAgent(id); From a0e27922fdd6667a785e52137293a05703208982 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 10:02:37 +0800 Subject: [PATCH 17/19] fix(agent-core-v2): restart queue after raced steer rollback and prefix skill blocks in merged steer --- .../src/agent/prompt/promptService.ts | 19 ++++-- .../agent-core-v2/test/agent/loop/stubs.ts | 11 +++- .../test/agent/prompt/promptService.test.ts | 64 ++++++++++++++++++- 3 files changed, 82 insertions(+), 12 deletions(-) diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 0627e7a2989..210cdd42210 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -109,23 +109,29 @@ interface Record extends PromptSnapshot { handle: PromptHandle; } +function bundledSkillBlockCount(message: ContextMessage): number { + return message.origin?.kind === 'user' ? (message.origin.skillActivations?.length ?? 0) : 0; +} + +function stripBundledSkillBlocks(message: ContextMessage): ContentPart[] { + return message.content.slice(bundledSkillBlockCount(message)); +} + function mergeSteerMessages(records: readonly Record[]): ContextMessage { const skillActivations = records.flatMap((item) => item.message.origin?.kind === 'user' ? (item.message.origin.skillActivations ?? []) : [], ); return { role: 'user', - content: records.flatMap((item) => item.message.content), + content: [ + ...records.flatMap((item) => item.message.content.slice(0, bundledSkillBlockCount(item.message))), + ...records.flatMap((item) => stripBundledSkillBlocks(item.message)), + ], toolCalls: [], origin: skillActivations.length === 0 ? USER_PROMPT_ORIGIN : { kind: 'user', skillActivations }, }; } -function stripBundledSkillBlocks(message: ContextMessage): ContentPart[] { - const bundled = message.origin?.kind === 'user' ? (message.origin.skillActivations?.length ?? 0) : 0; - return bundled === 0 ? message.content : message.content.slice(bundled); -} - export const promptLaunchingKey = defineState('prompt.launching', () => false); export class AgentPromptService implements IAgentPromptService { @@ -326,6 +332,7 @@ export class AgentPromptService implements IAgentPromptService { } if (turn === undefined || this.active !== activeAtEntry) { for (const { item, index } of removed.reverse()) this.pending.splice(index, 0, item); + if (this.active === undefined) void this.startNext(); throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'no active turn to steer into'); } for (const item of selected) { item.state = 'steered'; item.launchedDeferred.resolve(turn); } diff --git a/packages/agent-core-v2/test/agent/loop/stubs.ts b/packages/agent-core-v2/test/agent/loop/stubs.ts index a1b12101016..12dc196d6a0 100644 --- a/packages/agent-core-v2/test/agent/loop/stubs.ts +++ b/packages/agent-core-v2/test/agent/loop/stubs.ts @@ -1,6 +1,6 @@ import { toDisposable } from '#/_base/di/lifecycle'; import { Event } from '#/_base/event'; -import type { IAgentLoopService, LoopErrorHandler, LoopErrorHandlerRegistrationOptions, Step, Turn } from '#/agent/loop/loop'; +import type { IAgentLoopService, LoopErrorHandler, LoopErrorHandlerRegistrationOptions, Step, Turn, TurnResult } from '#/agent/loop/loop'; import type { StepRequest } from '#/agent/loop/stepRequest'; import { StepRequestQueue, type StepRequestBatch } from '#/agent/loop/stepRequestQueue'; import type { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor'; @@ -10,12 +10,13 @@ import type { ContextMessage } from '#/agent/contextMemory/types'; import { createHooks } from '#/hooks'; import type { IWireService } from '#/wire/wire'; -export interface StubLoopOptions { readonly hasActiveTurn?: boolean; readonly currentId?: string | number; readonly pendingTurnResult?: boolean } +export interface StubLoopOptions { readonly hasActiveTurn?: boolean; readonly currentId?: string | number; readonly pendingTurnResult?: boolean; readonly manualTurnResult?: boolean } export type StubLoop = IAgentLoopService & { readonly queue: StepRequestQueue; readonly launches: readonly number[]; readonly cancels: readonly { readonly turnId?: number; readonly reason?: unknown }[]; startTurn(): Turn; + settleActive(result?: TurnResult): void; drainNextBatch(context: { append(...messages: ContextMessage[]): void }): StepRequestBatch | undefined; }; const turnControllers = new WeakMap(); @@ -44,14 +45,18 @@ export function stubLoopWithHooks(options: StubLoopOptions = {}): StubLoop { const hooks = createHooks(['onWillBeginStep', 'onDidFinishStep']) as IAgentLoopService['hooks']; const queue = new StepRequestQueue(); const errorHandlers = registry(); const launches: number[] = []; const cancels: { turnId?: number; reason?: unknown }[] = []; let active: Turn | undefined; let nextId = typeof options.currentId === 'number' ? options.currentId : 0; + let releaseActiveResult: ((result: TurnResult) => void) | undefined; const startTurn = () => { const turn = makeTurn(nextId++); - const result = options.pendingTurnResult === true ? new Promise(() => {}) : turn.result; + const result = options.manualTurnResult === true + ? new Promise((resolve) => { releaseActiveResult = resolve; }) + : options.pendingTurnResult === true ? new Promise(() => {}) : turn.result; const configured = { ...turn, result }; launches.push(configured.id); active = configured; return configured; }; const stub: StubLoop = { _serviceBrand: undefined, hooks, queue, launches, cancels, startTurn, + settleActive(result = { type: 'completed', steps: 0, truncated: false }) { releaseActiveResult?.(result); }, enqueue(request, enqueueOptions) { let turn = active; if (request.admission === 'newTurn' || (request.admission === 'activeOrNewTurn' && turn === undefined)) turn = startTurn(); diff --git a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts index aa5bc2685a0..ebc63983bba 100644 --- a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts +++ b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts @@ -33,8 +33,9 @@ import { IFileService } from '#/app/file/fileService'; import { ISessionMediaStore } from '#/agent/media/sessionMediaStore'; import { stubContextMemory } from '../contextMemory/stubs'; -import { stubLoopWithHooks, stubToolExecutor, stubWire } from '../loop/stubs'; +import { stubLoopWithHooks, stubToolExecutor, stubWire, type StubLoopOptions } from '../loop/stubs'; import { registerStateServices } from '../../state/stubs'; +import { SteerStepRequest } from '#/agent/prompt/promptStepRequests'; function message(text: string): ContextMessage { return { role: 'user', content: [{ type: 'text', text }], toolCalls: [], origin: { kind: 'user' } }; @@ -56,11 +57,11 @@ const noopBlob: IAgentBlobService = { isBlobRef: () => false, }; -function harness() { +function harness(loopOptions: StubLoopOptions = { pendingTurnResult: true }) { const disposables = new DisposableStore(); onTestFinished(() => disposables.dispose()); const context = stubContextMemory(); - const loop = stubLoopWithHooks({ pendingTurnResult: true }); + const loop = stubLoopWithHooks(loopOptions); const fullCompaction = { _serviceBrand: undefined, compacting: null, @@ -377,4 +378,61 @@ describe('AgentPromptService', () => { await expect(steerPromise).rejects.toMatchObject({ code: 'prompt.not_found' }); expect(prompt.list().pending.map((item) => item.id)).toEqual(['b']); }); + + it('keeps bundled skill blocks at the merged message prefix when steering', async () => { + const { prompt, context, loop } = harness(); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const one = await prompt.enqueue({ message: bundledMessage('review', 'user A') }); + const two = await prompt.enqueue({ message: bundledMessage('security', 'user B') }); + + await prompt.steer([one.id, two.id]); + loop.drainNextBatch(context); + + const merged = context + .get() + .find( + (entry) => entry.origin?.kind === 'user' && entry.origin.skillActivations !== undefined, + ); + expect(merged?.content).toEqual([ + { type: 'text', text: 'review' }, + { type: 'text', text: 'security' }, + { type: 'text', text: 'user A' }, + { type: 'text', text: 'user B' }, + ]); + }); + + it('restarts the queue after restoring a steer raced by the active turn settling', async () => { + const { prompt, loop } = harness({ manualTurnResult: true }); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const queued = await prompt.enqueue({ id: 'queued', message: message('queued') }); + let steerEnqueued!: () => void; + const enqueued = new Promise((resolve) => { + steerEnqueued = resolve; + }); + let rejectSteer!: (reason?: unknown) => void; + const original = loop.enqueue.bind(loop); + vi.spyOn(loop, 'enqueue').mockImplementation((request, options) => { + if (request instanceof SteerStepRequest) { + return { + assigned: new Promise((_, reject) => { + rejectSteer = reject; + steerEnqueued(); + }), + abort: () => true, + }; + } + return original(request, options); + }); + + const steerPromise = prompt.steer([queued.id]); + await enqueued; + loop.settleActive(); + rejectSteer(new Error('held')); + + await expect(steerPromise).rejects.toMatchObject({ code: 'prompt.not_found' }); + await expect(queued.launched).resolves.toBeDefined(); + expect(prompt.list().active?.id).toBe('queued'); + }); }); From 7d7c911067ea24075a3abbfd11bc6086013754d9 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 10:12:27 +0800 Subject: [PATCH 18/19] fix(agent-core-v2): block queue advancement during steer admission --- .../src/agent/prompt/promptService.ts | 6 ++- .../test/agent/prompt/promptService.test.ts | 40 +++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 210cdd42210..9652d9d0840 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -140,6 +140,7 @@ export class AgentPromptService implements IAgentPromptService { private readonly pending: Record[] = []; private readonly steered = new Map(); private readonly reservedPromptIds = new Set(); + private steering = 0; private fullCompactionService: IAgentFullCompactionService | undefined; readonly hooks = { onBeforeSubmitPrompt: new OrderedHookSlot() }; @@ -313,6 +314,7 @@ export class AgentPromptService implements IAgentPromptService { if (selected.some((item) => !this.pending.includes(item)) || this.active !== activeAtEntry) { throw new Error2(ErrorCodes.PROMPT_NOT_FOUND, 'one or more prompts are no longer pending'); } + this.steering++; const removed: { readonly item: Record; readonly index: number }[] = []; for (const item of selected) { const index = this.pending.indexOf(item); @@ -329,6 +331,8 @@ export class AgentPromptService implements IAgentPromptService { turn = (await this.loop.enqueue(request).assigned).turn; } catch { turn = undefined; + } finally { + this.steering--; } if (turn === undefined || this.active !== activeAtEntry) { for (const { item, index } of removed.reverse()) this.pending.splice(index, 0, item); @@ -379,7 +383,7 @@ export class AgentPromptService implements IAgentPromptService { } private async startNext(): Promise { - if (this.active !== undefined || this.launching) return; + if (this.active !== undefined || this.launching || this.steering > 0) return; const item = this.pending.shift(); if (item === undefined) return; this.launching = true; try { diff --git a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts index ebc63983bba..37348152447 100644 --- a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts +++ b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts @@ -435,4 +435,44 @@ describe('AgentPromptService', () => { await expect(queued.launched).resolves.toBeDefined(); expect(prompt.list().active?.id).toBe('queued'); }); + + it('does not advance the queue while a steer assignment is in flight', async () => { + const { prompt, loop } = harness({ manualTurnResult: true }); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const a = await prompt.enqueue({ id: 'a', message: message('a') }); + await prompt.enqueue({ id: 'b', message: message('b') }); + let steerEnqueued!: () => void; + const enqueued = new Promise((resolve) => { + steerEnqueued = resolve; + }); + let rejectSteer!: (reason?: unknown) => void; + const original = loop.enqueue.bind(loop); + vi.spyOn(loop, 'enqueue').mockImplementation((request, options) => { + if (request instanceof SteerStepRequest) { + return { + assigned: new Promise((_, reject) => { + rejectSteer = reject; + steerEnqueued(); + }), + abort: () => true, + }; + } + return original(request, options); + }); + + const steerPromise = prompt.steer([a.id]); + await enqueued; + loop.settleActive(); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(loop.launches).toHaveLength(1); + rejectSteer(new Error('held')); + + await expect(steerPromise).rejects.toMatchObject({ code: 'prompt.not_found' }); + await expect(a.launched).resolves.toBeDefined(); + expect(prompt.list().active?.id).toBe('a'); + expect(prompt.list().pending.map((item) => item.id)).toEqual(['b']); + }); }); From e4965f55e09002fda8fdac561a563af8ade46fb0 Mon Sep 17 00:00:00 2001 From: chengluyu <2239547+chengluyu@users.noreply.github.com> Date: Tue, 18 Aug 2026 13:43:51 +0800 Subject: [PATCH 19/19] chore: drop the changeset for server-only protocol plumbing --- .changeset/kap-prompt-skills.md | 5 ----- 1 file changed, 5 deletions(-) delete mode 100644 .changeset/kap-prompt-skills.md diff --git a/.changeset/kap-prompt-skills.md b/.changeset/kap-prompt-skills.md deleted file mode 100644 index c8ee256c5bc..00000000000 --- a/.changeset/kap-prompt-skills.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"@moonshot-ai/kimi-code": patch ---- - -The session prompt submission API now accepts an optional `skills` field: one or more named skills activate together with the prompt as a single bundled turn (one undo unit), validated up front with zero side effects on rejection.