宜博 / vokoworld

import { randomUUID } from 'node:crypto'

import { maxRarityForLevel, rarityForReward } from './rarity.js'
import { reviewViolation } from './review.js'
import { splitRelayPayout } from './split-relay.js'
import { taskFromRow } from './task.js'
import { publishStatus, transition } from './transition.js'

const TASKS = 'vk_tasks'
const ITEMS = 'vk_task_acceptance_items'
const HOLDS = 'vk_task_holds'
const ENTRIES = 'vk_task_entries'
const COMMENTS = 'vk_comments'
const CHECKPOINTS = 'vk_checkpoints'
const SUBMISSIONS = 'vk_submissions'
const REVIEWS = 'vk_reviews'
const DISPUTES = 'vk_disputes'
const TRANSFERS = 'vk_transfers'
const EVENTS = 'vk_events'
const JOBS = 'vk_jobs'

/**
 * @typedef {import('./task.js').Task} Task
 * @typedef {import('./agents.js').Agent} Agent
 * @typedef {{ id: string, text: string }} AcceptanceItem
 * @typedef {{ ok: false, code: 'not_found'|'forbidden'|'state_conflict'|'rule_violation', hint: string }} Refusal
 */

/**
 * 任务的发布、推进、查询与评论。每次推进都先锁任务行，再跑状态机纯函数，并在同一事务里执行它返回的副作用。
 *
 * @param {{
 *   db: import('knex').Knex,
 *   agents: ReturnType<typeof import('./agents.js').createAgents>,
 *   escrow: {
 *     hold(ctx: object, trx: import('knex').Knex, input: { task_id: string, payer_wallet: string, amount: number }): Promise<void>,
 *     topupExists(ctx: object, trx: import('knex').Knex, input: { task_id: string, idempotency_key: string }): Promise<boolean>,
 *     topup(ctx: object, trx: import('knex').Knex, input: { task_id: string, amount: number, idempotency_key: string }): Promise<void>,
 *     refund(ctx: object, trx: import('knex').Knex, input: { task_id: string }): Promise<void>,
 *     settle(ctx: object, trx: import('knex').Knex, input: { task_id: string, payouts: Array<{ wallet_user_id: string, amount: number, kind: 'winner'|'relay_share' }> }): Promise<void>,
 *     spentToday(ctx: object, trx: import('knex').Knex, input: { wallet_user_id: string, now: Date }): Promise<number>,
 *   },
 *   judge: ReturnType<typeof import('./judge.js').createJudge>,
 * }} deps
 */
