AgentHub/server/lib/opencode-usage.js

/**
 * @file OpenCode (`opencode`) sessions, subagents and token usage, read from
 * OpenCode's own SQLite store (`~/.local/share/opencode/opencode.db`, under
 * `XDG_DATA_HOME` when set). This is how an OpenCode session reaches
 * `sessions` (as `tool = 'opencode'`), its tokens `token_usage` and its
 * subagents `agents` — OpenCode has no hooks AgentHub installs.
 *
 * What the store holds (opencode 1.19, verified against a file it wrote — not
 * a documented format):
 * - `session(id, parent_id, directory, title, time_created, time_updated,
 *   time_archived)` — a session the `task` tool spawned has `parent_id` set,
 *   and is shown as a subagent of its root session;
 * - `message(session_id, data)` — an assistant message's JSON carries
 *   `modelID` and `tokens: {input, output, reasoning, cache: {read, write}}`
 *   (input NOT including cache reads; reasoning is counted as output here);
 * - `part(message_id, session_id, data)` — the message's text, reasoning and
 *   tool parts, read only for a transcript.
 * The file is re-read when its mtime (or its non-empty `-wal`'s) changes; per
 * root session the newest `time_updated` seen lives in `opencode_sessions`,
 * so a session the user deleted stays deleted until OpenCode writes to it
 * again. Status: active while its tree was updated in the last 2 minutes or a
 * live dashboard pane holds it, else completed. S3: a member's store (and its
 * WAL) is read through the guard into a private copy, never opened in place.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

const fs = require("node:fs");
const os = require("node:os");
const path = require("node:path");
const { db, stmts, Database } = require("../db");
const { claimToolSessions } = require("./dashboard-runs");
const pricingSync = require("./pricing-sync");

const ACTIVE_MS = 120_000; // the window codex-usage.js calls a session active
const FUTURE_SLACK_MS = 1000;
const SESSION_ID = /^[A-Za-z0-9_-]{8,120}$/;
const MAX_COPY = 256 << 20;

/** OpenCode's data dir: `AGENTHUB_OPENCODE_DIR`, `$XDG_DATA_HOME/opencode`, else `~/.local/share/opencode`. */
function getOpencodeDir() {
  if (process.env.AGENTHUB_OPENCODE_DIR) return process.env.AGENTHUB_OPENCODE_DIR;
  const xdg = process.env.XDG_DATA_HOME;
  return path.join(xdg && path.isAbsolute(xdg) ? xdg : path.join(os.homedir(), ".local", "share"), "opencode");
}

/** Every store: `{path, mtime}` — this account's, then every member's. */
function storeFiles() {
  const members = require("./run-as")
    .memberHomes()
    .map((h) => path.join(h, ".local", "share", "opencode"));
  const out = [];
  for (const dir of [getOpencodeDir(), ...members]) {
    const file = path.join(dir, "opencode.db");
    try {
      const st = fs.lstatSync(file);
      if (!st.isFile()) continue;
      let mtime = st.mtimeMs;
      try {
        const wal = fs.lstatSync(`${file}-wal`);
        if (wal.isFile() && wal.size > 0) mtime = Math.max(mtime, wal.mtimeMs);
      } catch {
        /* no WAL right now */
      }
      out.push({ path: file, mtime: Math.floor(mtime) });
    } catch {
      /* no store */
    }
  }
  return out;
}

/**
 * Run `fn(conn)` on a store, read-only. A member's store (S3) and its WAL are
 * copied through the guard first: SQLite opens by NAME, and a member's FIFO
 * standing in for the `-wal` would block this thread.
 */
