HologramAgent-DSH-Memory-Port/lib/loop/tool-calls.js

/**
 * Schedules one assistant step's tool calls. Exclusive calls form barriers;
 * parallel calls use a bounded rolling pool and are reclassified before start.
 * Dispatch may overlap, while policy, results, and result context remain
 * model-ordered. Abort or an internal scheduler failure stops replenishment
 * and drains started calls.
 *
 * Reproduced from `dsh-agent-loop/lib/types/tool-calls.js:474-708`
 * (0.2.1-rc.2, MIT); the loop replacement deliberately changes no scheduling
 * behaviour, so this file is a port rather than a rewrite.
 *
 * ★ One substitution: the host reads its parallelism cap from
 * `ctx.agentLoop.config.maxParallelToolCalls` (dsh-agent-loop/lib/index.js:553),
 * a service this plugin does not provide (its only consumer is the Web settings
 * panel, which is out of scope). The caller passes the same value explicitly
 * through {@link ToolCallSchedulerOptions.maxParallelToolCalls} instead; the
 * cap is still a live getter, so a dynamic override keeps working.
 *
 * @module dsh-memory/lib/loop/tool-calls
 */

import { createToolResultMessage } from '@deepseek-ai/dsh-llm'
import { TOOL_ABORTED_BEFORE_DISPATCH, TOOL_RUNTIME_SCHEDULER } from '@deepseek-ai/dsh-tools'

/**
 * Throw on a value the compiler proved unreachable.
 *
 * Inlined rather than imported from `@deepseek-ai/dsh-util-values`, and that is a
 * G2b decision, not an oversight: `dsh-util-values` is a *dependency* of
 * `dsh-agent-loop`, not one of its peers (dsh-agent-loop/package.json). Adding it
 * here would therefore introduce a runtime dependency this plugin carries itself
 * — exactly what G2b exists to catch — and it would break H5 (no new deps) for a
 * three-line helper. The four lines below are the upstream body; the message
 * prefix stays identical so a log line still names the same call site.
 *
 * @param {never} value the value that cannot occur
 * @param {string} [context] name of the switch that produced it
 * @throws {Error} always
 */
function assertNever(value, context) {
  const rendered = JSON.stringify(value) ?? String(value)
  throw new Error(`unreachable variant${context ? ` in ${context}` : ''}: ${rendered}`)
}

/**
 * Schedule one assistant step's tool calls by their live concurrency mode.
 *
 * Ordinary completion and abort commit started-call results in order. Abort
 * drains them, records synthetic results for unstarted calls, and returns with
 * the signal still aborted after accepting started-call context through the
 * caller-supplied acceptor.
 *
 * @param {object} ctx loop context that owns the tool registry and carries the initiating Agent
 * @param {number} turn current turn number
 * @param {number} step current step number
 * @param {Array<object>} toolCalls assistant calls in model order
 * @param {AbortSignal} signal abort signal shared by the step
 * @param {(context: object) => void} acceptContext accepts committed result context for the next step
 * @param {{maxParallelToolCalls: () => number}} options the live parallelism cap
 * @returns {Promise<{concluded: boolean}>}
 */
export async function executeToolCalls(ctx, turn, step, toolCalls, signal, acceptContext, options) {
  const agent = ctx.agents.requireInitiator()
  const { session } = agent
  const planned = toolCalls.map((block) => ({
    block,
    exec: { callId: block.id, name: block.name, arguments: parseArguments(block.arguments), agent, signal },
  }))
  let next = 0
  let concluded = false
  while (next < planned.length) {
    const first = planned[next]
    const mode = ctx.tools.executionMode(first.exec).kind
    const outcome = await runGroup(
      ctx,
      turn,
      step,
      mode === 'parallel' ? planned.slice(next) : [first],
      mode,
      signal,
      acceptContext,
      options,
    )
    next += outcome.consumed
    concluded ||= outcome.concluded
    if (outcome.aborted) {
      for (const call of planned.slice(next)) appendSkippedToolCall(session, turn, step, call.block)
      return { concluded }
    }
  }
  return { concluded }
}

/** Parse model arguments, preserving invalid JSON as text and mapping empty input to `{}`. */
function parseArguments(raw) {
  try {
    return raw ? JSON.parse(raw) : {}
  } catch {
    return raw
  }
}

/**
 * Run one exclusive barrier or parallel pool.
 *
 * @param {object} ctx
 * @param {number} turn
 * @param {number} step
 * @param {Array<{block: object, exec: object}>} group
 * @param {'parallel'|'exclusive'} mode
 * @param {AbortSignal} signal
 * @param {(context: object) => void} acceptContext
 * @param {{maxParallelToolCalls: () => number}} options
 */
