Manus 整体架构：
让模型拥有“电脑”和执行循环

把能力、策略和请求做成显式对象

enum Capability {
  FILE_READ, FILE_WRITE,
  NET_CONNECT, PROCESS_EXEC,
  CAMERA, MICROPHONE, NOTIFY
}

type TaskPolicy = {
  taskId: String
  userId: String
  capabilities: Set<Capability>
  allowedPaths: List<PathRule>
  allowedHosts: List<HostRule>
  maxCpuMs: Long
  maxMemoryMb: Int
  maxDiskMb: Int
  maxWallTimeMs: Long
  requireApproval: Set<Capability>
  allowCloudHandoff: Boolean
}

type CapabilityToken = {
  tokenId: String
  taskId: String
  capability: Capability
  scope: Json
  expiresAt: Instant
  nonce: Bytes
  signature: Bytes
}

type ToolRequest = {
  taskId: String
  stepId: String
  toolName: String
  args: Json
  tokenId: String?
  idempotencyKey: String
}

策略不是提示词，要在执行边界强制检查

function buildPolicy(task): TaskPolicy {
  risk = classifyRisk(task.goal, task.tools)

  // 默认：无网络、无原生进程、只读任务目录
  p = TaskPolicy(
    taskId = task.id,
    userId = task.userId,
    capabilities = { FILE_READ },
    allowedPaths = [
      PathRule("workspace/" + task.id,
               read=true, write=true)
    ],
    allowedHosts = [],
    maxCpuMs = 30_000,
    maxMemoryMb = 512,
    maxDiskMb = 256,
    maxWallTimeMs = 120_000,
    requireApproval = {},
    allowCloudHandoff = true
  )

  if task.needsWeb:
    p.capabilities += NET_CONNECT
    p.allowedHosts = normalizeAllowlist(task.hosts)

  if risk >= HIGH:
    p.requireApproval += { FILE_WRITE, NET_CONNECT,
                           PROCESS_EXEC }

  return p
}

注意：路径必须 canonicalize 后再匹配，不能用字符串前缀判断；域名解析、重定向和代理也必须重新检查。

模型只能申请能力，不能自行制造权限

class CapabilityBroker {
  policyStore: PolicyStore
  approval: ApprovalService
  signer: KeyStore
  audit: AuditLog

  async function authorize(req, ctx)
      : Result<CapabilityToken> {
    policy = policyStore.get(ctx.taskId)
    cap = toolToCapability(req.toolName)

    if cap not in policy.capabilities:
      audit.deny(req, "capability_not_granted")
      return Denied

    scope = deriveScope(cap, req.args, policy)
    if scope.invalid:
      audit.deny(req, "scope_out_of_policy")
      return Denied

    if cap in policy.requireApproval:
      decision = await approval.request(
        taskId=ctx.taskId, summary=redact(req),
        risk=cap, timeoutMs=60_000)
      if decision != APPROVED:
        audit.deny(req, "user_not_approved")
        return Denied

    token = CapabilityToken(
      tokenId=randomId(), taskId=ctx.taskId,
      capability=cap, scope=scope,
      expiresAt=now() + 60.seconds,
      nonce=randomBytes(32), signature=null)
    token.signature = signer.sign(canonicalJson(token))
    audit.grant(req, token)
    return Granted(token)
  }
}

每次工具调用都重新验证，不信任模型参数

async function execute(req, ctx): ToolResult {
  await rateLimiter.check(ctx.taskId)
  await replayGuard.check(req.idempotencyKey)

  decision = await authorize(req, ctx)
  if decision is Denied:
    return ToolError("DENIED", decision.reason)

  token = decision.token
  if !signer.verify(token):
    return ToolError("INVALID_TOKEN")
  if token.taskId != ctx.taskId:
    return ToolError("TOKEN_TASK_MISMATCH")
  if token.expiresAt < now():
    return ToolError("TOKEN_EXPIRED")

  safeArgs = enforceScope(req.args, token.scope)
  audit.start(req, token)
  try:
    result = await sandbox.run(
      capability=token.capability,
      scope=token.scope, args=safeArgs,
      budget=ctx.policy.budget)
    audit.finish(req, result.summary)
    return redactForModel(result)
  catch e:
    audit.error(req, classify(e))
    return ToolError("EXECUTION_FAILED")
}

在安全的步骤边界生成 handoff manifest