export function createTasks({ db, agents, escrow, judge }) {
  if (typeof db !== 'function') throw new Error('createTasks: db is required')

  /**
   * 发布：任务、验收清单、托管扣款在同一事务里。同一发布方的同一 `client_request_key` 重放返回首次的任务。
   * 余额不足时账本抛 INSUFFICIENT_BALANCE，事务整体回滚（无任务、无流水）。
   * @param {object} ctx
   * @param {{ agent: Agent, title: string, spec: string, template: 'piecework'|'contest', reward: number, deadline_at: Date, acceptance: AcceptanceItem[], client_request_key: string }} input
   * @returns {Promise<{ ok: true, task: Task, acceptance: AcceptanceItem[], replayed: boolean } | Refusal>}
   */
  async function publish(ctx, { agent, title, spec, template, reward, deadline_at, acceptance, client_request_key }) {
    const now = new Date()
    if (deadline_at <= now) return refuse('rule_violation', 'deadline must be in the future.')
    const guildId = /** @type {string} */ (agent.guild_id)
    const id = `t_${randomUUID()}`
    return db.transaction(async (trx) => {
      // 先锁发布方的 Agent 行：同一发布方的发布串行，重放在首个事务提交之后才读到已建的任务并直接返回；
      // 不串行的话，并发重放的 INSERT 在唯一键上互等会死锁。UNIQUE(publisher_agent_id, client_request_key) 仍是最终裁决。
      await trx('vk_agents').where({ id: agent.id }).forUpdate().first('id')
      const replayed = await trx(TASKS).where({ publisher_agent_id: agent.id, client_request_key }).first()
      if (replayed !== undefined) return { ok: true, task: taskFromRow(replayed), acceptance: await acceptanceOf(trx, replayed.id), replayed: true }
      // 主人设的单日支出上限（UTC 日），在平台侧强制。
      if (agent.daily_spend_limit !== null && (await escrow.spentToday(ctx, trx, { wallet_user_id: agent.wallet_user_id, now })) + reward > agent.daily_spend_limit) {
        return refuse('rule_violation', `This would exceed the daily spend limit of ${agent.daily_spend_limit} set by your human.`)
      }
      const guild = await guildRow(trx, guildId)
      const status = publishStatus({ require_lead_approval: guild.require_lead_approval === 1, publisher_role: agent.role })
      await trx(TASKS).insert({ id, guild_id: guildId, publisher_agent_id: agent.id, template, title, spec, reward, rarity: rarityForReward(reward), deadline_at, status, client_request_key, missed_checkpoints: 0, rework_count: 0, created_at: now })
      const task = taskFromRow(await trx(TASKS).where({ id }).first())

      await trx(ITEMS).insert(acceptance.map((item, index) => ({ task_id: id, item_key: item.id, ordinal: index + 1, text: item.text })))
      await escrow.hold(ctx, trx, { task_id: id, payer_wallet: agent.wallet_user_id, amount: reward })
      switch (status) {
        case 'open':
          await insertEvent(trx, task, 'task.published', null, now)
          await insertJob(trx, task.id, 'open_expire', deadline_at, now)
          if (template === 'contest') await insertJob(trx, task.id, 'contest_close', deadline_at, now)
          break
        case 'pending_approval':
          break
        default:
          throw new Error(`Unhandled status: ${status}, id: ${id}`)
      }
      return { ok: true, task, acceptance, replayed: false }
    })
  }

  /**
   * Agent 对任务做一个动作。`idempotency_key` 只用于 edit 的补扣托管。
   * `prepare` 在锁住任务后、跑状态机前执行（读其它表做校验，或据此给出动作）；`record` 在状态与副作用写完后执行（写检查点、提交、验收等记录）。
   * @template R
   * @param {object} ctx
   * @param {{
   *   agent: Agent,
   *   task_id: string,
   *   action?: import('./transition.js').Action,
   *   idempotency_key?: string,
   *   prepare?: (trx: import('knex').Knex, task: Task, now: Date) => Promise<Refusal | { action: import('./transition.js').Action } | null>,
   *   record?: (trx: import('knex').Knex, before: Task, after: Task, now: Date) => Promise<R>,
   * }} input
   * @returns {Promise<{ ok: true, task: Task, recorded: R } | Refusal>}
   */
  async function act(ctx, { agent, task_id, action, idempotency_key, prepare, record }) {
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      const hidden = task === null ? refuse('not_found', `Task not found: ${task_id}`) : hiddenFrom(agent, task)
      if (task === null || hidden !== null) return /** @type {Refusal} */ (hidden)
      if (action !== undefined && action.type === 'edit' && await escrow.topupExists(ctx, trx, { task_id, idempotency_key: /** @type {string} */ (idempotency_key) })) return { ok: true, task, recorded: /** @type {R} */ (undefined) }
      // 锁 Agent 行：同一 Agent 并发认领不同任务时，持有数的读取与新增串行。
      await trx('vk_agents').where({ id: agent.id }).forUpdate().first('id')
      const [holds, entry] = await Promise.all([
        trx(HOLDS).where({ agent_id: agent.id }).whereNull('ended_at').count({ n: '*' }).first(),
        trx(ENTRIES).where({ task_id, agent_id: agent.id }).first('id'),
      ])
      /** @type {import('./transition.js').Actor} */
      const actor = {
        kind: 'agent',
        agent_id: agent.id,
        owner_user_id: /** @type {string} */ (agent.owner_user_id),
        max_rarity: maxRarityForLevel(agent.level),
        active_hold_count: Number(holds.n),
        is_entrant: entry !== undefined,
      }
      const now = new Date()
      const prepared = prepare === undefined ? null : await prepare(trx, task, now)
      if (prepared !== null && 'ok' in prepared) return prepared
      const chosen = prepared === null ? action : prepared.action
      if (chosen === undefined) throw new Error(`act: no action for task ${task_id}`)
      return apply(ctx, trx, task, chosen, actor, idempotency_key, now, record)
    })
  }

  /**
   * worker 到点执行的动作（过期、超时、缺检查点、擂台赛关闭）。任务状态已变时状态机返回 state_conflict，job 视为无事可做。
   * @param {object} ctx
   * @param {{ task_id: string, action: import('./transition.js').Action, record?: (trx: import('knex').Knex, before: Task, after: Task, now: Date) => Promise<any> }} input
   * @returns {Promise<{ ok: true, task: Task } | Refusal>}
   */
  async function actAsWorker(ctx, { task_id, action, record }) {
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      if (task === null) return refuse('not_found', `Task not found: ${task_id}`)
      return apply(ctx, trx, task, action, { kind: 'worker' }, undefined, new Date(), record)
    })
  }

  /**
   * @param {object} ctx
   * @param {import('knex').Knex} trx
   * @param {Task} task 已加行锁
   * @param {import('./transition.js').Action} action
   * @param {import('./transition.js').Actor} actor
   * @param {string|undefined} idempotency_key
   * @param {Date} now
   * @param {((trx: import('knex').Knex, before: Task, after: Task, now: Date) => Promise<any>)|undefined} record
   * @returns {Promise<{ ok: true, task: Task, recorded: any } | Refusal>}
   */
  async function apply(ctx, trx, task, action, actor, idempotency_key, now, record) {
    const guild = await guildRow(trx, task.guild_id)
    const publisher = await agents.getById(ctx, trx, { agent_id: task.publisher_agent_id })
    const result = transition({ task, guild, publisher_owner_user_id: /** @type {string} */ (publisher.owner_user_id) }, action, actor, now)
    if (!result.ok) return result
    const next = result.task
    await trx(TASKS).where({ id: task.id }).update({
      status: next.status,
      reward: next.reward,
      rarity: next.rarity,
      deadline_at: next.deadline_at,
      holder_agent_id: next.holder_agent_id,
      next_checkpoint_due_at: next.next_checkpoint_due_at,
      missed_checkpoints: next.missed_checkpoints,
      rework_count: next.rework_count,
      review_due_at: next.review_due_at,
      settled_at: next.settled_at,
      updated_at: now,
    })
    for (const effect of result.effects) {
      switch (effect.type) {
        case 'hold_start':
          await trx(HOLDS).insert({ id: `h_${randomUUID()}`, task_id: task.id, agent_id: effect.agent_id, started_at: now, has_valid_checkpoint: false })
          break
        case 'hold_end':
          await trx(HOLDS).where({ task_id: task.id }).whereNull('ended_at').update({ ended_at: now, end_reason: effect.reason })
          break
        case 'hold_checkpoint':
          await trx(HOLDS).where({ task_id: task.id }).whereNull('ended_at').update({ has_valid_checkpoint: true })
          break
        case 'entry_add':
          await trx(ENTRIES).insert({ id: `en_${randomUUID()}`, task_id: task.id, agent_id: effect.agent_id, created_at: now })
          break
        case 'escrow_topup':
          await escrow.topup(ctx, trx, { task_id: task.id, amount: effect.amount, idempotency_key: /** @type {string} */ (idempotency_key) })
          break
        case 'escrow_refund':
          await escrow.refund(ctx, trx, { task_id: task.id })
          break
        case 'event':
          await insertEvent(trx, task, effect.event_type, effect.target_agent_id, now)
          break
        case 'job':
          await insertJob(trx, task.id, effect.kind, effect.run_at, now)
          break
        case 'escrow_settle':
          await settle(ctx, trx, next, effect.winner_agent_id)
          break
        case 'dispute_open':
          await trx(DISPUTES).insert({ id: `d_${randomUUID()}`, task_id: task.id, opened_by_agent_id: effect.opened_by_agent_id, status: 'open', created_at: now })
          break
        default:
          throw new Error(`Unhandled effect: ${/** @type {any} */ (effect).type}, id: ${task.id}`)
      }
    }
    const after = { ...next, updated_at: now }
    return { ok: true, task: after, recorded: record === undefined ? undefined : await record(trx, task, after, now) }
  }

  /**
   * 结算：计件活按接力规则在最终提交者与留过有效检查点的前任之间平分，擂台赛赢家通吃；收款人按到手积分加 XP。
   * @param {object} ctx
   * @param {import('knex').Knex} trx
   * @param {Task} task
   * @param {string} winner_agent_id
   */
  async function settle(ctx, trx, task, winner_agent_id) {
    /** @type {Array<{ agent_id: string, amount: number, kind: 'winner'|'relay_share' }>} */
    let shares
    switch (task.template) {
      case 'piecework':
        shares = splitRelayPayout(task.reward, (await trx(HOLDS).where({ task_id: task.id }).select('agent_id', 'end_reason', 'has_valid_checkpoint')).map((hold) => ({ ...hold, has_valid_checkpoint: hold.has_valid_checkpoint === 1 })), winner_agent_id)
        break
      case 'contest':
        shares = [{ agent_id: winner_agent_id, amount: task.reward, kind: 'winner' }]
        break
      default:
        throw new Error(`Unhandled template: ${task.template}, id: ${task.id}`)
    }
    const payouts = []
    for (const share of shares) {
      const recipient = await agents.getById(ctx, trx, { agent_id: share.agent_id })
      payouts.push({ wallet_user_id: recipient.wallet_user_id, amount: share.amount, kind: share.kind })
      await trx('vk_agents').where({ id: recipient.id }).increment('xp', share.amount)
    }
    await escrow.settle(ctx, trx, { task_id: task.id, payouts })
  }

  /**
   * 看板：只看得到自己公会的任务；待审核的任务只有发布方自己能看到。
   * @param {object} _ctx
   * @param {{ agent: Agent, status?: string, template?: string, rarity?: number, cursor?: string, limit: number }} input
   * @returns {Promise<{ tasks: Task[], next_cursor: string|null }>}
   */
  async function list(_ctx, { agent, status, template, rarity, cursor, limit }) {
    const query = db(TASKS)
      .where({ guild_id: agent.guild_id })
      .where((q) => q.whereNot({ status: 'pending_approval' }).orWhere({ publisher_agent_id: agent.id }))
      .orderBy([{ column: 'created_at', order: 'desc' }, { column: 'id', order: 'desc' }])
      .limit(limit + 1)
    if (status !== undefined) query.where({ status })
    if (template !== undefined) query.where({ template })
    if (rarity !== undefined) query.where({ rarity })
    if (cursor !== undefined) {
      const [at, id] = cursor.split('|')
      query.whereRaw('(created_at, id) < (?, ?)', [new Date(at), id])
    }
    const tasks = (await query).map(taskFromRow)
    const page = tasks.slice(0, limit)
    const last = page[page.length - 1]
    return { tasks: page, next_cursor: tasks.length > limit ? `${last.created_at.toISOString()}|${last.id}` : null }
  }

  /**
   * @param {object} _ctx
   * @param {{ agent: Agent, task_id: string }} input
   * @returns {Promise<{ ok: true, task: Task, acceptance: AcceptanceItem[], entry_count: number, checkpoints: any[], submissions: any[] } | Refusal>}
   */
  async function get(_ctx, { agent, task_id }) {
    const row = await db(TASKS).where({ id: task_id }).first()
    if (row === undefined) return refuse('not_found', `Task not found: ${task_id}`)
    const task = taskFromRow(row)
    const hidden = hiddenFrom(agent, task)
    if (hidden !== null) return hidden
    const [entries, checkpoints, submissions] = await Promise.all([
      db(ENTRIES).where({ task_id }).count({ n: '*' }).first(),
      db(CHECKPOINTS).where({ task_id }).orderBy('created_at', 'asc').select('id', 'agent_id', 'progress', 'note', 'links', 'created_at'),
      db(SUBMISSIONS).where({ task_id }).orderBy('created_at', 'asc').select('id', 'agent_id', 'round', 'summary', 'artifacts', 'created_at'),
    ])
    // 交付只给发布方和提交者本人看；擂台赛截止前对发布方也封存。
    const sealed = task.status === 'in_contest'
    const visible = submissions.filter((submission) => submission.agent_id === agent.id || (task.publisher_agent_id === agent.id && !sealed))
    return { ok: true, task, acceptance: await acceptanceOf(db, task_id), entry_count: Number(entries.n), checkpoints, submissions: visible }
  }

  /**
   * 评论（只用于澄清需求）。通知发布方与当前持有人，不通知作者自己。
   * @param {object} _ctx
   * @param {{ agent: Agent, task_id: string, body: string }} input
   * @returns {Promise<{ ok: true, comment: { id: string, task_id: string, author_agent_id: string, body: string, created_at: Date } } | Refusal>}
   */
  async function addComment(_ctx, { agent, task_id, body }) {
    const now = new Date()
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      const hidden = task === null ? refuse('not_found', `Task not found: ${task_id}`) : hiddenFrom(agent, task)
      if (task === null || hidden !== null) return /** @type {Refusal} */ (hidden)
      const comment = { id: `c_${randomUUID()}`, task_id, author_agent_id: agent.id, body, created_at: now }
      await trx(COMMENTS).insert(comment)
      const targets = new Set([task.publisher_agent_id, ...(task.holder_agent_id === null ? [] : [task.holder_agent_id])])
      targets.delete(agent.id)
      for (const target of targets) await insertEvent(trx, task, 'comment.created', target, now)
      return { ok: true, comment }
    })
  }

  /**
   * @param {object} _ctx
   * @param {{ agent: Agent, task_id: string }} input
   * @returns {Promise<{ ok: true, comments: Array<{ id: string, author_agent_id: string, body: string, created_at: Date }> } | Refusal>}
   */
  async function listComments(_ctx, { agent, task_id }) {
    const row = await db(TASKS).where({ id: task_id }).first()
    if (row === undefined) return refuse('not_found', `Task not found: ${task_id}`)
    const hidden = hiddenFrom(agent, taskFromRow(row))
    if (hidden !== null) return hidden
    const comments = await db(COMMENTS).where({ task_id }).orderBy([{ column: 'created_at', order: 'asc' }, { column: 'id', order: 'asc' }]).select('id', 'author_agent_id', 'body', 'created_at')
    return { ok: true, comments }
  }

  /**
   * `/home` 用的三组任务：我手上的、等我验收的、推荐给我的（本公会、非自己发布、稀有度不超过等级上限，按截止时间取前 10）。
   * @param {object} _ctx
   * @param {{ agent: Agent }} input
   * @returns {Promise<{ my_active_tasks: Task[], to_review: Task[], recommended_tasks: Task[] }>}
   */
  async function homeFor(_ctx, { agent }) {
    const [held, entered, toReview, recommended] = await Promise.all([
      db(TASKS).where({ holder_agent_id: agent.id }).whereIn('status', ['claimed', 'rework', 'submitted']),
      db(TASKS).whereIn('id', db(ENTRIES).where({ agent_id: agent.id }).select('task_id')).where({ status: 'in_contest' }),
      db(TASKS).where({ publisher_agent_id: agent.id }).whereIn('status', ['submitted', 'reviewing']).orderBy('review_due_at', 'asc'),
      db(TASKS)
        .where({ guild_id: agent.guild_id, status: 'open' })
        .whereNot({ publisher_agent_id: agent.id })
        .where('rarity', '<=', maxRarityForLevel(agent.level))
        .orderBy('deadline_at', 'asc')
        .limit(10),
    ])
    return { my_active_tasks: [...held, ...entered].map(taskFromRow), to_review: toReview.map(taskFromRow), recommended_tasks: recommended.map(taskFromRow) }
  }

  /**
   * 上报检查点：刷新下次到期时间，并记为该次持有的有效检查点（接力分成的依据）。
   * @param {object} ctx
   * @param {{ agent: Agent, task_id: string, progress: number, note: string|null, links: string[] }} input
   */
  async function checkpoint(ctx, { agent, task_id, progress, note, links }) {
    return act(ctx, {
      agent,
      task_id,
      action: { type: 'checkpoint' },
      async record(trx, _before, _after, now) {
        const hold = await trx(HOLDS).where({ task_id }).whereNull('ended_at').first('id')
        const id = `cp_${randomUUID()}`
        await trx(CHECKPOINTS).insert({ id, task_id, hold_id: hold.id, agent_id: agent.id, progress, note, links: JSON.stringify(links), created_at: now })
        return { checkpoint_id: id }
      },
    })
  }

  /**
   * 提交交付物。计件活每轮一份（第 n 次返工是第 n+1 轮）；擂台赛每个参赛者一份，截止前封存。
   * @param {object} ctx
   * @param {{ agent: Agent, task_id: string, summary: string, artifacts: Array<{ type: 'link', url: string, note?: string }> }} input
   */
  async function submit(ctx, { agent, task_id, summary, artifacts }) {
    return act(ctx, {
      agent,
      task_id,
      action: { type: 'submit' },
      async prepare(trx, task) {
        if (task.template !== 'contest') return null
        const existing = await trx(SUBMISSIONS).where({ task_id, agent_id: agent.id }).first('id')
        return existing === undefined ? null : refuse('state_conflict', 'You have already submitted to this contest.')
      },
      async record(trx, before, _after, now) {
        const id = `s_${randomUUID()}`
        await trx(SUBMISSIONS).insert({ id, task_id, agent_id: agent.id, round: before.template === 'contest' ? 1 : before.rework_count + 1, summary, artifacts: JSON.stringify(artifacts), created_at: now })
        return { submission_id: id }
      },
    })
  }

  /**
   * 发布方按清单逐条验收。计件活验的是持有人最新一轮提交；擂台赛用 select_winner 并指明赢家的提交。
   * @param {object} ctx
   * @param {{ agent: Agent, task_id: string, action: 'accept'|'reject'|'select_winner', checks: Array<{ id: string, pass: boolean, reason: string }>, reason: string|null, submission_id: string|null }} input
   */
  async function review(ctx, { agent, task_id, action, checks, reason, submission_id }) {
    /** @type {string} */
    let reviewed
    return act(ctx, {
      agent,
      task_id,
      async prepare(trx) {
        const violation = reviewViolation({ action, checks, item_keys: (await acceptanceOf(trx, task_id)).map((item) => item.id) })
        if (violation !== null) return refuse('rule_violation', violation)
        switch (action) {
          case 'accept':
          case 'reject': {
            const latest = await trx(SUBMISSIONS).where({ task_id }).orderBy('created_at', 'desc').first('id')
            if (latest === undefined) return refuse('state_conflict', 'There is no submission to review yet.')
            reviewed = latest.id
            return { action: { type: action } }
          }
          case 'select_winner': {
            const winner = submission_id === null ? undefined : await trx(SUBMISSIONS).where({ id: submission_id, task_id }).first('id', 'agent_id')
            if (winner === undefined) return refuse('rule_violation', 'select_winner requires the submission_id of a submission to this task.')
            reviewed = winner.id
            return { action: { type: 'select_winner', winner_agent_id: winner.agent_id } }
          }
          default:
            throw new Error(`Unhandled review action: ${action}, id: ${task_id}`)
        }
      },
      async record(trx, _before, _after, now) {
        const id = `rv_${randomUUID()}`
        await trx(REVIEWS).insert({ id, task_id, submission_id: reviewed, reviewer_kind: 'publisher', action, reason, created_at: now })
        await trx('vk_review_checks').insert(checks.map((check) => ({ review_id: id, item_key: check.id, pass: check.pass, reason: check.reason })))
        return { review_id: id }
      },
    })
  }

  /**
   * 持有人把任务转给同公会的另一个 Agent，对方接受后才换人；同一任务同时只有一个待确认的转让。
   * @param {object} _ctx
   * @param {{ agent: Agent, task_id: string, to_agent_id: string }} input
   * @returns {Promise<{ ok: true, transfer_id: string, expires_at: Date } | Refusal>}
   */
  async function requestTransfer(_ctx, { agent, task_id, to_agent_id }) {
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      const hidden = task === null ? refuse('not_found', `Task not found: ${task_id}`) : hiddenFrom(agent, task)
      if (task === null || hidden !== null) return /** @type {Refusal} */ (hidden)
      if (task.status !== 'claimed') return refuse('state_conflict', `Task is ${task.status}; only a claimed task can be transferred.`)
      if (task.holder_agent_id !== agent.id) return refuse('forbidden', 'Only the current holder can transfer.')
      const target = await trx('vk_agents').where({ id: to_agent_id, guild_id: task.guild_id, status: 'active' }).first('id')
      if (target === undefined || to_agent_id === agent.id) return refuse('rule_violation', 'to_agent_id must be another active agent in this guild.')
      if ((await trx(TRANSFERS).where({ task_id, status: 'pending' }).first('id')) !== undefined) return refuse('state_conflict', 'This task already has a pending transfer.')
      const now = new Date()
      const guild = await guildRow(trx, task.guild_id)
      const transfer = { id: `tr_${randomUUID()}`, task_id, from_agent_id: agent.id, to_agent_id, status: 'pending', expires_at: new Date(now.getTime() + guild.transfer_ttl_sec * 1000), created_at: now }
      await trx(TRANSFERS).insert(transfer)
      await insertEvent(trx, task, 'task.transfer_requested', to_agent_id, now)
      await insertJob(trx, task_id, 'transfer_expire', transfer.expires_at, now)
      return { ok: true, transfer_id: transfer.id, expires_at: transfer.expires_at }
    })
  }

  /**
   * 被转让方接受或拒绝。接受时走状态机换持有人（自接自单、并发上限、稀有度同样适用）。
   * @param {object} ctx
   * @param {{ agent: Agent, task_id: string, decision: 'accept'|'decline' }} input
   * @returns {Promise<{ ok: true, task: Task } | Refusal>}
   */
  async function answerTransfer(ctx, { agent, task_id, decision }) {
    /** @param {import('knex').Knex} trx @param {Date} now */
    const pendingForMe = async (trx, now) => {
      const transfer = await trx(TRANSFERS).where({ task_id, to_agent_id: agent.id, status: 'pending' }).first('id', 'expires_at')
      return transfer === undefined || transfer.expires_at <= now ? null : transfer
    }
    switch (decision) {
      case 'accept':
        return act(ctx, {
          agent,
          task_id,
          action: { type: 'transfer_accept' },
          prepare: async (trx, _task, now) => ((await pendingForMe(trx, now)) === null ? refuse('state_conflict', 'There is no pending transfer of this task to you.') : null),
          record: async (trx, _before, _after, now) => {
            await trx(TRANSFERS).where({ task_id, to_agent_id: agent.id, status: 'pending' }).update({ status: 'accepted', resolved_at: now })
          },
        })
      case 'decline':
        return db.transaction(async (trx) => {
          const task = await lockTask(trx, task_id)
          const now = new Date()
          if (task === null || (await pendingForMe(trx, now)) === null) return refuse('state_conflict', 'There is no pending transfer of this task to you.')
          await trx(TRANSFERS).where({ task_id, to_agent_id: agent.id, status: 'pending' }).update({ status: 'declined', resolved_at: now })
          return { ok: true, task }
        })
      default:
        throw new Error(`Unhandled transfer decision: ${decision}, id: ${task_id}`)
    }
  }

  /**
   * worker：到期仍待确认的转让记为 expired。
   * @param {object} _ctx
   * @param {{ task_id: string }} input
   */
  async function expireTransfers(_ctx, { task_id }) {
    const now = new Date()
    await db(TRANSFERS).where({ task_id, status: 'pending' }).where('expires_at', '<=', now).update({ status: 'expired', resolved_at: now })
  }

  /**
   * 负责人审核公司 Agent 发布的任务：通过上板，驳回退款。
   * @param {object} ctx
   * @param {{ user_id: string, task_id: string, decision: 'approve'|'reject' }} input
   * @returns {Promise<{ ok: true, task: Task } | Refusal>}
   */
  async function decideApproval(ctx, { user_id, task_id, decision }) {
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      if (task === null) return refuse('not_found', `Task not found: ${task_id}`)
      const guild = await trx('vk_guilds').where({ id: task.guild_id }).first('lead_user_id')
      if (guild.lead_user_id !== user_id) return refuse('forbidden', 'Only the guild lead can review this task.')
      return apply(ctx, trx, task, { type: decision === 'approve' ? 'approve' : 'reject_approval' }, { kind: 'lead', user_id }, undefined, new Date(), undefined)
    })
  }

  /**
   * 负责人名下公会里等待审核的任务。
   * @param {object} _ctx
   * @param {{ user_id: string }} input
   * @returns {Promise<Task[]>}
   */
  async function listPendingApprovals(_ctx, { user_id }) {
    const rows = await db(TASKS).where({ status: 'pending_approval' }).whereIn('guild_id', db('vk_guilds').where({ lead_user_id: user_id }).select('id')).orderBy('created_at', 'asc')
    return rows.map(taskFromRow)
  }

  /**
   * 运营裁决争议：结算给接单方（计件活为持有人，擂台赛须指明赢家）或全额退回发布方。
   * @param {object} ctx
   * @param {{ user_id: string, task_id: string, outcome: 'settled'|'refunded', winner_agent_id: string|null }} input
   * @returns {Promise<{ ok: true, task: Task } | Refusal>}
   */
  async function resolveDispute(ctx, { user_id, task_id, outcome, winner_agent_id }) {
    return db.transaction(async (trx) => {
      const task = await lockTask(trx, task_id)
      if (task === null) return refuse('not_found', `Task not found: ${task_id}`)
      const winner = outcome === 'settled' ? (task.holder_agent_id === null ? winner_agent_id : task.holder_agent_id) : null
      if (outcome === 'settled' && winner === null) return refuse('rule_violation', 'winner_agent_id is required to settle a contest dispute.')
      if (winner !== null && task.template === 'contest' && (await trx(ENTRIES).where({ task_id, agent_id: winner }).first('id')) === undefined) return refuse('rule_violation', 'winner_agent_id must be an entrant of this contest.')
      const now = new Date()
      return apply(ctx, trx, task, { type: 'resolve', outcome, winner_agent_id: winner }, { kind: 'operator', user_id }, undefined, now, async () => {
        await trx(DISPUTES).where({ task_id }).whereNot({ status: 'resolved' }).update({ status: 'resolved', resolution: JSON.stringify({ outcome, winner_agent_id: winner }), resolved_by_user_id: user_id, resolved_at: now })
      })
    })
  }

  /**
   * 运营的仲裁队列：未关闭的争议及其任务。
   * @param {object} _ctx
   * @returns {Promise<Array<{ dispute_id: string, status: string, ai_proposal: { recommendation: string, reason: string }|null, opened_by_agent_id: string, created_at: Date, task: Task }>>}
   */
  async function listOpenDisputes(_ctx) {
    const disputes = await db(DISPUTES).whereNot({ status: 'resolved' }).orderBy('created_at', 'asc')
    const tasks = disputes.length === 0 ? [] : await db(TASKS).whereIn('id', disputes.map((dispute) => dispute.task_id))
    return disputes.map((dispute) => ({ dispute_id: dispute.id, status: dispute.status, ai_proposal: dispute.ai_proposal, opened_by_agent_id: dispute.opened_by_agent_id, created_at: dispute.created_at, task: taskFromRow(tasks.find((task) => task.id === dispute.task_id)) }))
  }

  /**
   * worker：擂台赛验收窗口到期，由大模型按清单选赢家后结算；没有提交或无人全部通过则退款。
   * 模型调用在事务外进行（不占任务行锁）；随后的状态机会重新检查任务状态与到期时间。
   * @param {object} ctx
   * @param {{ task_id: string }} input
   */
  async function autoSelectWinner(ctx, { task_id }) {
    const row = await db(TASKS).where({ id: task_id }).first()
    if (row === undefined) throw new Error(`vk_tasks row missing for review_auto_accept, id: ${task_id}`)
    const task = taskFromRow(row)
    if (task.status !== 'reviewing' || /** @type {Date} */ (task.review_due_at) > new Date()) return
    const submissions = await db(SUBMISSIONS).where({ task_id }).select('id', 'agent_id', 'summary', 'artifacts')
    const verdict = submissions.length === 0
      ? { winner_submission_id: null, checks: [], reason: 'No submissions.' }
      : await judge.pickContestWinner({ task, acceptance: await acceptanceOf(db, task_id), submissions: submissions.map((item) => ({ id: item.id, summary: item.summary, artifacts: item.artifacts })) })
    const winner = submissions.find((item) => item.id === verdict.winner_submission_id)
    await actAsWorker(ctx, {
      task_id,
      action: { type: 'review_auto_accept', winner_agent_id: winner === undefined ? null : winner.agent_id },
      async record(trx, _before, _after, now) {
        if (winner === undefined) return
        const id = `rv_${randomUUID()}`
        await trx(REVIEWS).insert({ id, task_id, submission_id: winner.id, reviewer_kind: 'auto', action: 'select_winner', reason: verdict.reason, created_at: now })
        await trx('vk_review_checks').insert(verdict.checks.map((check) => ({ review_id: id, item_key: check.id, pass: check.pass, reason: check.reason })))
      },
    })
  }

  /**
   * worker：给未关闭且尚无意见的争议生成 AI 仲裁意见，进入运营队列。只写意见，不动账。
   * @param {object} _ctx
   * @param {{ task_id: string }} input
   */
  async function arbitrate(_ctx, { task_id }) {
    const dispute = await db(DISPUTES).where({ task_id, status: 'open' }).first('id')
    if (dispute === undefined) return
    const task = taskFromRow(await db(TASKS).where({ id: task_id }).first())
    const [submissions, reviews, comments] = await Promise.all([
      db(SUBMISSIONS).where({ task_id }).orderBy('round', 'asc').select('round', 'summary', 'artifacts'),
      db(REVIEWS).where({ task_id }).orderBy('created_at', 'asc').select('id', 'action', 'reason'),
      db(COMMENTS).where({ task_id }).orderBy('created_at', 'asc').select('body'),
    ])
    const checks = reviews.length === 0 ? [] : await db('vk_review_checks').whereIn('review_id', reviews.map((item) => item.id)).select('review_id', 'item_key', 'pass', 'reason')
    const proposal = await judge.proposeArbitration({
      task,
      acceptance: await acceptanceOf(db, task_id),
      submissions,
      reviews: reviews.map((item) => ({ action: item.action, reason: item.reason, checks: checks.filter((check) => check.review_id === item.id).map((check) => ({ id: check.item_key, pass: check.pass === 1, reason: check.reason })) })),
      comments,
    })
    await db(DISPUTES).where({ id: dispute.id, status: 'open' }).update({ status: 'ai_proposed', ai_proposal: JSON.stringify(proposal) })
  }

  /**
   * 到期前提醒：任务仍在等待对应动作时给当事人发事件；状态已变则无事可做。
   * @param {object} _ctx
   * @param {{ task_id: string, kind: 'checkpoint_warn'|'review_warn' }} input
   */
  async function warn(_ctx, { task_id, kind }) {
    const row = await db(TASKS).where({ id: task_id }).first()
    if (row === undefined) throw new Error(`vk_tasks row missing for ${kind}, id: ${task_id}`)
    const task = taskFromRow(row)
    const now = new Date()
    switch (kind) {
      case 'checkpoint_warn':
        if (task.status === 'claimed') await insertEvent(db, task, 'task.checkpoint_due', task.holder_agent_id, now)
        break
      case 'review_warn':
        if (task.status === 'submitted' || task.status === 'reviewing') await insertEvent(db, task, 'task.review_due', task.publisher_agent_id, now)
        break
      default:
        throw new Error(`Unhandled warn kind: ${kind}, id: ${task_id}`)
    }
  }

  return { publish, act, actAsWorker, checkpoint, submit, review, requestTransfer, answerTransfer, expireTransfers, decideApproval, listPendingApprovals, resolveDispute, listOpenDisputes, autoSelectWinner, arbitrate, warn, list, get, addComment, listComments, homeFor }
}

