jobbi-bewerbung/scripts/jobsuche-runner.js

#!/usr/bin/env node
// Host-side runner for the per-user job search.
//
// Why it lives on the host and not in the app: the app container has neither
// `claude` nor `ollama` nor a docker socket. The app therefore only *queues*
// runs (table suchlaeufe); this script — started by cron every few minutes —
// picks them up and executes them. It replaces the two old single-user scripts
// (jobsuche-cron.sh and jobsuche-remote-cron.sh), whose search criteria were
// hard-coded into the prompt.
//
// Isolation: every run executes in its OWN throwaway Docker container
// (AGENT_IMAGE), with the requesting user's dedicated home directory mounted
// as /home/agent. That home IS that user's ~/.claude (memories, session
// contexts, history, settings) — never shared with another user. The three
// search skills are shared code baked into the image and symlinked into
// ~/.claude/skills by the image entrypoint.
//
// Per run it:
//   1. enqueues scheduled runs that are due (per user's Zeitplan),
//   2. claims queued runs (status angefordert -> laeuft) and runs them in
//      parallel up to MAX_PARALLEL — different users have different Ollama
//      keys and thus separate rate limits, so parallelism is safe across users,
//   3. builds the prompt from that user's Suchprofil + their Lebenslauf,
//   4. starts a container that runs the headless agent with THAT user's
//      credentials:
//        - BEWERBUNG_API_KEY = the user's API token (created if missing), so
//          every imported Jobangebot lands in their account and passes the
//          same dedup/blacklist path as the rest of the app,
//        - ANTHROPIC_* = the user's own Ollama endpoint+key, so the search is
//          billed to whoever requested it,
//   5. parses the agent's final "ERGEBNIS neu=… dubletten=… verworfen=…" line
//      and writes status + counts back, so the UI can show what happened.

const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const { spawn, spawnSync } = require('child_process');
const sqlite3 = require('sqlite3');

const suchprofil = require('../lib/suchprofil');

const DB_PATH = process.env.JOBSUCHE_DB || '/opt/jobbi-bewerbung/data/bewerbungen.db';
const LOG_DIR = process.env.JOBSUCHE_LOG_DIR || '/opt/jobbi-bewerbung/logs';
const API_URL = process.env.JOBSUCHE_API_URL || 'http://localhost:4327/api/v1';
const RUN_TIMEOUT_MS = (Number(process.env.JOBSUCHE_TIMEOUT) || 2700) * 1000;
// A run that has been 'laeuft' for longer than this lost its process (reboot,
// OOM); it would otherwise block the user's queue forever.
const STALE_MS = RUN_TIMEOUT_MS + 10 * 60 * 1000;
const LOG_RETENTION_DAYS = Number(process.env.JOBSUCHE_LOG_RETENTION_DAYS) || 30;

// Agent-Container. Das Image wird lokal gebaut (npm run docker:build-agent),
// nicht über die Registry verteilt — es läuft nur auf diesem Host.
const AGENT_IMAGE = process.env.JOBSUCHE_AGENT_IMAGE || 'jobbi-bewerbung-agent:latest';
// uid/gid, als die der Container läuft und der das pro-Benutzer-Home gehört.
const AGENT_UID = Number(process.env.JOBSUCHE_AGENT_UID) || 1000;
const AGENT_GID = Number(process.env.JOBSUCHE_AGENT_GID) || AGENT_UID;
// Pro-Benutzer-Verzeichnisse (je HOME = je ~/.claude).
const AGENTS_DIR = process.env.JOBSUCHE_AGENTS_DIR || '/opt/jobbi-bewerbung/agents';
// Gleichzeitig laufende Such-Container. Verschiedene Benutzer haben getrennte
// Ollama-Keys -> getrennte Rate-Limits, deshalb ist pro-Benutzer-Parallelität
// sicher. Der Wert begrenzt lediglich die Host-Last.
const MAX_PARALLEL = Number(process.env.JOBSUCHE_MAX_PARALLEL) || 4;