function withStore(file, fn) {
  const { ownerForRead, readFileSafe } = require("./run-as");
  let tmp = null;
  try {
    let open = file;
    if (require("./linux-accounts").s3Enabled() && ownerForRead(file)) {
      tmp = fs.mkdtempSync(path.join(os.tmpdir(), "agenthub-opencode-"));
      open = path.join(tmp, "opencode.db");
      fs.writeFileSync(open, readFileSafe(file, null, { max: MAX_COPY }));
      try {
        fs.writeFileSync(`${open}-wal`, readFileSafe(`${file}-wal`, null, { max: MAX_COPY }));
      } catch {
        /* no WAL (or not theirs): what is in the main file */
      }
    }
    const conn = new Database(open, { readonly: !tmp, fileMustExist: true });
    try {
      return fn(conn);
    } finally {
      conn.close();
    }
  } finally {
    if (tmp) fs.rmSync(tmp, { recursive: true, force: true });
  }
}

/** A session's model column (`{"id": …}` JSON, or a plain id), or null. */
function modelOf(raw) {
  if (typeof raw !== "string" || !raw) return null;
  try {
    const o = JSON.parse(raw);
    return typeof o?.id === "string" ? o.id.slice(0, 128) : null;
  } catch {
    return raw.slice(0, 128);
  }
}

/**
 * Everything a pass needs from one store: `[{root, children, usage, updated}]`
 * per root session — `usage` per model over the root and every descendant.
 */
function readStore(file) {
  return withStore(file, (conn) => {
    const rows = conn
      .prepare(
        `SELECT id, parent_id, directory, title, time_created, time_updated, time_archived, model
         FROM session`
      )
      .all();
    const byId = new Map(rows.map((r) => [r.id, r]));
    const rootOf = (r) => {
      let cur = r;
      for (let i = 0; i < 32 && cur.parent_id && byId.has(cur.parent_id); i++) {
        cur = byId.get(cur.parent_id);
      }
      return cur;
    };
    const trees = new Map();
    for (const r of rows) {
      if (!SESSION_ID.test(r.id)) continue;
      const root = rootOf(r);
      if (!SESSION_ID.test(root.id)) continue;
      const t = trees.get(root.id) || { root, children: [], usage: {}, updated: 0, model: null };
      if (r !== root) t.children.push(r);
      t.updated = Math.max(t.updated, Number(r.time_updated) || 0);
      trees.set(root.id, t);
    }
    const sums = conn
      .prepare(
        `SELECT session_id,
           json_extract(data, '$.modelID') AS model,
           SUM(COALESCE(json_extract(data, '$.tokens.input'), 0)) AS input,
           SUM(COALESCE(json_extract(data, '$.tokens.output'), 0)
             + COALESCE(json_extract(data, '$.tokens.reasoning'), 0)) AS output,
           SUM(COALESCE(json_extract(data, '$.tokens.cache.read'), 0)) AS cache_read,
           SUM(COALESCE(json_extract(data, '$.tokens.cache.write'), 0)) AS cache_write,
           MAX(time_created) AS last
         FROM message
         WHERE json_extract(data, '$.role') = 'assistant'
         GROUP BY session_id, model`
      )
      .all();
    for (const s of sums) {
      const r = byId.get(s.session_id);
      const t = r && trees.get(rootOf(r).id);
      if (!t) continue;
      const model = typeof s.model === "string" && s.model ? s.model.slice(0, 128) : "unknown";
      const u = (t.usage[model] ||= { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 });
      u.input += Number(s.input) || 0;
      u.output += Number(s.output) || 0;
      u.cacheRead += Number(s.cache_read) || 0;
      u.cacheWrite += Number(s.cache_write) || 0;
      if (r === t.root && (!t.lastAt || s.last > t.lastAt)) {
        t.lastAt = s.last;
        t.model = model;
      }
    }
    return [...trees.values()];
  });
}