type HandoffManifest = {
  taskId: String
  parentStepId: String
  checkpointHash: String
  eventCursor: String
  policyDigest: String
  inputArtifactRefs: List<Uri>
  outputArtifactRefs: List<Uri>
  allowedCapabilities: Set<Capability>
  privacyClass: "LOCAL_ONLY" | "CLOUD_OK"
  expiresAt: Instant
}

async function handoffToCloud(ctx): HandoffResult {
  if ctx.policy.allowCloudHandoff == false:
    return Rejected("policy")
  if ctx.privacyClass == LOCAL_ONLY:
    return Rejected("privacy")

  checkpoint = eventLog.commitCheckpoint(ctx)
  refs = artifactStore.uploadEncrypted(
    only=checkpoint.referencedArtifacts)
  manifest = sign(HandoffManifest(
    taskId=ctx.taskId,
    parentStepId=ctx.stepId,
    checkpointHash=checkpoint.hash,
    eventCursor=checkpoint.cursor,
    policyDigest=ctx.policy.digest,
    inputArtifactRefs=refs.inputs,
    outputArtifactRefs=[],
    allowedCapabilities=cloudSafeCaps(ctx),
    privacyClass=ctx.privacyClass,
    expiresAt=now()+10.minutes))
  return transport.send("cloud", manifest)
}

事件先落盘，结果幂等，再允许 Agent 继续

async function runStep(ctx, step): Event {
  stepKey = `${ctx.taskId}:${step.id}`
  if eventLog.hasCompleted(stepKey):
    return eventLog.getResult(stepKey)

  started = eventLog.append({
    type: "STEP_STARTED", stepId: step.id,
    idempotencyKey: stepKey})

  try:
    result = await broker.execute(
      request=step.toolRequest,
      ctx=ctx.withBudget(step.budget))
    completed = eventLog.append({
      type: "STEP_COMPLETED", stepId: step.id,
      resultRef=artifactStore.save(result),
      idempotencyKey=stepKey})
    return completed
  catch e:
    failed = eventLog.append({
      type: "STEP_FAILED", stepId: step.id,
      retryable=isRetryable(e), errorClass=classify(e)})
    scheduler.scheduleRetryOrHandoff(failed)
    return failed
}

先定义不可变事实与可重建索引

enum EventType {
  TASK_CREATED, GOAL_CHANGED,
  STEP_STARTED, STEP_COMPLETED, STEP_FAILED,
  MEMORY_CANDIDATE, MEMORY_CONFIRMED,
  CHECKPOINT_CREATED, TASK_PAUSED,
  HANDOFF_REQUESTED, HANDOFF_ACCEPTED,
  ARTIFACT_SAVED, TASK_COMPLETED
}

type Event = {
  eventId: String
  taskId: String
  seq: Long
  type: EventType
  createdAt: Instant
  payload: Json
  payloadRef: Uri?
  idempotencyKey: String?
  prevHash: String?
  hash: String
}

type TaskState = {
  taskId: String
  status: TaskStatus
  goal: String
  phase: String
  eventCursor: Long
  completedSteps: Set<String>
  pendingSteps: List<Step>
  blockedReasons: List<String>
  checkpointId: String?
  policyDigest: String
}

type ArtifactRef = {
  uri: Uri
  sha256: String
  mimeType: String
  sizeBytes: Long
  sensitivity: Sensitivity
}

热状态、任务日志、向量记忆和产物分工

class StorageLayout {
  // L0：当前模型调用，永不作为事实源
  contextBuilder: ContextBuilder

  // L1：当前任务热状态，可从事件重建
  sessionDb: SQLite<SessionState>

  // L2：append-only canonical event log
  eventLog: DurableEventLog

  // L3：可删除、可重建的语义索引
  vectorIndex: LocalVectorIndex
  memoryRecords: SQLite<MemoryRecord>

  // L4：大文件、原始输出、checkpoint payload
  artifactStore: EncryptedBlobStore

  // L5：可选云端副本，不是本地事实源
  syncQueue: SQLite<SyncEnvelope>
}

function persistEvent(event): CommitResult {
  // 事务：写入事件后才更新 cursor / 热状态
  tx = eventLog.beginTransaction()
  try:
    tx.append(event)
    next = reducer.apply(sessionDb.load(event.taskId), event)
    tx.upsertSession(next)
    tx.commit()
    return Committed(next.eventCursor)
  catch e:
    tx.rollback()
    return Failed("DURABLE_WRITE_FAILED")
}

