工具、输出与记录

Tool 使用本地离线模型选择已登记工具并消费真实反馈。输出订阅交付业务内容，记录安装保存调用事实；调用退出探针核对关闭故障与占用释放。

先执行 pnpm build，再执行下列命令。所有示例均在本机 Node.js 中运行。

让模型选择工具并依据真实反馈回答

跟随两轮离线请求：第一轮选择翻译节点，第二轮读取工具反馈形成回答。运行后得到「你好」，断言核对请求恰好两次，且第二次请求包含节点实际产生的反馈。

node examples/tool.mjs

import assert from 'node:assert/strict';
import { pathToFileURL } from 'node:url';
import { z } from 'zod';
import { Resource } from '@weiliang0121/capix-resources';
import { Scene, Node } from '@weiliang0121/capix-structure';
import { Runner, Resources } from '@weiliang0121/capix-runtime';
import { Interactions } from '@weiliang0121/capix-interaction';
import { requireModelRequest, requireModelReply } from '@weiliang0121/capix-model-provider';
import { ModelExchanges, ModelExchangeResource } from '@weiliang0121/capix-model-exchanges';
import { Tool, ToolModelResource } from '@weiliang0121/capix-tool';

// Node 仍是普通处理结构；Tool 从同一份登记生成目录并组织成员调用。
export function createToolRunner(provider) {
  const translate = new Node({ run: (work) => ({ hello: '你好' })[work.input.word] ?? work.input.word });
  const tool = new Tool({
    policyVersion: 'dictionary.v1',
    tools: {
      translate: {
        node: translate,
        description: 'Translate a word into Chinese.',
        input: z.strictObject({ word: z.string() }),
        result: { succeeded: (text) => ({ status: 'ok', content: [{ kind: 'text', text }] }) },
      },
    },
  });

  // 宿主登记独立交互；取消一次调用的等待，不会取消这个发送者。
  const contract = { key: 'example.tool.model.v1', request: requireModelRequest, reply: requireModelReply };
  const interactions = new Interactions().register(contract, { send: (request) => provider.complete(request) });
  const exchanges = new ModelExchanges();
  const resources = new Resources().register(new Resource(exchanges), ModelExchangeResource).register(new Resource({ exchanges, channel: interactions.bind(contract), ...(provider.inspectRequest ? { inspection: provider } : {}) }), ToolModelResource);
  return new Runner(new Scene(tool), resources, interactions).build();
}

// 离线协议示例：第一次请求选择工具，第二次使用真实工具反馈回答。
// 换成配置好的 ChatCompletionsProvider 后，Tool、节点和资源登记不变。
const requests = [];
const offlineModel = {
  async complete(request) {
    requests.push(request);
    const feedback = request.exchanges?.at(-1)?.feedback[0];
    return {
      formatVersion: 1,
      id: feedback ? 'answer' : 'choose-tool',
      model: 'offline-example',
      output: feedback ? feedback.content : [{ kind: 'tool-call', id: 'lookup', name: 'translate', arguments: { encoding: 'json-text', text: '{"word":"hello"}' } }],
      termination: { kind: 'completed', reason: feedback ? 'stop' : 'tool_calls' },
      usage: null,
      native: { format: 'chat-completions', version: 1 },
    };
  },
};

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  const runner = createToolRunner(offlineModel);
  const result = await runner.run({ request: { messages: [{ role: 'user', content: 'Translate hello.' }] } });
  assert.equal(result, '你好');
  assert.equal(requests.length, 2);
  assert.equal(requests[0].exchanges?.length ?? 0, 0);
  assert.deepEqual(requests[1].exchanges.at(-1).feedback[0].content, [{ kind: 'text', text: '你好' }]);
  assert.equal(runner.isRunning, false);
  console.log(result);
}

跨多次运行消费输出，再解除订阅

同一个输出订阅接收前两次调用的逐字增量，解除后第三次调用仍能返回最终值。运行后核对事件中的调用身份各不相同、解除订阅后不再新增事件，以及空闲取消返回 false。

node examples/runner-output.mjs

import assert from 'node:assert/strict';
import { Scene } from '@weiliang0121/capix-structure';
import { Resources, Runner } from '@weiliang0121/capix-runtime';

function text(value) {
  if (typeof value !== 'string') throw new TypeError('Expected text');
  return value;
}

