quorlasocial/dashboard/server/jobs.js

// Runs the project's Claude Code skills (account-audit, weekly-plan, weekly-report, …) headlessly
// through the Claude Agent SDK, one job at a time, with a strict tool allow-list.
// Publishing tools are always blocked: nothing goes live from the dashboard.
import fs from "node:fs";
import path from "node:path";
import crypto from "node:crypto";
import { query } from "@anthropic-ai/claude-agent-sdk";
import { PROJECT_ROOT, DATA_DIR, AI } from "./config.js";
import { getClient, listDocs, safeSlug } from "./clients.js";
import { toPdf } from "../../tools/report-pdf/build.mjs";
import { refreshSnapshot } from "./metrics.js";

const JOB_DIR = path.join(DATA_DIR, "jobs");
fs.mkdirSync(JOB_DIR, { recursive: true });

export const JOB_TYPES = {
  "account-audit": { label: "Account audit", needsClient: true, prompt: (c) => `/account-audit ${c.name}` },
  "weekly-plan": { label: "Weekly content plan", needsClient: true, prompt: (c) => `/weekly-plan ${c.name}` },
  "month-plan": { label: "Next month's content calendar", needsClient: true, prompt: (c) => {
    const d = new Date(); const next = new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth() + 1, 1));
    const month = next.toLocaleString("en-US", { month: "long", year: "numeric", timeZone: "UTC" });
    return `/weekly-plan ${c.name}

Plan all of ${month}, not just next week: every week that starts in ${month}, one calendar file per week (clients/${c.slug}/calendar/YYYY-Www.md). Check content_plan first so nothing is doubled, and add every planned post to the dashboard Calendar with plan_add (status idea or brief).`;
  } },
  "weekly-report": { label: "Weekly report", needsClient: true, prompt: (c) => `/weekly-report ${c.name}` },
  "competitor-scan": { label: "Find & compare competitors", needsClient: true, prompt: (c) => `/competitor-scan ${c.name}` },
  "assign-tasks": { label: "Assign this week's tasks", needsClient: false, prompt: () => `/assign-tasks` },
  custom: { label: "Custom request", needsClient: false, prompt: (c, text) => (c ? `Client: ${c.name} (clients/${c.slug}/).\n\n${text}` : text) },
};

const HEADLESS_RULES = `You are running headlessly from the agency dashboard. Nobody can answer questions mid-run.
- Don't stop to ask. Make sensible assumptions, finish the job, and list open questions for the team inside the files you write (and in your final message).
- Never publish, schedule or post anything, and never create or change records in Odoo from this run.
- Only write inside clients/, team/ and reports/. Export finished client-facing reports to PDF with the project's PDF builder.
- End with a short summary: what you produced (file paths) and anything the team must check.`;

const READ_TOOLS = [
  "Read", "Glob", "Grep", "WebSearch", "WebFetch", "Skill", "TodoWrite",
  "Edit(clients/**)", "Write(clients/**)", "Edit(team/**)", "Write(team/**)", "Edit(reports/**)", "Write(reports/**)",
  "Bash(node tools/report-pdf/build.mjs:*)",
  "mcp__social-manager__list_accounts", "mcp__social-manager__instagram_insights", "mcp__social-manager__instagram_recent_posts",
  "mcp__social-manager__instagram_comments", "mcp__social-manager__facebook_page_insights", "mcp__social-manager__facebook_recent_posts",
  "mcp__social-manager__audience_demographics", "mcp__social-manager__instagram_mentions", "mcp__social-manager__competitor_lookup", "mcp__social-manager__competitors", "mcp__social-manager__competitor_suggest",
  "mcp__social-manager__google_places", "mcp__social-manager__client_google_listing", "mcp__social-manager__google_trends", "mcp__social-manager__hashtag_top_posts",
  "mcp__social-manager__list_connections", "mcp__social-manager__suggested_post_times", "mcp__social-manager__content_library", "mcp__social-manager__brand_kit",
  "mcp__social-manager__caption_check", "mcp__social-manager__content_plan",
  // Calendar items are plans for the team (idea → brief → …), never posts: nothing is scheduled or published from a run.
  "mcp__social-manager__plan_add", "mcp__social-manager__plan_update",
  "mcp__odoo-mcp__search_records", "mcp__odoo-mcp__read_records", "mcp__odoo-mcp__read_record_by_id", "mcp__odoo-mcp__count_records",
];
// Write tools a run must never touch. Blocking also drops them from the request, so the model never sees them.
const BLOCKED_TOOLS = [
  "instagram_publish", "facebook_publish", "reply_comment", "hide_comment", "send_direct_message", "schedule_post",
  "approve_scheduled", "cancel_scheduled", "schedule_from_library", "competitor_add", "comment_inbox", "direct_messages", "list_scheduled",
].map((t) => `mcp__social-manager__${t}`);