任何状态变化都通过事件提交

async function appendEvent(taskId, type, payload, opts={}): Event {
  last = eventLog.last(taskId)
  event = Event(
    eventId = uuid(), taskId = taskId,
    seq = last.seq + 1, type = type,
    createdAt = clock.now(), payload = payload,
    payloadRef = opts.payloadRef,
    idempotencyKey = opts.idempotencyKey,
    prevHash = last.hash,
    hash = hashCanonical(last.hash, type, payload))

  if opts.idempotencyKey:
    old = eventLog.findByIdempotency(opts.idempotencyKey)
    if old != null: return old

  await persistEvent(event)
  return event
}

function reduce(state, event): TaskState {
  switch event.type:
    case STEP_COMPLETED:
      state.completedSteps.add(event.payload.stepId)
      state.pendingSteps.remove(event.payload.stepId)
    case TASK_PAUSED:
      state.status = PAUSED
    case HANDOFF_ACCEPTED:
      state.status = RUNNING_ON_CLOUD
    case TASK_COMPLETED:
      state.status = COMPLETED
  state.eventCursor = event.seq
  return state
}

保存语义状态，不保存不可移植的 KV-cache

type Checkpoint = {
  checkpointId: String
  taskId: String
  eventCursor: Long
  stateDigest: String
  goalSummary: String
  currentPhase: String
  completedSteps: List<String>
  pendingSteps: List<String>
  decisions: List<DecisionRef>
  openQuestions: List<String>
  memoryRefs: List<String>
  artifactRefs: List<ArtifactRef>
  policyDigest: String
}

async function createCheckpoint(taskId): Checkpoint {
  state = sessionDb.load(taskId)
  cp = Checkpoint(
    checkpointId=uuid(), taskId=taskId,
    eventCursor=state.eventCursor,
    stateDigest=hash(state),
    goalSummary=summarizeGoal(state),
    currentPhase=state.phase,
    completedSteps=state.completedSteps,
    pendingSteps=state.pendingSteps,
    decisions=extractDecisionRefs(taskId),
    openQuestions=state.blockedReasons,
    memoryRefs=memoryDb.refsForTask(taskId),
    artifactRefs=artifactStore.refsForTask(taskId),
    policyDigest=state.policyDigest)
  ref = await artifactStore.putJson(cp)
  await appendEvent(taskId, CHECKPOINT_CREATED,
      {checkpointId: cp.checkpointId}, {payloadRef: ref.uri})
  return cp
}

混合检索：语义 + 关键词 + 新鲜度 + 任务相关性

async function retrieveMemory(query, ctx): List<MemoryRecord> {
  parsed = queryParser.parse(query)
  filters = {
    status: "active",
    scope: inSet(["user", "device", "task"]),
    sensitivity: allowedBy(ctx.policy),
    taskId: ctx.taskId or ANY
  }

  vectorHits = vectorIndex.search(
    embed(query), topK=40, filters=filters)
  keywordHits = memoryDb.fullTextSearch(
    parsed.terms, topK=40, filters=filters)
  candidates = deduplicate(vectorHits + keywordHits)

  scored = candidates.map(m => ({
    memory: m,
    score: 0.45 * m.semanticScore
      + 0.20 * m.lexicalScore
      + 0.15 * taskRelevance(m, ctx)
      + 0.10 * m.confidence
      + 0.10 * freshness(m, clock.now())
  }))

  ranked = rerank(scored, query, ctx)
  return diversifyAndBudget(
    ranked, maxTokens=ctx.memoryTokenBudget)
}

function freshness(m, now): Float {
  halfLife = {
    preference: 180.days,
    lesson: 30.days,
    tool_failure: 14.days,
    task_state: INFINITE
  }[m.kind]
  return validAt(m, now)
    ? expDecay(now - m.validFrom, halfLife)
    : 0.0
}

把检索结果变成有来源的上下文片段

async function buildContext(taskId, step): Context {
  state = sessionDb.load(taskId)
  recent = eventLog.tail(taskId, maxEvents=20)
  memories = await retrieveMemory(
    query=step.goal, ctx=state.context)

  // 模型看到的是证据包，不是无来源的散文
  evidence = memories.map(m => {
    id: m.memoryId,
    text: m.content,
    source: m.sourceEventId,
    artifact: m.sourceArtifactRef,
    confidence: m.confidence,
    validUntil: m.validUntil
  })

  return Context(
    stableSystemPrompt=systemPrompt,
    goal=state.goal,
    checkpoint=state.checkpointSummary,
    activeConstraints=state.activeConstraints,
    recentEvents=redact(recent),
    retrievedMemory=evidence,
    artifactRefs=state.artifactRefs,
    toolSchema=toolRegistry.for(step))
}