const CFG = 'cfg:';

const db = new sqlite3.Database(DB_PATH);
const all = (sql, p = []) => new Promise((res, rej) => db.all(sql, p, (e, r) => (e ? rej(e) : res(r))));
const get = (sql, p = []) => new Promise((res, rej) => db.get(sql, p, (e, r) => (e ? rej(e) : res(r))));
const run = (sql, p = []) => new Promise((res, rej) => db.run(sql, p, function (e) { e ? rej(e) : res(this); }));

function log(msg) {
  console.log(`${new Date().toISOString()} ${msg}`);
}

async function cfgWert(userId, key) {
  const row = await get('SELECT value FROM app_state WHERE user_id = ? AND key = ?', [userId, CFG + key]);
  return row && row.value ? row.value : '';
}

// The user's API token, created on first use. Without it the agent could not
// import anything; making the user generate one by hand first would just be a
// trap ("profile active, nothing happens").
async function ensureApiKey(userId) {
  const vorhanden = await cfgWert(userId, 'API_TOKEN');
  if (vorhanden) return vorhanden;
  const token = crypto.randomBytes(32).toString('hex');
  await run(
    'INSERT INTO app_state (user_id, key, value) VALUES (?, ?, ?) ' +
    'ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value',
    [userId, CFG + 'API_TOKEN', token]
  );
  log(`  API-Key für Benutzer ${userId} automatisch erzeugt.`);
  return token;
}

// The CV text the search profile is derived from. Prefer the Lebenslauf, fall
// back to a Profil/Kurzprofil — the point is to give the model the user's actual
// roles and technologies rather than a hard-coded list.
async function profilText(userId) {
  const rows = await all(
    `SELECT typ, inhalt FROM basis_dokumente
      WHERE user_id = ? AND typ IN ('Lebenslauf', 'Profil/Kurzprofil')
        AND inhalt IS NOT NULL AND inhalt != ''
      ORDER BY CASE typ WHEN 'Lebenslauf' THEN 0 ELSE 1 END, id ASC`,
    [userId]
  );
  return rows.map((r) => (r.inhalt || '').trim()).filter(Boolean).join('\n\n---\n\n');
}

// Legt das pro-Benutzer-Home an (falls noch nicht vorhanden) und übergibt es
// an den Container-Agent. Bleibt über Läufe hinweg bestehen, damit Memories
// und Contexte erhalten bleiben.
function ensureAgentDir(userId) {
  const dir = path.join(AGENTS_DIR, String(userId));
  fs.mkdirSync(dir, { recursive: true });
  spawnSync('chown', ['-R', `${AGENT_UID}:${AGENT_GID}`, dir]);
  return dir;
}

// 1. Queue scheduled runs that are due.
async function faelligeEinreihen() {
  const jetzt = new Date();
  const users = await all('SELECT user_id FROM suchprofil WHERE aktiv = 1');
  for (const { user_id: userId } of users) {
    const profil = suchprofil.fromRow(await get('SELECT * FROM suchprofil WHERE user_id = ?', [userId]));
    if (suchprofil.validate(profil).length) continue; // unbrauchbares Profil -> kein Zeitplan-Lauf

    // Only scheduled runs count here: a manual test run must not swallow the
    // day's automatic run.
    const letzter = await get(
      "SELECT angefordert_at FROM suchlaeufe WHERE user_id = ? AND ausloeser = 'zeitplan' ORDER BY id DESC LIMIT 1",
      [userId]
    );
    const letzterAt = letzter ? `${letzter.angefordert_at}Z`.replace(' ', 'T') : null;
    if (!suchprofil.istFaellig(profil, jetzt, letzterAt)) continue;

    const offen = await get(
      "SELECT id FROM suchlaeufe WHERE user_id = ? AND status IN ('angefordert','laeuft') LIMIT 1",
      [userId]
    );
    if (offen) continue; // läuft schon etwas für diesen Benutzer

    await run("INSERT INTO suchlaeufe (user_id, status, ausloeser) VALUES (?, 'angefordert', 'zeitplan')", [userId]);
    log(`Zeitplan: Lauf für Benutzer ${userId} eingereiht.`);
  }
}