// Path-scoped rules like Write(clients/**) can miss (absolute paths, other spellings), so file writes are also
// checked here by real path. Everything else not on the allow-list is refused, with the reason in the job log.
const WRITE_DIRS = ["clients", "team", "reports"].map((d) => path.resolve(PROJECT_ROOT, d));
const FILE_TOOLS = new Set(["Write", "Edit", "MultiEdit", "NotebookEdit"]);
function permissionCheck(job) {
  return async (tool, input) => {
    const file = input.file_path || input.notebook_path;
    if (FILE_TOOLS.has(tool) && file) {
      const full = path.resolve(PROJECT_ROOT, file);
      if (WRITE_DIRS.some((d) => full === d || full.startsWith(d + path.sep))) return { behavior: "allow", updatedInput: input };
      log(job, "warn", `Refused ${tool} outside clients/, team/ and reports/: ${file}`);
      return { behavior: "deny", message: `Only files inside clients/, team/ and reports/ can be written. Write to clients/<slug>/ instead.` };
    }
    log(job, "warn", `Refused tool not allowed in runs: ${tool}`);
    return { behavior: "deny", message: `${tool} isn't available in dashboard runs.` };
  };
}

// Our own model servers (Ollama on a laptop) break on very long requests, so those runs send only what the job needs:
// the built-in tools runs use, the job's own skill and the social-manager server (≈13k tokens instead of ≈25k).
const LEAN_TOOLS = ["Read", "Glob", "Grep", "Edit", "Write", "Bash", "Skill", "TodoWrite"];
function leanOptions(prompt) {
  if (!process.env.ANTHROPIC_BASE_URL) return { skills: "all" };
  const skill = prompt.match(/^\/([a-z0-9-]+)/)?.[1];
  return {
    skills: skill ? [skill] : "all",
    tools: LEAN_TOOLS,
    strictMcpConfig: true,
    mcpServers: { "social-manager": { command: "node", args: [path.join(PROJECT_ROOT, "connector", "src", "server.js")] } },
  };
}

const jobs = new Map();
const queue = [];
let running = null;

const jobFile = (id) => path.join(JOB_DIR, `${id}.json`);
const persist = (job) => fs.writeFileSync(jobFile(job.id), JSON.stringify({ ...job, abort: undefined }));

// Load history; anything that was mid-run when the server stopped is marked interrupted.
for (const f of fs.readdirSync(JOB_DIR).filter((x) => x.endsWith(".json"))) {
  try {
    const job = JSON.parse(fs.readFileSync(path.join(JOB_DIR, f), "utf8"));
    if (job.status === "running" || job.status === "queued") Object.assign(job, { status: "interrupted", finishedAt: job.finishedAt || new Date().toISOString() });
    jobs.set(job.id, job);
  } catch { /* ignore corrupt job files */ }
}

const publicJob = ({ abort, ...job }, { withLog = false } = {}) => (withLog ? job : { ...job, log: undefined, logLines: job.log?.length || 0 });

export function listJobs({ slug } = {}) {
  return [...jobs.values()]
    .filter((j) => !slug || j.slug === slug)
    .sort((a, b) => b.createdAt.localeCompare(a.createdAt))
    .slice(0, 100)
    .map((j) => publicJob(j));
}

export function getJob(id) {
  const job = jobs.get(id);
  return job ? publicJob(job, { withLog: true }) : null;
}

export function createJob({ type, slug, text }, user) {
  const def = JOB_TYPES[type];
  if (!def) throw Object.assign(new Error("Unknown job type."), { status: 400 });
  const client = slug ? (safeSlug(slug), getClient(slug)) : null;
  if (def.needsClient && !client) throw Object.assign(new Error("Pick a client."), { status: 400 });
  if (type === "custom" && !String(text || "").trim()) throw Object.assign(new Error("Describe what you need."), { status: 400 });
  const job = {
    id: crypto.randomUUID(),
    type,
    label: def.label,
    slug: client?.slug || null,
    clientName: client?.name || null,
    prompt: def.prompt(client, String(text || "").trim()),
    status: "queued",
    createdBy: user.email,
    createdAt: new Date().toISOString(),
    log: [],
  };
  jobs.set(job.id, job);
  persist(job);
  queue.push(job.id);
  pump();
  return publicJob(job);
}

export function cancelJob(id) {
  const job = jobs.get(id);
  if (!job) return null;
  if (job.status === "queued") {
    queue.splice(queue.indexOf(id), 1);
    Object.assign(job, { status: "cancelled", finishedAt: new Date().toISOString() });
    persist(job);
  } else if (job.status === "running") job.abort?.abort();
  return publicJob(job);
}

function log(job, kind, text) {
  job.log.push({ t: new Date().toISOString(), kind, text: String(text).slice(0, 4000) });
  if (job.log.length % 5 === 0) persist(job);
}

function describeTool(block) {
  const i = block.input || {};
  const target = i.file_path || i.path || i.pattern || i.command || i.skill || i.query || i.url || i.account || "";
  return `${block.name.replace(/^mcp__([^_]+(?:-[^_]+)*)__/, "$1 · ")}${target ? ` — ${String(target).replace(PROJECT_ROOT, "").replace(/^[\\/]+/, "").replace(/\\/g, "/").slice(0, 160)}` : ""}`;
}

