工作流与单Agent

import assert from 'node:assert/strict';
import { createHash } from 'node:crypto';
import { setTimeout as sleep } from 'node:timers/promises';

const session = { userId: 'u1' };
const ticket = { id: 'T1', owner: 'u1', notes: [] };
const state = { taskId: 'task-1', steps: 0, facts: null, written: false,
  activeRemainingMs: 1000 };
const operations = new Map();
const writeAction = { name: 'append_note', ticketId: 'T1', text: '已核对，待客服跟进。' };
const digest = action => createHash('sha256').update(JSON.stringify(action)).digest('hex');

async function withTimeout(fn, milliseconds) {
  if (!Number.isFinite(milliseconds) || milliseconds <= 0) throw new Error('活动时间耗尽');
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(new Error('步骤超时')), milliseconds);
  try { return await fn(controller.signal); }
  finally { clearTimeout(timer); }
}

async function mockDecide(signal) {
  await sleep(2, undefined, { signal }); // 此模拟任务实际响应取消信号。
  if (!state.facts) return { name: 'read_ticket', ticketId: 'T1' };
  if (!state.written) return writeAction;
  return { name: 'finish' };
}

async function execute(action, approval, signal) {
  signal.throwIfAborted();
  if (action.ticketId !== ticket.id || ticket.owner !== session.userId) {
    throw new Error('无权访问该工单');
  }
  if (action.name === 'read_ticket') {
    await sleep(2, undefined, { signal });
    return { status: 'ok', facts: { id: ticket.id, noteCount: ticket.notes.length } };
  }
  if (action.name !== 'append_note' || typeof action.text !== 'string' ||
      !action.text.trim() || action.text.length > 120) throw new Error('非法动作');
  const operationId = state.taskId + ':note-1';
  const fingerprint = digest(action);
  const saved = operations.get(operationId);
  if (saved) {
    if (saved.fingerprint !== fingerprint) throw new Error('幂等键参数冲突');
    return saved.result;
  }
  if (!approval || approval.userId !== session.userId ||
      approval.operationId !== operationId || approval.digest !== fingerprint ||
      !Number.isFinite(approval.expiresAt) || approval.expiresAt <= Date.now()) {
    return { status: 'needs_confirmation', operationId, digest: fingerprint, action };
  }
  signal.throwIfAborted();
  // 单进程同步临界段；真实服务要以事务保证写入和操作结果共同提交。
  ticket.notes.push(action.text);
  const result = { status: 'ok', noteId: 'note-' + ticket.notes.length };
  operations.set(operationId, { fingerprint, result });
  return result;
}

async function run(approval) {
  const started = performance.now();
  const remaining = () => state.activeRemainingMs - (performance.now() - started);
  try {
  while (state.steps < 6) {
    state.steps += 1; // 步数跨确认恢复保留，不因恢复而重置。
    const action = await withTimeout(mockDecide, Math.min(100, remaining()));
    if (action.name === 'finish') {
      if (!state.written) throw new Error('实际目标尚未完成');
      return { status: 'completed' };
    }
    const result = await withTimeout(
      signal => execute(action, approval, signal), Math.min(100, remaining()));
    if (result.status === 'needs_confirmation') return result;
    if (action.name === 'read_ticket') state.facts = result.facts;
    if (action.name === 'append_note') state.written = true;
  }
  throw new Error('步骤预算耗尽');
  } finally {
    // 活动消耗跨恢复保留；等待用户确认的空闲时间不计入此预算。
    state.activeRemainingMs = Math.max(0, remaining());
  }
}

const pending = await run();
console.log(pending.status);
const invalidApproval = { userId: 'u1', operationId: pending.operationId,
  digest: pending.digest }; // 缺失过期时间不能获得执行权。
assert.equal((await withTimeout(signal => execute(writeAction, invalidApproval, signal), 100))
  .status, 'needs_confirmation');
assert.equal(ticket.notes.length, 0);
const approval = { // 模拟用户看过 pending.action 后，服务端保存的确认记录。
  userId: 'u1', operationId: pending.operationId, digest: pending.digest,
  expiresAt: Date.now() + 60000,
};
console.log((await run(approval)).status);
await withTimeout(signal => execute(writeAction, approval, signal), 100);
console.log('实际备注条数=' + ticket.notes.length);
assert.equal(ticket.notes.length, 1);
state.activeRemainingMs = 0;
await assert.rejects(() => run(approval), /活动时间耗尽/);
await assert.rejects(() => withTimeout(signal => sleep(20, undefined, { signal }), 1),
  error => error.name === 'AbortError');

import assert from 'node:assert/strict';

function initial(taskId, maxSteps = 4) {
  if (!taskId || !Number.isInteger(maxSteps) || maxSteps < 1) {
    throw new Error('INVALID_TASK');
  }
  return { taskId, version: 0, status: 'created',
    steps: 0, maxSteps, pending: null, result: null };
}
function reduce(state, event) {
  if (event.version !== state.version + 1) throw new Error('STALE_EVENT');
  if (['completed', 'cancelled'].includes(state.status)) {
    throw new Error('TERMINAL_STATE');
  }
  const next = structuredClone(state);
  if (event.type === 'START' && state.status === 'created') {
    next.status = 'running';
  } else if (event.type === 'PLAN' && state.status === 'running') {
    if (state.steps >= state.maxSteps) throw new Error('STEP_LIMIT');
    if (!event.operation?.id || !event.operation?.digest) {
      throw new Error('INVALID_OPERATION');
    }
    next.steps += 1;
    next.pending = structuredClone(event.operation);
    next.status = 'awaiting_confirmation';
  } else if (event.type === 'APPROVE' &&
             state.status === 'awaiting_confirmation') {
    if (event.digest !== state.pending.digest) throw new Error('CHANGED_ACTION');
    next.status = 'running';
  } else if (event.type === 'APPLIED' && state.status === 'running') {
    if (!state.pending || event.operationId !== state.pending.id ||
        typeof event.resultId !== 'string' || !event.resultId) {
      throw new Error('RESULT_MISMATCH');
    }
    next.status = 'completed';
    next.result = { operationId: event.operationId, resultId: event.resultId };
    next.pending = null;
  } else if (event.type === 'CANCEL') {
    next.status = 'cancelled';
  } else {
    throw new Error('INVALID_TRANSITION');
  }
  next.version = event.version;
  return next;
}
let state = initial('task-1');
state = reduce(state, { type: 'START', version: 1 });
state = reduce(state, { type: 'PLAN', version: 2,
  operation: { id: 'op-1', digest: 'fixture-digest' } });
const restored = JSON.parse(JSON.stringify(state));
assert.equal(restored.status, 'awaiting_confirmation');
assert.throws(() => reduce(restored, { type: 'APPROVE',
  version: 3, digest: 'different' }), /CHANGED_ACTION/);
assert.throws(() => reduce(restored, { type: 'APPROVE',
  version: 2, digest: 'fixture-digest' }), /STALE_EVENT/);
state = reduce(restored, { type: 'APPROVE', version: 3, digest: 'fixture-digest' });
state = reduce(state, { type: 'APPLIED', version: 4,
  operationId: 'op-1', resultId: 'note-1' });
assert.equal(state.status, 'completed');
assert.throws(() => reduce(state, { type: 'START', version: 5 }),
  /TERMINAL_STATE/);
console.log({ status: state.status, version: state.version, steps: state.steps });