/**
 * 成员可见性：别的公会的任务 → forbidden；待审核任务只对发布方可见，其他成员按不存在处理。
 * @param {Agent} agent
 * @param {Task} task
 * @returns {Refusal|null}
 */
function hiddenFrom(agent, task) {
  if (agent.guild_id !== task.guild_id) return refuse('forbidden', 'You are not a member of this guild.')
  if (task.status === 'pending_approval' && task.publisher_agent_id !== agent.id) return refuse('not_found', `Task not found: ${task.id}`)
  return null
}

/**
 * @param {import('knex').Knex} trx
 * @param {string} task_id
 * @returns {Promise<Task|null>}
 */
async function lockTask(trx, task_id) {
  const row = await trx(TASKS).where({ id: task_id }).forUpdate().first()
  return row === undefined ? null : taskFromRow(row)
}

/**
 * @param {import('knex').Knex} runner
 * @param {string} guild_id
 */
async function guildRow(runner, guild_id) {
  const guild = await runner('vk_guilds').where({ id: guild_id }).first('require_lead_approval', 'checkpoint_interval_sec', 'piece_review_window_sec', 'contest_review_window_sec', 'transfer_ttl_sec')
  if (guild === undefined) throw new Error(`vk_guilds row missing, id: ${guild_id}`)
  return guild
}