const getStore = db.prepare("SELECT * FROM opencode_stores WHERE path = ?");
const upsertStore = db.prepare(
  `INSERT INTO opencode_stores (path, mtime_ms, updated_at)
   VALUES (?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
   ON CONFLICT(path) DO UPDATE SET mtime_ms = excluded.mtime_ms, updated_at = excluded.updated_at`
);
const getTracked = db.prepare("SELECT * FROM opencode_sessions WHERE session_id = ?");
const upsertTracked = db.prepare(
  `INSERT INTO opencode_sessions (session_id, store, time_updated, updated_at)
   VALUES (?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
   ON CONFLICT(session_id) DO UPDATE SET store = excluded.store,
     time_updated = excluded.time_updated, updated_at = excluded.updated_at`
);
const activeOpencode = db.prepare(
  "SELECT s.id, t.time_updated FROM sessions s JOIN opencode_sessions t ON t.session_id = s.id WHERE s.tool = 'opencode' AND s.status = 'active'"
);
const heldByPane = db.prepare(
  "SELECT 1 FROM dashboard_runs WHERE session_id = ? AND status IN ('running', 'spawning') LIMIT 1"
);
const insertSession = db.prepare(
  `INSERT INTO sessions (id, name, status, cwd, model, started_at, ended_at, updated_at, metadata, tool)
   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'opencode')`
);
const updateSession = db.prepare(
  `UPDATE sessions SET name = ?, status = ?, model = COALESCE(?, model), ended_at = ?,
     updated_at = ?, cwd = COALESCE(cwd, ?), started_at = COALESCE(?, started_at)
   WHERE id = ? AND tool = 'opencode'`
);
const settle = db.prepare(
  "UPDATE sessions SET status = 'completed', ended_at = ? WHERE id = ? AND tool = 'opencode'"
);
const clearUsage = db.prepare("DELETE FROM token_usage WHERE session_id = ?");
const agentTimes = db.prepare(
  "UPDATE agents SET started_at = COALESCE(?, started_at), ended_at = ? WHERE id = ?"
);

const fresh = (t, now) => t - now < FUTURE_SLACK_MS && now - t < ACTIVE_MS;
const iso = (t) => (Number.isFinite(Number(t)) && Number(t) > 0 ? new Date(Number(t)).toISOString() : null);

/** Write one root session's row, usage and subagents. @returns its broadcast type, or null. */
function writeSession(store, t, now, broadcast) {
  const r = t.root;
  const existing = stmts.getSession.get(r.id);
  if (existing && existing.tool !== "opencode") return null;
  const cwd = typeof r.directory === "string" && path.isAbsolute(r.directory) ? r.directory : null;
  const runAs = require("./run-as");
  // S3: a member's store names only their own folder and session.
  if (!runAs.sameOwner(store, cwd, existing?.cwd)) return null;
  const lastWrite = iso(t.updated) || new Date(now).toISOString();
  const active = !r.time_archived && (fresh(t.updated, now) || !!heldByPane.get(r.id));
  const status = active ? "active" : "completed";
  const endedAt = active ? null : lastWrite;
  const name =
    (typeof r.title === "string" && r.title.trim().slice(0, 200)) ||
    existing?.name ||
    `OpenCode ${r.id.slice(4, 12)}`;
  const model = t.model || modelOf(r.model);
  const agentEvents = [];
  db.transaction(() => {
    if (existing) {
      updateSession.run(name, status, model, endedAt, lastWrite, cwd, iso(r.time_created), r.id);
    } else {
      insertSession.run(
        r.id,
        name,
        status,
        cwd,
        model,
        iso(r.time_created) || lastWrite,
        endedAt,
        lastWrite,
        JSON.stringify({ source: "opencode-store" })
      );
    }
    claimToolSessions(r.id);
    clearUsage.run(r.id);
    for (const [m, u] of Object.entries(t.usage)) {
      stmts.setTokenUsage.run(
        r.id,
        m,
        "standard",
        "global",
        "standard",
        u.input,
        u.output,
        u.cacheRead,
        u.cacheWrite,
        0,
        0,
        0,
        0
      );
    }
    for (const c of t.children) {
      const id = `opencode-${c.id}`;
      const done = !active || !fresh(Number(c.time_updated) || 0, now);
      const agentStatus = done ? "completed" : "working";
      const ended = done ? iso(c.time_updated) : null;
      const label = (typeof c.title === "string" && c.title.trim().slice(0, 120)) || c.id;
      const prior = stmts.getAgent.get(id);
      if (prior && prior.session_id !== r.id) continue;
      if (!prior) {
        stmts.insertAgent.run(
          id,
          r.id,
          label,
          "subagent",
          "task",
          agentStatus,
          label,
          null,
          JSON.stringify({ source: "opencode-store", transcript: c.id })
        );
      } else if (prior.status !== agentStatus || prior.name !== label) {
        stmts.updateAgent.run(label, agentStatus, label, null, ended, null, id);
      } else continue;
      agentTimes.run(iso(c.time_created), ended, id);
      agentEvents.push([prior ? "agent_updated" : "agent_created", id]);
    }
    upsertTracked.run(r.id, store, t.updated);
  })();
  for (const [type, id] of agentEvents) broadcast(type, stmts.getAgent.get(id));
  return existing ? "session_updated" : "session_created";
}

