BountyBoard/services/work-performer/server.js

#!/usr/bin/env node
// work-performer: standalone placeholder Work Performer Service (§5.2, §9.2).
// Plain Node http server, no framework. For each job it prepares
// /work/{jobId}/TASK.md and runs the Claude Code CLI; if the CLI is missing
// or fails to start (e.g. no credentials mounted), it produces a simulated
// result so the whole flow stays demonstrable offline. The HTTP contract —
// not this implementation — is the deliverable.
'use strict';

const http = require('http');
const https = require('https');
const crypto = require('crypto');
const fs = require('fs');
const path = require('path');
const { execFile, execFileSync } = require('child_process');

const PORT = parseInt(process.env.PORT || '8091', 10);
const TOKEN = process.env.WORK_PERFORMER_TOKEN || '';
const WORK_DIR = process.env.WORK_DIR || '/work';
const PUBLIC_BASE = process.env.PUBLIC_BASE_URL || `http://work-performer:${PORT}`;

const jobs = new Map(); // jobId -> {status, taskId, request, startedAt, finishedAt, cancel}
const queue = [];
let running = false; // single-job concurrency (§9.2)

const log = (msg, extra) =>
  console.log(JSON.stringify({ ts: new Date().toISOString(), msg, ...extra }));

function json(res, status, body) {
  const data = JSON.stringify(body);
  res.writeHead(status, { 'Content-Type': 'application/json' });
  res.end(data);
}
const errJson = (res, status, code, message) =>
  json(res, status, { error: { code, message } });

function readBody(req) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    let size = 0;
    req.on('data', (c) => {
      size += c.length;
      if (size > 4 << 20) { reject(new Error('body too large')); req.destroy(); return; }
      chunks.push(c);
    });
    req.on('end', () => resolve(Buffer.concat(chunks)));
    req.on('error', reject);
  });
}

function download(url, dest) {
  return new Promise((resolve, reject) => {
    const mod = url.startsWith('https:') ? https : http;
    const file = fs.createWriteStream(dest);
    mod.get(url, (res) => {
      if (res.statusCode !== 200) {
        file.close(); fs.rmSync(dest, { force: true });
        reject(new Error(`download ${url}: status ${res.statusCode}`));
        return;
      }
      res.pipe(file);
      file.on('finish', () => file.close(resolve));
    }).on('error', (e) => { file.close(); fs.rmSync(dest, { force: true }); reject(e); });
  });
}

function postCallback(urlStr, payload) {
  return new Promise((resolve, reject) => {
    const body = Buffer.from(JSON.stringify(payload));
    const sig = crypto.createHmac('sha256', TOKEN).update(body).digest('hex');
    const url = new URL(urlStr);
    const mod = url.protocol === 'https:' ? https : http;
    const req = mod.request(url, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Content-Length': body.length,
        'X-Signature': sig,
      },
    }, (res) => { res.resume(); resolve(res.statusCode); });
    req.on('error', reject);
    req.end(body);
  });
}

function claudeAvailable() {
  try {
    execFileSync('claude', ['--version'], { timeout: 15000, stdio: 'pipe' });
    return true;
  } catch (e) {
    return false;
  }
}

function buildTaskMD(r) {
  const lines = [`# ${r.title}`, '', r.description || '', ''];
  if ((r.acceptanceCriteria || []).length) {
    lines.push('## Acceptance criteria', '');
    r.acceptanceCriteria.forEach((c) => lines.push(`- ${c}`));
    lines.push('');
  }
  if ((r.links || []).length) {
    lines.push('## Links', '');
    r.links.forEach((l) => lines.push(`- ${l}`));
    lines.push('');
  }
  if ((r.attachments || []).length) {
    lines.push('## Attachments (downloaded into ./attachments)', '');
    r.attachments.forEach((a) => lines.push(`- ${a.name}`));
    lines.push('');
  }
  if (r.context && r.context.instructions) {
    lines.push('## Extra instructions', '', r.context.instructions, '');
  }
  lines.push('Produce your changes in this directory. Write a SUMMARY.md describing what you did.');
  return lines.join('\n');
}