// Runs whose process died: release them so the user is not stuck.
async function verwaisteAufraeumen() {
  const res = await run(
    `UPDATE suchlaeufe SET status = 'fehler', beendet_at = CURRENT_TIMESTAMP,
       fehler = 'Lauf abgebrochen (Container nicht mehr vorhanden).'
     WHERE status = 'laeuft'
       AND gestartet_at IS NOT NULL
       AND (julianday('now') - julianday(gestartet_at)) * 86400000 > ?`,
    [STALE_MS]
  );
  if (res.changes) log(`${res.changes} verwaiste(n) Lauf/Läufe als Fehler markiert.`);
}

// Claim the oldest queued run. The conditional UPDATE is the claim: a second
// runner starting concurrently gets changes = 0 and moves on.
async function naechstenLaufClaimen() {
  const kandidat = await get("SELECT * FROM suchlaeufe WHERE status = 'angefordert' ORDER BY id ASC LIMIT 1");
  if (!kandidat) return null;
  const res = await run(
    "UPDATE suchlaeufe SET status = 'laeuft', gestartet_at = CURRENT_TIMESTAMP WHERE id = ? AND status = 'angefordert'",
    [kandidat.id]
  );
  if (!res.changes) return null;
  return get('SELECT * FROM suchlaeufe WHERE id = ?', [kandidat.id]);
}

async function laufAbschliessen(id, status, felder = {}) {
  await run(
    `UPDATE suchlaeufe SET status = ?, beendet_at = CURRENT_TIMESTAMP,
       neu = ?, dubletten = ?, verworfen = ?, fehler = ?, log_datei = ?
     WHERE id = ?`,
    [status, felder.neu ?? null, felder.dubletten ?? null, felder.verworfen ?? null,
     felder.fehler || null, felder.log || null, id]
  );
}

// Start the headless agent inside a throwaway container for this user. Only the
// curated env below is passed in (not process.env) so no host secrets leak.
function agentStarten(prompt, envObj, logStream, containerName, agentDir) {
  return new Promise((resolve) => {
    const args = [
      'run', '--rm',
      '--name', containerName,
      // --network host: der Agent erreicht den Tracker (localhost:4327), das
      // Internet (WebSearch/WebFetch/Arbeitsagentur) und den Ollama-Endpoint
      // des Benutzers — alles über die Host-Netzwerk-Sicht.
      '--network', 'host',
      '--user', `${AGENT_UID}:${AGENT_GID}`,
      '-e', 'HOME=/home/agent',
      '-e', 'IS_SANDBOX=1',
    ];
    for (const [k, v] of Object.entries(envObj)) args.push('-e', `${k}=${v}`);
    args.push(
      '-v', `${agentDir}:/home/agent`,
      '-w', '/work',
      AGENT_IMAGE,
      'claude', '--dangerously-skip-permissions', '-p', prompt,
    );

    const kind = spawn('docker', args, { stdio: ['ignore', 'pipe', 'pipe'] });

    let ausgabe = '';
    const sammeln = (buf) => {
      const s = buf.toString();
      ausgabe += s;
      logStream.write(s);
    };
    kind.stdout.on('data', sammeln);
    kind.stderr.on('data', sammeln);

    // Beim Timeout den `docker run`-Client killen UND den Container per Name
    // stoppen — ein kill des Clients allein ließe den Container sonst weiterlaufen.
    const timer = setTimeout(() => {
      spawnSync('docker', ['kill', containerName], { stdio: 'ignore' });
      kind.kill('SIGTERM');
      setTimeout(() => kind.kill('SIGKILL'), 10000);
    }, RUN_TIMEOUT_MS);

    kind.on('error', (err) => {
      clearTimeout(timer);
      resolve({ code: -1, ausgabe, fehler: `Agent-Container nicht startbar: ${err.message}` });
    });
    kind.on('close', (code, signal) => {
      clearTimeout(timer);
      const abgebrochen = signal === 'SIGTERM' || signal === 'SIGKILL';
      resolve({
        code,
        ausgabe,
        fehler: abgebrochen ? `Zeitlimit von ${RUN_TIMEOUT_MS / 1000}s überschritten.` : null,
      });
    });
  });
}

