scrum4me-docker/bin/run-one-job.ts

#!/usr/bin/env tsx
// run-one-job.ts — handelt één geclaimde Scrum4Me-ClaudeJob af.
//
// Architectuur (zie docs/plans/queue-loop-extraction.md in Scrum4Me-repo):
// scrum4me-docker/bin/run-agent.sh roept dit script per iteratie aan.
// Stappen:
//   1. getAuth → resolved userId/tokenId uit SCRUM4ME_TOKEN.
//   2. quota-probe (was Claude's verantwoordelijkheid in CLAUDE.md stappen 0.x).
//   3. resetStaleClaimedJobs → tryClaimJob, met LISTEN-fallback (270s) bij lege queue.
//   4. getFullJobContext → resolved JobConfig + kind-specifieke payload.
//   5. attachWorktreeToJob (alleen TASK_IMPLEMENTATION).
//   6. Schrijf payload naar /tmp/job-<id>/payload.json.
//   7. Bouw CLI-args uit ctx.config + mapBudgetToEffort.
//   8. setInterval(60s) lease-renewal voor SPRINT_IMPLEMENTATION.
//   9. spawn 'claude' met inherited stdio + scan voor token-expiry-patterns.
//  10. try/finally: bij Claude-exit≠0 zonder update_job_status → rollbackClaim.
//  11. cleanup payload + prisma.$disconnect().
//
// Exit-codes:
//   0  = job afgehandeld of geen job binnen wait-deadline (idle)
//   1  = generieke fout (claim, context-fetch, worktree, spawn)
//   3  = TOKEN_EXPIRED detected → run-agent.sh schrijft TOKEN_EXPIRED marker

import { spawn, spawnSync } from 'node:child_process'
import { mkdirSync, rmSync, writeFileSync } from 'node:fs'

import { Client as PgClient } from 'pg'

import { getAuth } from '/opt/scrum4me-mcp/src/auth.js'
import { prisma } from '/opt/scrum4me-mcp/src/prisma.js'
import {
  attachWorktreeToJob,
  getFullJobContext,
  resetStaleClaimedJobs,
  rollbackClaim,
  tryClaimJob,
} from '/opt/scrum4me-mcp/src/tools/wait-for-job.js'
import { releaseLocksOnTerminal } from '/opt/scrum4me-mcp/src/git/job-locks.js'
import { mapBudgetToEffort } from '/opt/scrum4me-mcp/src/lib/job-config.js'
import { getKindPromptText } from '/opt/scrum4me-mcp/src/lib/kind-prompts.js'

// ----- logging --------------------------------------------------------
const log = (msg: string) =>
  console.log(`${new Date().toISOString()} [run-one-job] ${msg}`)
const logError = (msg: string) =>
  console.error(`${new Date().toISOString()} [run-one-job] ERROR ${msg}`)

// ----- constants ------------------------------------------------------
const WAIT_DEADLINE_SECONDS = 270 // ruim binnen MAX_WAIT_SECONDS van wait_for_job
const POLL_INTERVAL_MS = 5000
const HEARTBEAT_INTERVAL_MS = 60_000
const MCP_CONFIG = '/opt/agent/mcp-config.json'
const QUOTA_PROBE_PATH = '/opt/agent/bin/worker-quota-probe.sh'
const QUOTA_BACKOFF_CAP_MS = 30 * 60 * 1000
const TOKEN_EXPIRY_PATTERNS: RegExp[] = [
  /invalid_api_key/i,
  /authentication.*failed/i,
  /401.*unauthor/i,
  /OAuth.*expired/i,
]