/**
 * @param {import('knex').Knex} runner
 * @param {string} task_id
 * @returns {Promise<AcceptanceItem[]>}
 */
async function acceptanceOf(runner, task_id) {
  const items = await runner(ITEMS).where({ task_id }).orderBy('ordinal', 'asc').select('item_key', 'text')
  return items.map((item) => ({ id: item.item_key, text: item.text }))
}

/**
 * @param {import('knex').Knex} trx
 * @param {{ id: string, guild_id: string }} task
 * @param {string} type
 * @param {string|null} target_agent_id null 表示发给整个公会
 * @param {Date} now
 */
export async function insertEvent(trx, task, type, target_agent_id, now) {
  const [event_id] = await trx(EVENTS).insert({ guild_id: task.guild_id, target_agent_id, type, task_id: task.id, payload: '{}', created_at: now })
  // 给登记了 Webhook 的目标 Agent（公会广播则是全体成员）各排一条投递，由 worker 签名发送。
  const receivers = trx('vk_agents').whereNotNull('webhook_url').where({ status: 'active' }).where(target_agent_id === null ? { guild_id: task.guild_id } : { id: target_agent_id }).select('id')
  const deliveries = (await receivers).map((agent) => ({ id: `wd_${randomUUID()}`, event_id, agent_id: agent.id, status: 'pending', attempts: 0, next_attempt_at: now, created_at: now }))
  if (deliveries.length > 0) await trx('vk_webhook_deliveries').insert(deliveries)
}

/**
 * 同一对象同一类型同一时刻的 job 只排一条（dedupe_key UNIQUE）。
 * @param {import('knex').Knex} trx
 * @param {string} task_id job 的 ref_id：任务类 job 是任务 id，求助通知是求助 id
 * @param {string} kind
 * @param {Date} run_at
 * @param {Date} now
 */
export async function insertJob(trx, task_id, kind, run_at, now) {
  await trx(JOBS)
    .insert({ id: `j_${randomUUID()}`, kind, ref_id: task_id, run_at, status: 'pending', attempts: 0, dedupe_key: `${kind}:${task_id}:${run_at.toISOString()}`, created_at: now })
    .onConflict('dedupe_key')
    .ignore()
}

/**
 * @template {Refusal['code']} C
 * @param {C} code
 * @param {string} hint
 * @returns {{ ok: false, code: C, hint: string }}
 */
function refuse(code, hint) {
  return { ok: false, code, hint }
}