dsh-ops-cron/lib/fire.js

/**
 * Fire path: new root Session, wrapped prompt, hide from workspace list, history.
 * Collaborators are injected so tests drive the same functions as Host apply.
 */

import { randomUUID } from 'node:crypto'
import { wrapScheduledPrompt, scheduledPromptSource } from './prompt.js'
import { decideDispatch } from './scheduler.js'
import { appendRun, getJob, patchRun, upsertJob } from './store.js'
import { ACTIVE_RUN_STATUSES } from './fire-status.js'
import {
  findLabelOverlapRun,
  isRetriggerable,
  normalizeLabels,
  projectJobState,
} from './monitor.js'

export { ACTIVE_RUN_STATUSES } from './fire-status.js'

export function createRunRecord(job, decision, now, trigger) {
  return {
    id: randomUUID(),
    jobId: job.id,
    scheduledAt: decision.scheduledAt ?? now,
    actualAt: now,
    status: 'queued',
    stateEnteredAt: now,
    exitedAt: null,
    trigger: trigger || 'schedule',
    sessionId: null,
    error: null,
    summary: '',
    reason: decision.reason || null,
    outputRef: null,
  }
}

export const TITLE_PREFIX = '定时任务 · '

export function extractAssistantText(messages, maxChars = 4000) {
  if (!Array.isArray(messages)) return ''
  for (let i = messages.length - 1; i >= 0; i--) {
    const message = messages[i]
    if (!message || message.role !== 'assistant') continue
    const parts = []
    for (const block of message.content || []) {
      if (block && block.type === 'text' && typeof block.text === 'string' && block.text.trim()) {
        parts.push(block.text.trim())
      }
    }
    const text = parts.join('\n').trim()
    if (!text) continue
    return text.length > maxChars ? `${text.slice(0, maxChars)}…` : text
  }
  return ''
}

export function publicJob(job, runs = null) {
  if (!job) return null
  const jobRuns = Array.isArray(runs)
    ? runs.filter((run) => run?.jobId === job.id)
    : null
  const projection = jobRuns
    ? projectJobState(job, jobRuns)
    : projectJobState(job, job.lastStatus
      ? [{ status: job.lastStatus, stateEnteredAt: job.lastRunAt, actualAt: job.lastRunAt, exitedAt: job.lastRunAt }]
      : [])
  // When runs not provided, prefer lastStatus-based active hint only for running/queued tip.
  let stateInfo = projection
  if (!jobRuns && (job.lastStatus === 'running' || job.lastStatus === 'queued')) {
    stateInfo = {
      state: job.lastStatus === 'queued' ? 'pending' : 'running',
      stateEnteredAt: job.lastRunAt || job.updatedAt || null,
      activeRunId: null,
      stuck: false,
    }
  } else if (!jobRuns && job.enabled === false) {
    stateInfo = {
      state: 'paused',
      stateEnteredAt: job.updatedAt || job.createdAt || null,
      activeRunId: null,
      stuck: false,
    }
  } else if (!jobRuns && job.schedule?.kind === 'at' && job.nextRunAt == null
    && (job.lastStatus === 'succeeded' || job.lastStatus === 'failed' || job.lastStatus === 'skipped')) {
    stateInfo = {
      state: job.lastStatus === 'succeeded' ? 'succeeded' : 'failed',
      stateEnteredAt: job.lastRunAt || null,
      activeRunId: null,
      stuck: false,
    }
  } else if (!jobRuns) {
    stateInfo = {
      state: 'idle',
      stateEnteredAt: job.lastRunAt || job.updatedAt || job.createdAt || null,
      activeRunId: null,
      stuck: false,
    }
  }
  return {
    id: job.id,
    name: job.name,
    prompt: job.prompt,
    enabled: job.enabled !== false,
    cwd: job.cwd || '',
    timeoutMinutes: job.timeoutMinutes || 10,
    provider: job.provider || '',
    model: job.model || '',
    reasoningEffort: job.reasoningEffort || '',
    agentPreset: job.agentPreset || '',
    agentAccess: job.agentAccess === 'deny' ? 'deny' : 'allow',
    permissionPreset: job.permissionPreset || '',
    ownerEmpNo: job.ownerEmpNo || '',
    ownerDisplayName: job.ownerDisplayName || '',
    delivery: job.delivery && job.delivery.kind === 'im'
      ? {
        kind: 'im',
        botId: job.delivery.botId,
        targetId: job.delivery.targetId,
        ...job.delivery.mentionJid ? { mentionJid: job.delivery.mentionJid } : {},
        ...job.delivery.mentionName ? { mentionName: job.delivery.mentionName } : {},
      }
      : { kind: 'dsh' },
    mirrorToSession: job.mirrorToSession === true,
    labels: normalizeLabels(job.labels),
    ...job.watch ? { watch: job.watch } : {},
    ...job.progress ? { progress: job.progress } : {},
    ...job.report ? { report: job.report } : {},
    persistHistory: job.persistHistory || { kind: 'retain', limit: 200 },
    state: stateInfo.state,
    stateEnteredAt: stateInfo.stateEnteredAt,
    stuck: stateInfo.stuck === true,
    retriggerable: isRetriggerable(job, jobRuns || []),
    ...job.origin ? { origin: job.origin } : {},
    schedule: job.schedule,
    createdAt: job.createdAt,
    updatedAt: job.updatedAt,
    lastRunAt: job.lastRunAt,
    lastStatus: job.lastStatus,
    nextRunAt: job.nextRunAt,
  }
}