async function laufAusfuehren(lauf) {
  const userId = lauf.user_id;
  const user = await get('SELECT username FROM users WHERE id = ?', [userId]);
  const name = user ? user.username : `#${userId}`;
  const containerName = `jobbi-agent-${userId}-${lauf.id}`;
  log(`Lauf ${lauf.id} (Benutzer ${name}, ${lauf.ausloeser}) startet im Container ${containerName}.`);

  const profil = suchprofil.fromRow(await get('SELECT * FROM suchprofil WHERE user_id = ?', [userId]));
  const fehler = suchprofil.validate(profil);
  if (fehler.length) {
    await laufAbschliessen(lauf.id, 'fehler', { fehler: fehler.join(' ') });
    log(`  abgebrochen: ${fehler.join(' ')}`);
    return;
  }

  const ollamaKey = await cfgWert(userId, 'OLLAMA_API_KEY');
  if (!ollamaKey) {
    const msg = 'Kein eigener Ollama-API-Schlüssel hinterlegt — der Suchlauf läuft über das KI-Kontingent des Benutzers. Bitte in den Einstellungen eintragen.';
    await laufAbschliessen(lauf.id, 'fehler', { fehler: msg });
    log(`  abgebrochen: kein Ollama-Key für ${name}.`);
    return;
  }

  const apiKey = await ensureApiKey(userId);
  const modell = (await cfgWert(userId, 'OLLAMA_MODEL')) || 'glm-5.2:cloud';
  const prompt = suchprofil.buildPrompt(profil, await profilText(userId));

  fs.mkdirSync(LOG_DIR, { recursive: true });
  const stamp = new Date().toISOString().replace(/[:.]/g, '-');
  const logDatei = path.join(LOG_DIR, `jobsuche_${name}_${stamp}.log`);
  const logStream = fs.createWriteStream(logDatei, { flags: 'a' });
  logStream.write(`=== Suchlauf ${lauf.id} — Benutzer ${name} (${lauf.ausloeser}) ===\n`);
  logStream.write(`Modus: ${profil.modus} | Städte: ${profil.staedte.join(', ') || '–'}\n`);
  logStream.write(`Modell: ${modell} | Container: ${containerName}\n${'-'.repeat(50)}\n`);

  // Nur kuratierte Env-Vars in den Container — keine Host-Secrets.
  const envObj = {
    ANTHROPIC_BASE_URL: (await cfgWert(userId, 'OLLAMA_HOST')) || 'https://ollama.com',
    ANTHROPIC_AUTH_TOKEN: ollamaKey,
    ANTHROPIC_MODEL: modell,
    ANTHROPIC_SMALL_FAST_MODEL: modell,
    BEWERBUNG_API_URL: API_URL,
    BEWERBUNG_API_KEY: apiKey,
    JOBSUCHE_MODEL: modell,
  };

  const agentDir = ensureAgentDir(userId);
  const { code, ausgabe, fehler: laufFehler } = await agentStarten(prompt, envObj, logStream, containerName, agentDir);
  const ergebnis = suchprofil.parseErgebnis(ausgabe);

  logStream.write(`\n${'-'.repeat(50)}\nExit-Code: ${code}\n`);
  logStream.end();

  if (laufFehler) {
    await laufAbschliessen(lauf.id, 'fehler', { fehler: laufFehler, log: logDatei });
    log(`  Lauf ${lauf.id} fehlgeschlagen: ${laufFehler}`);
    return;
  }
  if (code !== 0 && !ergebnis) {
    await laufAbschliessen(lauf.id, 'fehler', {
      fehler: `Suchlauf mit Exit-Code ${code} beendet (Details im Log).`,
      log: logDatei,
    });
    log(`  Lauf ${lauf.id} fehlgeschlagen (Exit ${code}).`);
    return;
  }
  if (!ergebnis) {
    // The agent finished but did not report its tally — treat as done with unknown
    // numbers rather than as a failure; the import itself already happened.
    await laufAbschliessen(lauf.id, 'fertig', { neu: 0, dubletten: 0, verworfen: 0, log: logDatei });
    log(`  Lauf ${lauf.id} beendet, aber ohne ERGEBNIS-Zeile.`);
    return;
  }

  await laufAbschliessen(lauf.id, 'fertig', { ...ergebnis, log: logDatei });
  log(`  Lauf ${lauf.id} fertig: ${ergebnis.neu} neu, ${ergebnis.dubletten} Dubletten, ${ergebnis.verworfen} verworfen.`);
}