async function runGroup(ctx, turn, step, group, mode, signal, acceptContext, options) {
  const { session } = ctx.agents.requireInitiator()
  const maxParallelToolCalls = options.maxParallelToolCalls()
  const slots = group.map(() => undefined)
  const callSeqs = group.map(() => undefined)
  let nextToStart = 0
  let committed = 0
  let started = 0
  let aborted = signal.aborted
  let concluded = false
  let schedulerFailure
  const throwSchedulerFailure = () => {
    if (schedulerFailure !== undefined) throw schedulerFailure.error
  }
  const commitReady = async () => {
    while (committed < group.length) {
      const slot = slots[committed]
      if (slot === undefined) break
      const call = group[committed]
      const result = slot.needsPost
        ? await ctx.tools[TOOL_RUNTIME_SCHEDULER].finalize(slot.exec, slot.result)
        : ctx.tools[TOOL_RUNTIME_SCHEDULER].finish(slot.exec, slot.result)
      appendToolResult(session, turn, step, call.block, result, callSeqs[committed])
      for (const context of result.additionalContexts ?? []) acceptContext(context)
      concluded ||= result.concludesTurn === true
      committed++
    }
  }
  const inFlight = new Map()
  const startCall = async (index) => {
    const call = group[index]
    callSeqs[index] = appendToolCall(session, turn, step, call.block)
    started++
    const prepared = await ctx.tools[TOOL_RUNTIME_SCHEDULER].prepare(call.exec)
    throwSchedulerFailure()
    switch (prepared.kind) {
      case 'dispatch': {
        const promise = ctx.tools[TOOL_RUNTIME_SCHEDULER].dispatch(prepared.exec).then(
          (outcome) => {
            slots[index] = { exec: prepared.exec, result: outcome.result, needsPost: outcome.kind === 'post-result' }
            return index
          },
          (error) => {
            schedulerFailure ??= { error }
            return index
          },
        )
        inFlight.set(index, promise)
        break
      }
      case 'post-result':
        slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: true }
        break
      case 'final-result':
        slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: false }
        break
      default:
        assertNever(prepared, 'tool-call scheduler prepare result')
    }
  }
  const fillPool = async () => {
    while (!aborted && nextToStart < group.length && inFlight.size < maxParallelToolCalls) {
      const nextCall = group[nextToStart]
      if (nextToStart > 0 && mode === 'parallel' && ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break
      await startCall(nextToStart)
      nextToStart++
      throwSchedulerFailure()
      await commitReady()
      throwSchedulerFailure()
      if (signal.aborted) aborted = true
    }
  }
  try {
    await fillPool()
    while (inFlight.size > 0) {
      const settledIndex = await Promise.race(inFlight.values())
      inFlight.delete(settledIndex)
      throwSchedulerFailure()
      await commitReady()
      throwSchedulerFailure()
      if (signal.aborted) aborted = true
      await fillPool()
    }
  } catch (error) {
    schedulerFailure ??= { error }
    await Promise.allSettled(inFlight.values())
    throw schedulerFailure.error
  }
  if (aborted) {
    for (const call of group.slice(started)) appendSkippedToolCall(session, turn, step, call.block)
    return { consumed: group.length, aborted: true, concluded }
  }
  if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')
  return { consumed: started, aborted: false, concluded }
}

/** Append the durable call/result pair for a model call skipped after cancellation. */
function appendSkippedToolCall(session, turn, step, block) {
  const callSeq = appendToolCall(session, turn, step, block)
  appendToolResult(
    session,
    turn,
    step,
    block,
    {
      content: [{ type: 'text', text: 'Error: tool call aborted before dispatch' }],
      isError: true,
      error: {
        message: 'tool call aborted before dispatch',
        info: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
      },
    },
    callSeq,
  )
}

/** Append a started call and return the event seq that its result must cite. */
function appendToolCall(session, turn, step, block) {
  return session.append('tool/call', { turn, step, callId: block.id, name: block.name, arguments: block.arguments }).seq
}

/** Append a model-ordered result linked to its call event. */
function appendToolResult(session, turn, step, block, result, callSeq) {
  const message = createToolResultMessage({
    callId: block.id,
    content: result.content,
    isError: result.isError,
  })
  session.append(
    'tool/result',
    {
      turn,
      step,
      message,
      ...(result.error?.info ? { error: result.error.info } : {}),
      ...(result.meta !== undefined ? { meta: result.meta } : {}),
    },
    { surfaceOp: 'append', sourceEventSeqs: [callSeq] },
  )
}