/**
 * Claim one occurrence: overlap → skipped history row; otherwise queued run.
 * Also skips when another job with the same role+task labels is already active.
 */
export function claimOccurrence(state, jobId, now, trigger, policies) {
  const job = getJob(state, jobId)
  if (!job) {
    const error = new Error('job not found')
    error.code = 'NOT_FOUND'
    throw error
  }
  const runs = (state.runs || []).filter((run) => run.jobId === jobId)
  let decision = trigger === 'run-now'
    ? decideRunNow(job, runs, now, policies)
    : decideDispatch({
      job,
      runs,
      now,
      overlapPolicy: policies?.overlapPolicy || state.settings?.overlapPolicy,
      misfirePolicy: policies?.misfirePolicy || state.settings?.misfirePolicy,
      graceMs: policies?.graceMs,
    })

  if (decision.action === 'fire') {
    const overlap = findLabelOverlapRun(state, job)
    if (overlap) {
      decision = {
        action: 'skip',
        reason: 'label_overlap',
        scheduledAt: decision.scheduledAt ?? now,
        nextRunAt: Object.hasOwn(decision, 'nextRunAt') ? decision.nextRunAt : job.nextRunAt,
      }
    }
  }

  if (decision.action === 'wait') {
    return { state, job: publicJob(job, state.runs), decision, run: null }
  }

  const run = createRunRecord(job, decision, now, trigger)
  if (decision.action === 'skip') {
    run.status = 'skipped'
    run.stateEnteredAt = now
    run.exitedAt = now
    run.reason = decision.reason
    const oneshot = job.schedule?.kind === 'at'
    if (decision.reason === 'overlap') {
      run.summary = 'Skipped because a run is already queued or running'
    } else if (decision.reason === 'label_overlap') {
      run.summary = 'Skipped because another job with the same role+task labels is already running'
    } else if (decision.reason === 'misfire' && oneshot) {
      run.summary = 'Skipped: one-shot time missed (host downtime or delayed tick; no backlog). nextRunAt cleared.'
      run.error = 'misfire:one-shot'
    } else if (decision.reason === 'misfire') {
      run.summary = 'Skipped missed occurrence after host downtime (no backlog)'
      run.error = 'misfire'
    } else {
      run.summary = `Skipped (${decision.reason || 'policy'})`
    }
    const alreadySkipped = runs.some((row) => (
      row
      && row.status === 'skipped'
      && row.reason === decision.reason
      && row.scheduledAt === decision.scheduledAt
    ))
    const nextJob = {
      ...job,
      updatedAt: now,
      lastRunAt: now,
      lastStatus: 'skipped',
      nextRunAt: Object.hasOwn(decision, 'nextRunAt') ? decision.nextRunAt : job.nextRunAt,
    }
    let next = upsertJob(state, nextJob)
    if (!alreadySkipped) next = appendRun(next, run)
    return { state: next, job: publicJob(nextJob, next.runs), decision, run: alreadySkipped ? null : run }
  }

  run.status = 'queued'
  run.stateEnteredAt = now
  const nextJob = {
    ...job,
    updatedAt: now,
    nextRunAt: trigger === 'run-now'
      ? job.nextRunAt
      : (Object.hasOwn(decision, 'nextRunAt') ? decision.nextRunAt : job.nextRunAt),
  }
  let next = upsertJob(state, nextJob)
  next = appendRun(next, run)
  return { state: next, job: publicJob(nextJob, next.runs), decision, run }
}

function decideRunNow(job, runs, now, policies) {
  const overlapPolicy = policies?.overlapPolicy || 'skip'
  if (overlapPolicy === 'skip' && runs.some((run) => run.status === 'queued' || run.status === 'running')) {
    return { action: 'skip', reason: 'overlap', scheduledAt: now, nextRunAt: job.nextRunAt }
  }
  return { action: 'fire', scheduledAt: now, nextRunAt: job.nextRunAt }
}

function failRun(state, runId, job, now, message) {
  return settleRun(state, runId, {
    status: 'failed',
    error: message,
    summary: message,
  }, now)
}