// ----- quota probe ----------------------------------------------------
// Soft-fail: als de probe geen rate-limit-headers krijgt (sommige Anthropic
// endpoints retourneren ze niet) of een transient netwerkfout heeft, log
// een warning en ga door. Alleen bij gemeten quota-overschrijding sleepen.
// Dit spiegelt het gedrag van CLAUDE.md stap 0.4 ("anders: ga door").
async function quotaProbe(userId: string): Promise<void> {
  const probe = spawnSync(QUOTA_PROBE_PATH, [], { encoding: 'utf8' })
  if (probe.status !== 0) {
    log(
      `quota probe non-zero status=${probe.status} stdout=${probe.stdout.slice(0, 200).trim()} — continuing without gate`,
    )
    return
  }
  let parsed: { pct?: number; limit?: number; remaining?: number; reset_at_iso?: string; error?: string }
  try {
    parsed = JSON.parse(probe.stdout)
  } catch {
    log(`quota probe stdout not JSON (continuing): ${probe.stdout.slice(0, 200)}`)
    return
  }
  if (parsed.pct === undefined) {
    log(`quota probe no pct (continuing): error=${parsed.error ?? '-'}`)
    return
  }

  const user = await prisma.user.findUnique({
    where: { id: userId },
    select: { min_quota_pct: true },
  })
  const minPct = user?.min_quota_pct ?? 0

  log(
    `quota probe pct=${parsed.pct} min_quota_pct=${minPct} ` +
      `reset_at=${parsed.reset_at_iso ?? '-'}`,
  )

  if (parsed.pct < minPct) {
    let sleepMs = QUOTA_BACKOFF_CAP_MS
    if (parsed.reset_at_iso) {
      const resetAt = new Date(parsed.reset_at_iso).getTime()
      const delta = resetAt - Date.now()
      if (delta > 0 && delta < sleepMs) sleepMs = delta
    }
    log(`quota below min — sleeping ${Math.round(sleepMs / 1000)}s until reset`)
    await new Promise((resolve) => setTimeout(resolve, sleepMs))
  }
}

// ----- LISTEN-fallback voor lege queue -------------------------------
async function waitForEnqueue(userId: string): Promise<void> {
  const dburl = process.env.DATABASE_URL
  if (!dburl) throw new Error('DATABASE_URL not set')
  const client = new PgClient({ connectionString: dburl })
  await client.connect()
  await client.query('LISTEN scrum4me_changes')

  const deadline = Date.now() + WAIT_DEADLINE_SECONDS * 1000

  try {
    while (Date.now() < deadline) {
      await new Promise<void>((resolve) => {
        const pollTimer = setTimeout(resolve, POLL_INTERVAL_MS)
        const onNotify = (msg: { payload?: string }) => {
          try {
            const payload = JSON.parse(msg.payload ?? '{}')
            if (
              payload.type === 'claude_job_enqueued' &&
              payload.user_id === userId
            ) {
              clearTimeout(pollTimer)
              client.removeListener('notification', onNotify)
              resolve()
            }
          } catch {
            // ignore parse errors
          }
        }
        client.on('notification', onNotify)
      })
      // Out of the inner promise — caller will retry tryClaimJob.
      return
    }
  } finally {
    await client.end().catch(() => {})
  }
}