const scene = new Scene({
  output: { result: text, delta: text },
  run(work) {
    for (const character of work.input) work.output.emit(character);
    return work.input;
  },
});
const runner = new Runner(scene, new Resources()).build();

// 订阅属于 Runner，每条事件保留本次根调用的 contextId。
const events = [];
const unsubscribe = runner.output.subscribe({
  receive(event) {
    events.push(event);
    if (event.kind === 'delta') process.stdout.write(event.delta);
  },
});

assert.equal(await runner.run('第一轮输出\n'), '第一轮输出\n');
assert.equal(await runner.run('第二轮仍使用同一个订阅\n'), '第二轮仍使用同一个订阅\n');
unsubscribe();

// 停止消费不影响执行；最终结果仍从本次 Promise 取得。
const eventCount = events.length;
const final = await runner.run('解除订阅后的最终结果');
assert.equal(final, '解除订阅后的最终结果');
assert.equal(events.length, eventCount);
assert.equal(
  events
    .filter((event) => event.kind === 'delta')
    .map((event) => event.delta)
    .join(''),
  '第一轮输出\n第二轮仍使用同一个订阅\n',
);
assert.deepEqual(
  events.filter((event) => event.kind === 'completed').map((event) => event.result),
  ['第一轮输出\n', '第二轮仍使用同一个订阅\n'],
);
assert.equal(new Set(events.map((event) => event.entryId)).size, 2);
assert.equal(new Set(events.map((event) => event.contextId)).size, 2);
assert.equal(runner.isRunning, false);
console.log(final);

// 宿主需要停止当前运行时调用 runner.cancel()；空闲时不会影响下一轮。
const cancelled = runner.cancel();
assert.equal(cancelled, false);
console.log('空闲取消：', cancelled);

同时写业务记录与记录真实调用

节点通过资源写一条自定义记录，可选的 bindRecording 安装另行记录实际调用。运行后核对自定义正文与输入一致，并看到 Scene 和 Node 两条成功调用记录；本例没有配置磁盘持久化。

node examples/records.mjs

import assert from 'node:assert/strict';
import { RecordBook, Recorder, RecordResource } from '@weiliang0121/capix-records';
import { Resource } from '@weiliang0121/capix-resources';
import { Node, Scene } from '@weiliang0121/capix-structure';
import { Resources, Runner } from '@weiliang0121/capix-runtime';
import { bindRecording } from '@weiliang0121/capix-runtime/recording';

// Records live independently; resources grant specific operations on this entity.
const recorder = new Recorder(new RecordBook());
const resources = new Resources().register(new Resource(recorder), RecordResource.write);
const note = new Node({
  deps: { records: RecordResource.write.request() },
  run: (work) => work.resources.records.create({ message: work.input }),
});

// This optional preset records actual calls in the same entity through its own windows.
const runner = new Runner(new Scene(note), resources, undefined, { observations: [bindRecording(recorder)] }).build();
const result = await runner.run('记录是一种可通过资源访问的实体');
console.log(result.kind, result.data.value);
console.log('已结束调用：', [...recorder.book.values()].filter((record) => record.kind === 'invocation' && record.status === 'succeeded').length);
console.log('记录自身写入的资源事实：', [...recorder.book.values()].filter((record) => record.kind === 'resource' && record.access === 'records.construct').length);

assert.equal(result.kind, 'custom');
assert.deepEqual(result.data.value, { message: '记录是一种可通过资源访问的实体' });
assert.equal([...recorder.book.values()].filter((record) => record.kind === 'invocation' && record.status === 'succeeded').length, 2);
assert.equal([...recorder.book.values()].filter((record) => record.kind === 'custom').length, 1);
assert.equal(runner.isRunning, false);

核对关闭故障与占用释放

这是一组诊断探针：主动制造正文与清理错误，检查两者都被保留，并核对 Runner 占用及本地 CP 占用释放。运行后 JSON 中的 checks 应全部为 true；低层身份复用只用于检查占用，不是业务恢复写法，也没有保存可恢复的 point。

node examples/invocation-exit.mjs