/** Persist a terminal status so overlap skip can release the job. */
export function settleRun(state, runId, terminal, now) {
  const run = (state.runs || []).find((row) => row.id === runId)
  if (!run) return state
  const job = getJob(state, run.jobId)
  const status = terminal?.status && !ACTIVE_RUN_STATUSES.has(terminal.status)
    ? terminal.status
    : 'succeeded'
  // Keep stateEnteredAt / actualAt from queued→running so cron_runs duration is real.
  // Only stamp exitedAt (and lastRunAt on the job) at settle time.
  let next = patchRun(state, runId, {
    status,
    summary: terminal?.summary || '',
    error: terminal?.error || null,
    actualAt: run.actualAt || now,
    stateEnteredAt: run.stateEnteredAt || run.actualAt || now,
    exitedAt: now,
    sessionId: run.sessionId,
    ...terminal?.outputRef ? { outputRef: terminal.outputRef } : {},
  })
  if (job) {
    next = upsertJob(next, {
      ...job,
      lastRunAt: now,
      lastStatus: status,
      updatedAt: now,
    })
  }
  return next
}

/**
 * Execute an already-claimed queued run via injected session/archive helpers.
 * Performs IO through `deps`; the caller persists each returned state.
 *
 * @param {object} state
 * @param {string} runId
 * @param {object} deps
 * @param {() => number} [deps.now]
 * @param {(args: object) => Promise<{ sessionId: string, status?: string, summary?: string }>} deps.createAndPrompt
 * @param {(sessionId: string) => Promise<void>} [deps.archiveSession]
 * @param {(sessionId: string) => Promise<{ status?: string, summary?: string, error?: string }>} [deps.waitForTurn]
 */
export async function executeClaimedRun(state, runId, deps) {
  const run = (state.runs || []).find((row) => row.id === runId)
  if (!run) {
    const error = new Error('run not found')
    error.code = 'NOT_FOUND'
    throw error
  }
  const job = getJob(state, run.jobId)
  if (!job) {
    const error = new Error('job not found')
    error.code = 'NOT_FOUND'
    throw error
  }
  const clock = deps.now || (() => Date.now())
  const mark = clock()
  let next = patchRun(state, runId, {
    status: 'running',
    actualAt: mark,
    stateEnteredAt: mark,
    exitedAt: null,
  })
  next = upsertJob(next, { ...job, lastRunAt: mark, lastStatus: 'running', updatedAt: mark })

  const text = wrapScheduledPrompt(job.prompt, {
    jobName: job.name,
    jobId: job.id,
    runId,
    scheduledAt: new Date(run.scheduledAt).toISOString(),
  })
  const source = scheduledPromptSource()

  if (typeof deps.createAndPrompt !== 'function') {
    next = failRun(next, runId, job, clock(), 'session port unavailable')
    return {
      state: next,
      run: (next.runs || []).find((row) => row.id === runId),
      job: publicJob(getJob(next, job.id), next.runs),
    }
  }

  try {
    const created = await deps.createAndPrompt({
      job,
      run: { ...run, status: 'running' },
      text,
      source,
    })
    const sessionId = created?.sessionId
    if (!sessionId) throw new Error('createAndPrompt did not return sessionId')
    next = patchRun(next, runId, { sessionId })
    if (typeof deps.archiveSession === 'function') {
      await deps.archiveSession(sessionId)
    }
    const reported = created.status || 'succeeded'
    if (typeof deps.waitForTurn === 'function' && ACTIVE_RUN_STATUSES.has(reported)) {
      const waited = await deps.waitForTurn(sessionId)
      const ended = clock()
      next = settleRun(next, runId, waited || { status: 'succeeded', summary: created.summary }, ended)
      return {
        state: next,
        run: (next.runs || []).find((row) => row.id === runId),
        job: publicJob(getJob(next, job.id), next.runs),
      }
    }
    if (ACTIVE_RUN_STATUSES.has(reported)) {
      // Leave running so overlap skip still works while a turn is in flight.
      // Caller (dispatchRun / live waitForTurn) must settle to a terminal status.
      return {
        state: next,
        run: (next.runs || []).find((row) => row.id === runId),
        job: publicJob(getJob(next, job.id), next.runs),
      }
    }
    const ended = clock()
    next = settleRun(next, runId, {
      status: reported,
      summary: created.summary || 'Dispatched to a new session',
      error: created.error || null,
    }, ended)
    return {
      state: next,
      run: (next.runs || []).find((row) => row.id === runId),
      job: publicJob(getJob(next, job.id), next.runs),
    }
  } catch (error) {
    const message = error instanceof Error ? error.message : String(error)
    next = failRun(next, runId, job, clock(), message)
    return {
      state: next,
      run: (next.runs || []).find((row) => row.id === runId),
      job: publicJob(getJob(next, job.id), next.runs),
    }
  }
}

export function interruptActiveRuns(state, now, reason = 'host_interrupted') {
  let next = state
  for (const run of state.runs || []) {
    if (run.status !== 'queued' && run.status !== 'running') continue
    next = patchRun(next, run.id, {
      status: 'failed',
      error: reason,
      summary: 'Host stopped before this run finished',
      actualAt: run.actualAt || now,
      stateEnteredAt: run.stateEnteredAt || run.actualAt || now,
      exitedAt: now,
    })
    const job = getJob(next, run.jobId)
    if (job) {
      next = upsertJob(next, { ...job, lastStatus: 'failed', lastRunAt: now, updatedAt: now })
    }
  }
  return next
}

export { wrapScheduledPrompt }