function listProducedFiles(dir, before) {
  const out = [];
  const walk = (d) => {
    for (const entry of fs.readdirSync(d, { withFileTypes: true })) {
      if (entry.name === '.git' || entry.name === 'attachments') continue;
      const full = path.join(d, entry.name);
      if (entry.isDirectory()) { walk(full); continue; }
      const rel = path.relative(dir, full);
      if (!before.has(rel) && fs.statSync(full).size <= 20 << 20) out.push(rel);
    }
  };
  walk(dir);
  return out.slice(0, 20);
}

async function runJob(jobId) {
  const job = jobs.get(jobId);
  if (!job || job.status !== 'queued') return;
  job.status = 'running';
  job.startedAt = new Date().toISOString();
  const r = job.request;
  const dir = path.join(WORK_DIR, jobId);
  let result;
  try {
    fs.mkdirSync(path.join(dir, 'attachments'), { recursive: true });

    // optional repo clone (§9.2)
    if (r.context && r.context.repositoryUrl) {
      const args = ['clone', '--depth', '1'];
      if (r.context.branch) args.push('-b', r.context.branch);
      args.push(r.context.repositoryUrl, path.join(dir, 'repo'));
      await new Promise((resolve) => {
        execFile('git', args, { timeout: 120000 }, (err, _o, stderr) => {
          if (err) log('git clone failed', { jobId, err: String(stderr || err) });
          resolve();
        });
      });
    }
    for (const a of r.attachments || []) {
      try {
        await download(a.url, path.join(dir, 'attachments', path.basename(a.name)));
      } catch (e) { log('attachment download failed', { jobId, name: a.name, err: e.message }); }
    }
    fs.writeFileSync(path.join(dir, 'TASK.md'), buildTaskMD(r));
    const before = new Set(['TASK.md']);

    if (claudeAvailable()) {
      result = await new Promise((resolve) => {
        const child = execFile('claude',
          ['-p', fs.readFileSync(path.join(dir, 'TASK.md'), 'utf8'),
            '--output-format', 'json', '--dangerously-skip-permissions'],
          { cwd: dir, timeout: 30 * 60 * 1000, maxBuffer: 32 << 20 },
          (err, stdout, stderr) => {
            if (job.cancel) { resolve({ status: 'failed', summary: 'job canceled', log: '' }); return; }
            if (err) {
              resolve({ status: 'failed', summary: `claude execution failed: ${err.message}`,
                log: String(stderr || '').slice(-4000) });
              return;
            }
            let summary = 'Claude Code completed the task.';
            try {
              const parsed = JSON.parse(stdout);
              summary = parsed.result || parsed.summary || summary;
            } catch (e) { summary = String(stdout).slice(0, 1000) || summary; }
            resolve({ status: 'succeeded', summary: String(summary).slice(0, 4000),
              log: String(stdout).slice(-4000) });
          });
        job.kill = () => child.kill('SIGTERM');
      });
    } else {
      // offline placeholder result keeps the end-to-end flow demonstrable
      const summary = `Simulated work performer result (Claude Code CLI not available in this container).\n` +
        `Reviewed task "${r.title}" with ${(r.acceptanceCriteria || []).length} acceptance criteria.`;
      fs.writeFileSync(path.join(dir, 'SUMMARY.md'),
        `# Simulated result\n\n${summary}\n\nThis placeholder proves the §5.2 contract end to end.`);
      result = { status: 'succeeded', summary, log: 'claude CLI unavailable; produced simulated SUMMARY.md' };
    }

    const artifacts = listProducedFiles(dir, before).map((rel) => ({
      name: rel.replace(/\//g, '_'),
      url: `${PUBLIC_BASE}/artifacts/${jobId}/${encodeURIComponent(rel)}`,
    }));
    result.artifacts = artifacts;
  } catch (e) {
    result = { status: 'failed', summary: `job crashed: ${e.message}`, log: '', artifacts: [] };
  }

  job.status = result.status;
  job.finishedAt = new Date().toISOString();
  const payload = {
    jobId, taskId: r.taskId, status: result.status,
    summary: result.summary, artifacts: result.artifacts || [], log: result.log || '',
  };
  for (let attempt = 1; attempt <= 3; attempt++) {
    try {
      const code = await postCallback(r.callbackUrl, payload);
      log('callback delivered', { jobId, code });
      break;
    } catch (e) {
      log('callback failed', { jobId, attempt, err: e.message });
      await new Promise((s) => setTimeout(s, attempt * 2000));
    }
  }
}

async function pump() {
  if (running) return;
  const next = queue.shift();
  if (!next) return;
  running = true;
  try { await runJob(next); } finally {
    running = false;
    setImmediate(pump);
  }
}

const server = http.createServer(async (req, res) => {
  const url = new URL(req.url, `http://${req.headers.host}`);

  if (req.method === 'GET' && url.pathname === '/healthz') {
    return json(res, 200, { status: 'ok' });
  }

  // artifact downloads are unauthenticated within the compose network
  const artMatch = url.pathname.match(/^\/artifacts\/([\w]+)\/(.+)$/);
  if (req.method === 'GET' && artMatch) {
    const file = path.join(WORK_DIR, artMatch[1], decodeURIComponent(artMatch[2]));
    if (!file.startsWith(path.join(WORK_DIR, artMatch[1]) + path.sep) || !fs.existsSync(file)) {
      return errJson(res, 404, 'not_found', 'artifact not found');
    }
    res.writeHead(200, { 'Content-Type': 'application/octet-stream' });
    return fs.createReadStream(file).pipe(res);
  }

  if (TOKEN && req.headers.authorization !== `Bearer ${TOKEN}`) {
    return errJson(res, 401, 'unauthorized', 'missing or invalid bearer token');
  }

  if (req.method === 'POST' && url.pathname === '/v1/jobs') {
    let body;
    try { body = JSON.parse(await readBody(req)); }
    catch (e) { return errJson(res, 400, 'bad_request', e.message); }
    if (!body.taskId || !body.title || !body.callbackUrl) {
      return errJson(res, 400, 'bad_request', 'taskId, title and callbackUrl are required');
    }
    const jobId = 'wp_' + crypto.randomBytes(8).toString('hex');
    jobs.set(jobId, { status: 'queued', taskId: body.taskId, request: body,
      startedAt: null, finishedAt: null, cancel: false });
    queue.push(jobId);
    setImmediate(pump);
    log('job queued', { jobId, taskId: body.taskId });
    return json(res, 202, { jobId, status: 'queued' });
  }

  const jobMatch = url.pathname.match(/^\/v1\/jobs\/(wp_[\w]+)$/);
  if (jobMatch) {
    const job = jobs.get(jobMatch[1]);
    if (!job) return errJson(res, 404, 'not_found', 'job not found');
    if (req.method === 'GET') {
      return json(res, 200, { jobId: jobMatch[1], status: job.status,
        startedAt: job.startedAt, finishedAt: job.finishedAt });
    }
    if (req.method === 'DELETE') { // best-effort cancel (§5.2)
      job.cancel = true;
      const idx = queue.indexOf(jobMatch[1]);
      if (idx >= 0) { queue.splice(idx, 1); job.status = 'failed'; job.finishedAt = new Date().toISOString(); }
      if (job.kill) try { job.kill(); } catch (e) { /* already gone */ }
      return json(res, 200, { jobId: jobMatch[1], status: 'cancel_requested' });
    }
  }

  errJson(res, 404, 'not_found', 'unknown endpoint');
});

server.listen(PORT, () => log('work-performer listening', { port: PORT, claude: claudeAvailable() }));
process.on('SIGTERM', () => server.close(() => process.exit(0)));