let running = null;
let lastPending = 0;

/**
 * One pass: every store whose file changed is read, and each root session
 * updated since it was last written is rewritten; an `active` one that went
 * quiet is settled.
 * @returns {Promise<{files: number, sessions: number}>}
 */
function ingestOpencodeUsage({ broadcast = () => {}, now = Date.now() } = {}) {
  if (running) return running;
  running = (async () => {
    let read = 0;
    let written = 0;
    const done = new Set();
    for (const f of storeFiles()) {
      const seen = getStore.get(f.path);
      if (seen && seen.mtime_ms === f.mtime) continue;
      let trees;
      try {
        trees = readStore(f.path);
      } catch (err) {
        if (getStore.get(f.path)?.mtime_ms !== f.mtime) {
          console.warn(`[opencode-usage] skipped ${f.path}: ${err.message}`);
        }
        upsertStore.run(f.path, f.mtime);
        continue;
      }
      read += 1;
      for (const t of trees) {
        const tracked = getTracked.get(t.root.id);
        const changed = !tracked || tracked.time_updated !== t.updated;
        const s = stmts.getSession.get(t.root.id);
        // Unchanged and deleted by the user: stays deleted.
        if (!changed && (!s || s.status !== "active")) continue;
        try {
          const type = writeSession(f.path, t, now, broadcast);
          if (!type) continue;
          done.add(t.root.id);
          written += 1;
          broadcast(type, stmts.getSession.get(t.root.id));
        } catch (err) {
          console.warn(`[opencode-usage] session ${t.root.id}: ${err.message}`);
        }
      }
      upsertStore.run(f.path, f.mtime);
      await new Promise((r) => setImmediate(r));
    }
    // An active session in a store that did not change: settled once quiet.
    let pending = 0;
    for (const { id, time_updated: updated } of activeOpencode.all()) {
      if (done.has(id)) {
        pending += 1;
        continue;
      }
      if (fresh(updated, now) || heldByPane.get(id)) {
        pending += 1;
        continue;
      }
      settle.run(iso(updated) || new Date(now).toISOString(), id);
      for (const a of stmts.listAgentsBySession.all(id)) {
        if (a.status === "completed" || a.status === "error") continue;
        stmts.updateAgent.run(null, "completed", null, null, iso(updated), null, a.id);
        broadcast("agent_updated", stmts.getAgent.get(a.id));
      }
      written += 1;
      broadcast("session_updated", stmts.getSession.get(id));
    }
    lastPending = pending;
    return { files: read, sessions: written };
  })().finally(() => {
    running = null;
  });
  return running;
}

/** A tool part's input/output, kept small. */
const short = (v, max = 10240) => {
  const s = typeof v === "string" ? v : v === undefined ? "" : JSON.stringify(v);
  return s.length > max ? `${s.slice(0, max)}[truncated]` : s;
};
const TEXT_MAX = 1024 * 1024;

/**
 * One OpenCode session's conversation (`sessionId`, or one of its subagent
 * sessions by `childId`), in the Claude transcript shape the client renders:
 * `[{type, sender, timestamp, content}]`, oldest first. Null when unknown.
 */