// Diagnostic probes, not a recommended application pattern. No real model or disk store.
// `node examples/invocation-exit.mjs` exits nonzero while an exit guarantee is unmet.
import { Resource, ResourceOffer, ResourceSupply } from '@weiliang0121/capix-resources';
import { CallableContext, Invocation, InvocationResources, invocationResourceFailures } from '@weiliang0121/capix-invocation';
import { MemoryCheckpointStore } from '@weiliang0121/capix-checkpoint';
import { Scene, Resources, Runner } from '@weiliang0121/capix-runtime';

function deferred() {
  let resolve;
  const promise = new Promise((done) => {
    resolve = done;
  });
  return { promise, resolve };
}

function contains(error, expected, seen = new Set()) {
  if (error === expected) return true;
  if (!(error instanceof Error) || seen.has(error)) return false;
  seen.add(error);
  return contains(error.cause, expected, seen) || (error instanceof AggregateError && error.errors.some((item) => contains(item, expected, seen)));
}

function pendingResource(cleanupFailure) {
  const response = deferred();
  const offer = new ResourceOffer({
    methods: {
      read(_origin, _input, operation) {
        operation.onClose(() => {
          throw cleanupFailure;
        });
        return response.promise;
      },
    },
  });
  let pending;
  return {
    offer,
    start(endpoint) {
      pending = endpoint.read();
      void pending.catch(() => undefined); // Probe owns this intentionally unawaited operation.
    },
    async finish() {
      response.resolve('late resource result');
      await pending?.catch(() => undefined);
    },
  };
}

async function failurePreservation() {
  const bodyFailure = new Error('business body failed');
  const cleanupFailure = new Error('resource cleanup failed');
  const pending = pendingResource(cleanupFailure);
  const scene = new Scene({
    deps: { data: pending.offer.request() },
    run(work) {
      pending.start(work.resources.data);
      throw bodyFailure;
    },
  });
  const runner = new Runner(scene, new Resources().register(new Resource(null), pending.offer)).build();
  let delivered;
  try {
    await runner.run(null);
  } catch (error) {
    delivered = error;
  } finally {
    await pending.finish();
  }
  return {
    probe: 'body failure plus cleanup failure',
    checks: { bodyFailurePreserved: contains(delivered, bodyFailure), cleanupFailurePreserved: contains(delivered, cleanupFailure), runnerReleased: !runner.isRunning },
    observed: { delivered: delivered?.message, cause: delivered?.cause?.message },
  };
}

async function checkpointRelease() {
  const cleanupFailure = new Error('resource cleanup failed');
  const pending = pendingResource(cleanupFailure);
  const deps = { data: pending.offer.request() };
  const supply = ResourceSupply.capture([pending.offer.provide(() => null)], undefined);
  const resources = new InvocationResources(deps, supply.entries);
  const context = new CallableContext(null, 'node', {}, resources);
  const store = new MemoryCheckpointStore();
  const call = new Invocation({
    context,
    signal: context.signal,
    checkpoints: { store },
    work: {
      execute(current) {
        pending.start(current.resources.values.data);
        return 'body result';
      },
    },
  });
  resources.attach({ contextId: context.id, signal: context.signal, failures: invocationResourceFailures, validate: () => call.requireActive(), error: (error, allow) => call.reject(error, allow) });
  let delivered;
  let nextResult;
  let nextError;
  try {
    try {
      await call.run();
    } catch (error) {
      delivered = error;
    }
    // Diagnostic only: reuse the Context ID to query local exclusion, not to resume business.
    // No CP point was written, so this probe does not claim a recoverable execution exists.
    const nextContext = new CallableContext(null, 'node');
    nextContext.installIdentity(context.id);
    const next = new Invocation({ context: nextContext, signal: nextContext.signal, checkpoints: { store }, work: { execute: () => 'second result' } });
    try {
      nextResult = await next.run();
    } catch (error) {
      nextError = error;
    }
  } finally {
    await pending.finish();
  }
  return {
    probe: 'low-level cleanup and checkpoint claim release',
    checks: { cleanupFailurePreserved: contains(delivered, cleanupFailure), checkpointClaimReleased: nextResult === 'second result' },
    observed: { delivered: delivered?.message, contextStatus: context.status, invocationFailureRecorded: !!call.failure, nextError: nextError?.message },
  };
}

const probes = [await failurePreservation(), await checkpointRelease()];
console.log(JSON.stringify(probes, null, 2));
if (probes.some((probe) => Object.values(probe.checks).some((passed) => !passed))) process.exitCode = 1;