// ----- main -----------------------------------------------------------
async function main(): Promise<number> {
  log('claim attempt starting')
  const { userId, tokenId } = await getAuth()
  log(`auth ok user_id=${userId} token_id=${tokenId}`)

  // 1. Quota probe (gate vóór elke claim).
  try {
    await quotaProbe(userId)
  } catch (err) {
    logError(`quota probe error: ${(err as Error).message}`)
    return 1
  }

  // 2. Reset stale claims, then attempt to claim.
  await resetStaleClaimedJobs(userId)
  let jobId = await tryClaimJob(userId, tokenId)
  if (!jobId) {
    log(`no job claimed — LISTEN scrum4me_changes deadline=${WAIT_DEADLINE_SECONDS}s`)
    await waitForEnqueue(userId)
    await resetStaleClaimedJobs(userId)
    jobId = await tryClaimJob(userId, tokenId)
  }
  if (!jobId) {
    log(`claim timeout after ${WAIT_DEADLINE_SECONDS}s — exiting 0`)
    return 0
  }

  log(`claimed job_id=${jobId}`)

  // 3. Resolve full context.
  let ctx: Awaited<ReturnType<typeof getFullJobContext>> = null
  try {
    ctx = await getFullJobContext(jobId)
  } catch (err) {
    logError(`getFullJobContext error job_id=${jobId} ${(err as Error).message}`)
    log(`rollback claim job_id=${jobId} reason=context_fetch_failed`)
    await rollbackClaim(jobId)
    return 1
  }
  if (!ctx) {
    logError(`getFullJobContext returned null for job_id=${jobId}`)
    await rollbackClaim(jobId)
    return 1
  }

  // 4. Attach worktree for TASK_IMPLEMENTATION; sprint/idea-jobs hebben hun
  //    eigen worktree-pad al ingevuld door getFullJobContext.
  // We werken hier met `any` omdat de return-type van getFullJobContext een
  // discriminated union is en TypeScript hier zonder kind-narrow geen velden
  // exposed; de runtime checks dekken alle paden af.
  const ctxAny = ctx as any
  let worktreePath: string | null =
    ctxAny.worktree_path ?? ctxAny.primary_worktree_path ?? null

  if (ctx.kind === 'TASK_IMPLEMENTATION') {
    if (!ctxAny.story || !ctxAny.task) {
      logError(`TASK_IMPLEMENTATION job has incomplete story/task context`)
      await rollbackClaim(jobId)
      await releaseLocksOnTerminal(jobId)
      return 1
    }
    const wt = await attachWorktreeToJob(
      ctxAny.product.id,
      jobId,
      ctxAny.story.id,
      ctxAny.task.repo_url,
    )
    if ('error' in wt) {
      logError(`attachWorktreeToJob error job_id=${jobId} ${wt.error}`)
      log(`rollback claim job_id=${jobId} reason=worktree_attach_failed`)
      await rollbackClaim(jobId)
      await releaseLocksOnTerminal(jobId)
      return 1
    }
    worktreePath = wt.worktree_path
    ctxAny.worktree_path = wt.worktree_path
    ctxAny.branch_name = wt.branch_name
    log(`worktree path=${wt.worktree_path} branch=${wt.branch_name}`)
  } else if (worktreePath) {
    log(`worktree path=${worktreePath} (pre-resolved)`)
  }

  // 5. Resolved config — log voor audit.
  const cfg = ctx.config
  const effort = mapBudgetToEffort(cfg.thinking_budget)
  log(
    `config job_id=${jobId} model=${cfg.model} ` +
      `permission_mode=${cfg.permission_mode} ` +
      `thinking_budget=${cfg.thinking_budget} effort=${effort ?? '-'} ` +
      `max_turns=${cfg.max_turns ?? 'null'} ` +
      `allowed_tools_count=${cfg.allowed_tools?.length ?? 0}`,
  )

  // 6. Write payload to /tmp/job-<id>/payload.json.
  const payloadDir = `/tmp/job-${jobId}`
  const payloadPath = `${payloadDir}/payload.json`
  mkdirSync(payloadDir, { recursive: true })
  const payloadJson = JSON.stringify(ctx, null, 2)
  writeFileSync(payloadPath, payloadJson, 'utf8')
  log(
    `payload written path=${payloadPath} size_bytes=${Buffer.byteLength(payloadJson)}`,
  )

  // 7. Build CLI args.
  const promptText = getKindPromptText(ctx.kind).replace('$PAYLOAD_PATH', payloadPath)
  const args: string[] = [
    '-p',
    promptText,
    '--model',
    cfg.model,
    '--permission-mode',
    cfg.permission_mode,
    '--allowedTools',
    (cfg.allowed_tools ?? []).join(','),
    '--mcp-config',
    MCP_CONFIG,
    '--add-dir',
    '/opt/agent',
    '--output-format',
    'text',
  ]
  if (effort) args.push('--effort', effort)

  const cwd = worktreePath ?? '/opt/agent'
  // Log args zonder de volledige prompt-tekst (kan kilo's groot zijn).
  const argsForLog = args
    .map((a, i) => (i === 1 ? `<prompt-${promptText.length}-chars>` : a))
    .join(' ')
  log(`spawn claude job_id=${jobId} cwd=${cwd} args="${argsForLog}"`)

  // 8. Lease-renewal heartbeat for SPRINT_IMPLEMENTATION.
  let heartbeatTimer: NodeJS.Timeout | null = null
  if (ctx.kind === 'SPRINT_IMPLEMENTATION') {
    heartbeatTimer = setInterval(() => {
      prisma
        .$executeRaw`UPDATE claude_jobs SET lease_until = NOW() + INTERVAL '5 minutes' WHERE id = ${jobId}`
        .then(() => {
          log(
            `heartbeat tick job_id=${jobId} lease_until=${new Date(
              Date.now() + 5 * 60_000,
            ).toISOString()}`,
          )
        })
        .catch((err: Error) => {
          logError(`heartbeat error: ${err.message}`)
        })
    }, HEARTBEAT_INTERVAL_MS)
  }

  // 9. Spawn Claude.
  const start = Date.now()
  let exitCode: number | null = null
  let stdoutBuf = ''
  try {
    await new Promise<void>((resolve, reject) => {
      const child = spawn('claude', args, { cwd })
      child.stdout.on('data', (chunk) => {
        const s = chunk.toString()
        process.stdout.write(s)
        stdoutBuf += s
      })
      child.stderr.on('data', (chunk) => {
        const s = chunk.toString()
        process.stderr.write(s)
        stdoutBuf += s
      })
      child.on('error', (err) => reject(err))
      child.on('close', (code) => {
        exitCode = code
        resolve()
      })
    })
  } catch (err) {
    logError(`spawn error: ${(err as Error).message}`)
    if (heartbeatTimer) clearInterval(heartbeatTimer)
    await rollbackClaim(jobId).catch(() => {})
    await releaseLocksOnTerminal(jobId).catch(() => {})
    return 1
  } finally {
    if (heartbeatTimer) clearInterval(heartbeatTimer)
  }

  const durationMs = Date.now() - start
  log(
    `claude done job_id=${jobId} exit_code=${exitCode ?? 'null'} ` +
      `duration_ms=${durationMs} wall_clock_seconds=${Math.round(durationMs / 1000)}`,
  )

  // 10. Token-expiry detection.
  let tokenExpired = false
  for (const pat of TOKEN_EXPIRY_PATTERNS) {
    if (pat.test(stdoutBuf)) {
      tokenExpired = true
      log(`TOKEN_EXPIRED detected pattern="${pat.source}" exiting code=3`)
      break
    }
  }

  // 11. Rollback claim if Claude exited non-zero without updating job-status.
  // PBI-50 lease-driven recovery vangt resterende stale jobs op na 5 min.
  if (exitCode !== 0 && !tokenExpired) {
    const jobNow = await prisma.claudeJob.findUnique({
      where: { id: jobId },
      select: { status: true },
    })
    if (jobNow?.status === 'CLAIMED' || jobNow?.status === 'RUNNING') {
      log(
        `rollback claim job_id=${jobId} reason=claude_exit_${exitCode}_without_status_update`,
      )
      await rollbackClaim(jobId).catch(() => {})
      await releaseLocksOnTerminal(jobId).catch(() => {})
    }
  }

  // 12. Cleanup payload directory.
  try {
    rmSync(payloadDir, { recursive: true, force: true })
  } catch {
    // non-fatal
  }
  log(`cleanup payload_removed=true heartbeat_stopped=${heartbeatTimer !== null}`)

  if (tokenExpired) return 3
  return exitCode ?? 1
}

// ----- entry ----------------------------------------------------------
process.on('SIGTERM', () => {
  prisma.$disconnect().finally(() => process.exit(143))
})

main()
  .then(async (code) => {
    log(`exit code=${code}`)
    await prisma.$disconnect().catch(() => {})
    process.exit(code)
  })
  .catch(async (err) => {
    logError(`fatal: ${(err as Error).stack ?? err}`)
    await prisma.$disconnect().catch(() => {})
    process.exit(1)
  })