function transcriptMessages(sessionId, childId = null) {
  const tracked = getTracked.get(sessionId);
  if (!tracked) return null;
  const target = childId || sessionId;
  if (!SESSION_ID.test(target)) return null;
  return withStore(tracked.store, (conn) => {
    if (childId) {
      // Only a session of this tree.
      let cur = conn.prepare("SELECT id, parent_id FROM session WHERE id = ?").get(childId);
      for (let i = 0; i < 32 && cur && cur.id !== sessionId; i++) {
        cur = cur.parent_id
          ? conn.prepare("SELECT id, parent_id FROM session WHERE id = ?").get(cur.parent_id)
          : null;
      }
      if (!cur) return null;
    }
    const messages = conn
      .prepare("SELECT id, time_created, data FROM message WHERE session_id = ? ORDER BY time_created, id")
      .all(target);
    const parts = conn
      .prepare("SELECT message_id, data FROM part WHERE session_id = ? ORDER BY time_created, id")
      .all(target);
    const byMessage = new Map();
    for (const p of parts) {
      let d;
      try {
        d = JSON.parse(p.data);
      } catch {
        continue;
      }
      (byMessage.get(p.message_id) || byMessage.set(p.message_id, []).get(p.message_id)).push(d);
    }
    const out = [];
    for (const m of messages) {
      let info;
      try {
        info = JSON.parse(m.data);
      } catch {
        continue;
      }
      const at = iso(m.time_created);
      const role = info.role === "user" ? "user" : info.role === "assistant" ? "assistant" : null;
      if (!role) continue;
      const content = [];
      const results = [];
      for (const p of byMessage.get(m.id) || []) {
        if (p.type === "text" && typeof p.text === "string" && p.text.trim() && !p.synthetic) {
          content.push({ type: "text", text: short(p.text, TEXT_MAX) });
        } else if (p.type === "reasoning" && typeof p.text === "string" && p.text.trim()) {
          content.push({ type: "thinking", text: short(p.text) });
        } else if (p.type === "tool") {
          const st = p.state || {};
          content.push({
            type: "tool_use",
            name: p.tool || "unknown",
            id: p.callID || null,
            input: st.input && typeof st.input === "object" ? st.input : {},
          });
          if (st.status === "completed" || st.status === "error") {
            results.push({
              type: "tool_result",
              id: p.callID || null,
              output: short(st.status === "error" ? st.error : st.output),
              is_error: st.status === "error",
            });
          }
        }
      }
      if (content.length) {
        const sender = role === "user" ? (childId && !out.length ? "orchestrator" : "user") : "assistant";
        const msg = { type: role, sender, timestamp: at, content };
        if (role === "assistant" && info.modelID) msg.model = info.modelID;
        out.push(msg);
      }
      if (results.length) out.push({ type: "user", sender: "tool", timestamp: at, content: results });
    }
    return out;
  });
}

/** A root session's subagent sessions: `[{id, name, rowId}]`. */
function subagentSessions(sessionId) {
  return stmts
    .listAgentsBySession.all(sessionId)
    .filter((a) => a.id.startsWith("opencode-"))
    .map((a) => ({ id: a.id.slice("opencode-".length), name: a.name, rowId: a.id }));
}

/**
 * First pass at once, then every DASHBOARD_OPENCODE_POLL_MS (default 15 s; 0
 * disables) — once a minute after a pass that found nothing to do.
 */
function startOpencodeUsagePoll(broadcast) {
  const POLL_MS = process.env.DASHBOARD_OPENCODE_POLL_MS
    ? Number(process.env.DASHBOARD_OPENCODE_POLL_MS)
    : 15_000;
  if (!Number.isFinite(POLL_MS) || POLL_MS <= 0) return;
  require("./backoff-poll").startBackoffPoll(
    () =>
      ingestOpencodeUsage({ broadcast }).then((r) => {
        if (r.sessions > 0 && pricingSync.syncEnabled()) void pricingSync.syncPricing(db);
        return r.files + r.sessions + lastPending > 0;
      }),
    { baseMs: POLL_MS }
  );
}

module.exports = {
  getOpencodeDir,
  ingestOpencodeUsage,
  transcriptMessages,
  subagentSessions,
  startOpencodeUsagePoll,
};