Merge branch 'main' into feat/skill-group-selector

Author: creatiVision

packages/agent-core-v2/src/agent/prompt/promptService.ts

@@ -109,6 +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.slice(0, bundledSkillBlockCount(item.message))),
+      ...records.flatMap((item) => stripBundledSkillBlocks(item.message)),
+    ],
+    toolCalls: [],
+    origin: skillActivations.length === 0 ? USER_PROMPT_ORIGIN : { kind: 'user', skillActivations },
+  };
+}
+
 export const promptLaunchingKey = defineState<boolean>('prompt.launching', () => false);
 
 export class AgentPromptService implements IAgentPromptService {
@@ -117,6 +140,7 @@ export class AgentPromptService implements IAgentPromptService {
   private readonly pending: Record[] = [];
   private readonly steered = new Map<string, Record[]>();
   private readonly reservedPromptIds = new Set<string>();
+  private steering = 0;
   private fullCompactionService: IAgentFullCompactionService | undefined;
   readonly hooks = { onBeforeSubmitPrompt: new OrderedHookSlot<PromptSubmitContext>() };
 
@@ -284,22 +308,41 @@ 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);
+    const activeAtEntry = this.active;
+    const { message: rerouted, captions } = this.extractCompressionCaptions(mergeSteerMessages(selected));
+    await this.materializeDaemonRefs(rerouted);
+    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);
+      removed.push({ item, index });
+      this.pending.splice(index, 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');
+    let turn: Turn | undefined;
+    try {
+      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);
+      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); }
     this.steered.set(this.active.id, [...(this.steered.get(this.active.id) ?? []), ...selected]);
     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: selected.map((x) => x.id), content: selected.flatMap((item) => stripBundledSkillBlocks(item.message)), steeredAt: new Date().toISOString() }),
     );
     return selected.map((item) => item.handle);
   }
@@ -322,6 +365,7 @@ export class AgentPromptService implements IAgentPromptService {
 
   async inject(message: ContextMessage): Promise<Turn | undefined> {
     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 }),
@@ -339,17 +383,13 @@ export class AgentPromptService implements IAgentPromptService {
   }
 
   private async startNext(): Promise<void> {
-    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 {
       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' });
@@ -381,6 +421,13 @@ export class AgentPromptService implements IAgentPromptService {
     void this.startNext();
   }
 
+  private async materializeDaemonRefs(message: ContextMessage): Promise<void> {
+    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<boolean> {
     const ctx = { promptMessage, isSteer, block: false }; await this.hooks.onBeforeSubmitPrompt.run(ctx); return ctx.block;
   }
@@ -420,7 +467,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() })); }
 }

packages/agent-core-v2/src/agent/skill/skill.ts

@@ -19,11 +19,18 @@ export interface PromptWithSkillsInput {
   readonly skills: readonly PromptSkillActivation[];
 }
 
+export interface PromptWithSkillsResult {
+  readonly turn_id?: number;
+  readonly prompt_id: string;
+  readonly created_at: string;
+  readonly state: 'running' | 'queued' | 'blocked';
+}
+
 export interface IAgentSkillService {
   readonly _serviceBrand: undefined;
 
   activate(input: SkillActivationInput): Promise<PromptLaunchResult>;
-  promptWithSkills(input: PromptWithSkillsInput): Promise<PromptLaunchResult | undefined>;
+  promptWithSkills(input: PromptWithSkillsInput): Promise<PromptWithSkillsResult>;
   recordModelToolActivation(origin: SkillActivationOrigin): void;
 }
 

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';
@@ -24,6 +24,7 @@ import {
   IAgentSkillService,
   type PromptSkillActivation,
   type PromptWithSkillsInput,
+  type PromptWithSkillsResult,
   type SkillActivationInput,
 } from './skill';
 import { SkillActivate, skillKey } from './skillOps';
@@ -114,7 +115,7 @@ export class AgentSkillService extends Service implements IAgentSkillService {
     return { turn_id: turn.id };
   }
 
-  async promptWithSkills(input: PromptWithSkillsInput): Promise<PromptLaunchResult | undefined> {
+  async promptWithSkills(input: PromptWithSkillsInput): Promise<PromptWithSkillsResult> {
     if (input.input.length === 0) {
       throw new Error2(ErrorCodes.REQUEST_INVALID, 'promptWithSkills requires a non-empty prompt');
     }
@@ -139,20 +140,33 @@ 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: [],
         origin: {
           kind: 'user',
           skillActivations: prepared.map((activation) => activation.entry),
         },
-      },
-    });
-    if (handle.state === 'pending') return undefined;
-    const turn = await handle.launched;
-    return turn === undefined ? undefined : { turn_id: turn.id };
+      });
+      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();
+    }
   }
 
   recordModelToolActivation(origin: SkillActivationOrigin): void {

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<Turn, AbortController>();
@@ -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<never>(() => {}) : turn.result;
+    const result = options.manualTurnResult === true
+      ? new Promise<TurnResult>((resolve) => { releaseActiveResult = resolve; })
+      : options.pendingTurnResult === true ? new Promise<never>(() => {}) : 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();