Checkpoint + Event Replay + Policy Revalidation

async function resumeTask(taskId): ResumeResult {
  cp = checkpointStore.latest(taskId)
  if cp == null:
    cp = checkpointStore.fromTaskCreated(taskId)

  currentPolicy = policyStore.current(taskId)
  state = checkpointToState(cp)
  tail = eventLog.readAfter(taskId, cp.eventCursor)
  for event in tail:
    state = reduce(state, event)

  if cp.policyDigest != currentPolicy.digest:
    state = revalidatePendingSteps(state, currentPolicy)
  if state.status in {COMPLETED, CANCELLED}:
    return Done(state)

  next = planner.nextRunnableStep(state)
  ctx = await buildContext(taskId, next)
  return runner.resume(
    taskId=taskId, state=state,
    nextStep=next, context=ctx)
}

拆成可抢占、可重试、可休眠的小步骤

async function schedulerTick() {
  tasks = taskStore.findRunnable(limit=20)
  for task in tasks:
    device = deviceMonitor.snapshot()
    if !resourcePolicy.allows(task, device):
      await appendEvent(task.id, TASK_PAUSED,
        {reason: resourcePolicy.reason(device)})
      continue

    lease = leaseStore.acquire(task.id,
      owner=device.id, ttl=60.seconds)
    if !lease.acquired: continue

    try:
      step = planner.nextRunnableStep(task.state)
      result = await runIdempotentStep(task, step)
      if shouldCheckpoint(task, result):
        await createCheckpoint(task.id)
    finally:
      leaseStore.release(lease)
}

async function runIdempotentStep(task, step) {
  key = `${task.id}:${step.id}:${step.inputHash}`
  old = eventLog.completedByIdempotency(key)
  if old: return old.payload
  await appendEvent(task.id, STEP_STARTED,
    {stepId: step.id}, {idempotencyKey: key})
  result = await toolRunner.execute(step)
  await appendEvent(task.id, STEP_COMPLETED,
    {stepId: step.id, resultRef: save(result)},
    {idempotencyKey: key})
  return result
}

同步 canonical records，不直接复制向量索引

async function syncOutbox() {
  batch = syncQueue.nextBatch(maxBytes=1_000_000)
  for item in batch:
    if !privacyPolicy.allows(item):
      syncQueue.markBlocked(item, "LOCAL_ONLY")
      continue

    envelope = encryptAndSign({
      deviceId: device.id,
      taskId: item.taskId,
      cursor: item.cursor,
      type: item.type,
      payloadRef: item.payloadRef,
      previousHash: item.previousHash
    })
    ack = await transport.send(envelope)
    if ack.accepted:
      syncQueue.ack(item.id, ack.serverCursor)
}

async function mergeCloudDelta(delta) {
  verifySignature(delta)
  for event in delta.events:
    if !eventLog.has(event.eventId):
      await persistEvent(event)
  for memory in delta.memoryRecords:
    memoryDb.merge(resolveConflict(
      memoryDb.find(memory.memoryId), memory))
  await rebuildDerivedIndexesIfNeeded(delta)
}

向量数据库是派生数据，损坏后可重建

async function rebuildVectorIndex(modelVersion) {
  next = LocalVectorIndex.createTemporary(
    modelVersion=modelVersion)
  cursor = 0
  while true:
    batch = memoryDb.activeRecordsAfter(cursor,
      limit=128, allowed=privacyPolicy.localIndex())
    if batch.empty: break
    vectors = await embedder.embed(
      batch.map(x => x.content))
    next.upsert(batch.zip(vectors))
    cursor = batch[-1].rowId
    await scheduler.yieldIfNeeded()
  next.commitAtomically()
}

async function deleteMemory(memoryId) {
  memoryDb.markDeleted(memoryId)
  vectorIndex.delete(memoryId)
  syncQueue.enqueue({type: "MEMORY_DELETE", memoryId})
  // 原始来源事件按审计/合规策略保留或擦除
  await audit.record("MEMORY_DELETED", memoryId)
}