function docStamp(slug) {
  if (!slug) return new Map();
  return new Map(listDocs(slug).map((d) => [d.path, d.updatedAt]));
}

/** Runs on our own model server (Ollama etc.): every model slot points at the chosen model, and the Anthropic key stays home. */
function runEnv() {
  if (!process.env.ANTHROPIC_BASE_URL) return undefined;
  // Summarise at ~23k tokens (35% of the 64k window set on the Ollama model): a laptop reads ~100–150 tokens/s and the
  // agent gives up on a request after 5 minutes, while the spare window leaves room for big tool results.
  const env = {
    ...process.env, ANTHROPIC_AUTH_TOKEN: process.env.ANTHROPIC_AUTH_TOKEN || "ollama", CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: "1",
    CLAUDE_CODE_AUTO_COMPACT_WINDOW: process.env.CLAUDE_CODE_AUTO_COMPACT_WINDOW || "65536",
    CLAUDE_AUTOCOMPACT_PCT_OVERRIDE: process.env.CLAUDE_AUTOCOMPACT_PCT_OVERRIDE || "35",
    CLAUDE_CODE_MAX_OUTPUT_TOKENS: process.env.CLAUDE_CODE_MAX_OUTPUT_TOKENS || "8000",
  };
  delete env.ANTHROPIC_API_KEY;
  for (const k of ["ANTHROPIC_DEFAULT_OPUS_MODEL", "ANTHROPIC_DEFAULT_SONNET_MODEL", "ANTHROPIC_DEFAULT_HAIKU_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL"]) env[k] = AI.model;
  return env;
}

async function run(job) {
  Object.assign(job, { status: "running", startedAt: new Date().toISOString() });
  job.abort = new AbortController();
  persist(job);
  const before = docStamp(job.slug);
  log(job, "info", `Started: ${job.prompt.split("\n")[0]}`);
  // Report charts are drawn from the stored numbers, so bring them up to date first.
  if (job.slug && /^\/(account-audit|weekly-report|competitor-scan)/.test(job.prompt)) {
    try { await refreshSnapshot(job.slug); log(job, "info", "Refreshed the account numbers for the charts."); }
    catch (e) { log(job, "warn", `Couldn't refresh account numbers (${e.message}); charts use the last stored ones.`); }
  }
  try {
    for await (const msg of query({
      prompt: job.prompt,
      options: {
        cwd: PROJECT_ROOT,
        settingSources: ["project"],
        ...leanOptions(job.prompt),
        model: AI.model,
        maxTurns: AI.maxTurns,
        env: runEnv(),
        permissionMode: "default",
        allowedTools: READ_TOOLS,
        canUseTool: permissionCheck(job),
        disallowedTools: BLOCKED_TOOLS,
        abortController: job.abort,
        systemPrompt: { type: "preset", preset: "claude_code", append: HEADLESS_RULES },
        stderr: (data) => { if (/error/i.test(data)) log(job, "stderr", data.trim()); },
      },
    })) {
      if (msg.type === "assistant") {
        for (const block of msg.message?.content || []) {
          if (block.type === "text" && block.text.trim()) log(job, "text", block.text.trim());
          else if (block.type === "tool_use") log(job, "tool", describeTool(block));
        }
      } else if (msg.type === "result") {
        job.costUsd = msg.total_cost_usd ?? null;
        job.turns = msg.num_turns ?? null;
        for (const d of msg.permission_denials || []) log(job, "warn", `Denied: ${d.tool_name} ${JSON.stringify(d.tool_input).slice(0, 200)}`);
        if (msg.subtype === "success" && !msg.is_error) {
          job.result = msg.result;
          job.status = "succeeded";
        } else {
          job.status = "failed";
          job.error = msg.result || msg.subtype;
        }
      }
    }
    if (job.status === "running") job.status = "succeeded";
  } catch (e) {
    job.status = job.abort.signal.aborted ? "cancelled" : "failed";
    job.error = e.message;
    log(job, "error", e.message);
  }

  // Which documents changed? Make sure each changed client/internal report has a fresh PDF.
  if (job.slug) {
    const changed = listDocs(job.slug).filter((d) => d.kind !== "setup" && before.get(d.path) !== d.updatedAt);
    job.files = changed.map((d) => d.path);
    for (const d of changed.filter((x) => (x.kind === "audit" || x.kind === "report" || x.kind === "competitors") && (!x.pdf || x.pdf.stale))) {
      try {
        toPdf(path.join(safeSlug(job.slug), d.path));
        log(job, "info", `PDF exported: ${d.path.replace(/\.md$/, ".pdf")}`);
      } catch (e) {
        log(job, "error", `PDF export failed for ${d.path}: ${e.message}`);
      }
    }
  }
  job.finishedAt = new Date().toISOString();
  log(job, "info", `Finished: ${job.status}${job.costUsd != null ? ` · $${job.costUsd.toFixed(2)}` : ""}`);
  delete job.abort;
  persist(job);
}

async function pump() {
  if (running || !queue.length) return;
  const job = jobs.get(queue.shift());
  if (!job || job.status !== "queued") return pump();
  running = job;
  try { await run(job); } finally { running = null; pump(); }
}