// Führt eine Liste von Läufen mit höchstens `parallel` gleichzeitig aus. Jeder
// Lauf läuft in seinem eigenen Container; verschiedene Benutzer haben getrennte
// Ollama-Keys, deshalb ist pro-Benutzer-Parallelität sicher.
async function runPool(items, parallel, fn) {
  const queue = items.slice();
  const workers = Array.from({ length: Math.min(parallel, queue.length || 1) }, async () => {
    while (queue.length) {
      const item = queue.shift();
      try {
        await fn(item);
      } catch (e) {
        log(`  Lauf ${item.id} unerwartet abgebrochen: ${e && e.message ? e.message : e}`);
        try {
          await laufAbschliessen(item.id, 'fehler', { fehler: `Runner-Ausnahme: ${e && e.message ? e.message : e}` });
        } catch (_) { /* Status ist dann zwar offen, aber ein Folge-Tick räumt auf. */ }
      }
    }
  });
  await Promise.all(workers);
}

function alteLogsAufraeumen() {
  try {
    const grenze = Date.now() - LOG_RETENTION_DAYS * 86400000;
    for (const f of fs.readdirSync(LOG_DIR)) {
      if (!f.startsWith('jobsuche_') || !f.endsWith('.log')) continue;
      const p = path.join(LOG_DIR, f);
      if (fs.statSync(p).mtimeMs < grenze) fs.unlinkSync(p);
    }
  } catch (e) { /* Aufräumen ist best effort */ }
}

async function main() {
  if (!fs.existsSync(DB_PATH)) {
    console.error(`Datenbank nicht gefunden: ${DB_PATH}`);
    process.exit(1);
  }
  await run('PRAGMA busy_timeout = 10000'); // die App schreibt parallel

  await verwaisteAufraeumen();
  await faelligeEinreihen();

  // Alle wartenden Läufe einsammeln (pro Benutzer steht ohnehin nur einer an)
  // und mit begrenzter Parallelität abarbeiten.
  const laeufe = [];
  let lauf;
  while ((lauf = await naechstenLaufClaimen())) laeufe.push(lauf);
  if (!laeufe.length) {
    log('Nichts zu tun.');
  } else {
    log(`${laeufe.length} Lauf/Läufe eingereiht, starte bis zu ${MAX_PARALLEL} parallel.`);
    await runPool(laeufe, MAX_PARALLEL, laufAusfuehren);
  }

  alteLogsAufraeumen();
  db.close();
}

main().catch((e) => {
  console.error('Runner-Fehler:', e);
  process.exit(1);
});