platform/src/servers/api/activity/router.ts

import { readdir, stat, readFile, writeFile, mkdir, realpath, open } from 'node:fs/promises';
import { join, dirname, resolve as resolvePath } from 'node:path';
import { createRouter } from '../../create-router';
import { parseTailLine } from './progress';

// "Activity" — follow the agent's background work live. Two producers, one primitive (tail a file):
//  - Harness `run_in_background` tasks write /tmp/claude-<uid>/<encoded-cwd>/tasks/<task_id>.output
//    (append-only, live). Discovered by scanning; the FILE is the source of truth.
//  - Detached (setsid) jobs write an arbitrary log; they self-register via POST /announce with the log
//    path, so they're followable too (the setsid case is on the critical path, not a footnote).
// Owner-only (the account gate confines non-owners to /auth+/music). Paths are traversal-guarded.

export const activityRouter = createRouter();

import { DATA_PATH } from '../../data-path';

const HOME_DIR = process.env.HOME_DIR ?? process.env.HOME ?? '';
const ANNOUNCED_PATH = join(DATA_PATH, 'activity', 'announced.json');
const ALLOWED_ROOTS = ['/tmp', DATA_PATH, HOME_DIR].filter(Boolean);
const ACTIVE_WINDOW_MS = 120_000; // a task file touched within this is considered "active"

type TaskFile = { taskId: string; path: string; cwdLabel: string; sizeBytes: number; mtimeMs: number };
type Announced = { name: string; path: string; ts: number };

// Scan the claude tmp tree for harness background-task output files (…/tasks/<task_id>.output).
async function listTaskFiles(): Promise<TaskFile[]> {
  const out: TaskFile[] = [];
  let tmp: string[];
  try {
    tmp = await readdir('/tmp');
  } catch {
    return out;
  }
  for (const name of tmp) {
    if (!name.startsWith('claude-')) continue;
    const base = join('/tmp', name);
    let cwdDirs: import('node:fs').Dirent[];
    try {
      cwdDirs = await readdir(base, { withFileTypes: true });
    } catch {
      continue;
    }
    for (const cwd of cwdDirs) {
      if (!cwd.isDirectory()) continue;
      const tasksDir = join(base, cwd.name, 'tasks');
      let files: string[];
      try {
        files = await readdir(tasksDir);
      } catch {
        continue;
      }
      for (const f of files) {
        if (!f.endsWith('.output')) continue;
        const p = join(tasksDir, f);
        const st = await stat(p).catch(() => null);
        if (!st) continue;
        out.push({ taskId: f.slice(0, -'.output'.length), path: p, cwdLabel: cwd.name, sizeBytes: st.size, mtimeMs: st.mtimeMs });
      }
    }
  }
  return out;
}

async function readAnnounced(): Promise<Announced[]> {
  try {
    return JSON.parse(await readFile(ANNOUNCED_PATH, 'utf8')) as Announced[];
  } catch {
    return [];
  }
}

/** Resolve a path only if it lands under an allowed root (guards traversal); the file need not exist yet. */
async function resolveAllowed(p: string): Promise<string | null> {
  const under = (abs: string) => ALLOWED_ROOTS.some((root) => abs === root || abs.startsWith(root + '/'));
  try {
    const real = await realpath(p);
    return under(real) ? real : null;
  } catch {
    const abs = resolvePath(p);
    return under(abs) ? abs : null;
  }
}

// GET /tasks — registry of followable background work.
activityRouter.get('/tasks', async (ctx) => {
  const [harness, announced] = await Promise.all([listTaskFiles(), readAnnounced()]);
  const now = Date.now();
  return ctx.json({
    tasks: harness
      .sort((a, b) => b.mtimeMs - a.mtimeMs)
      .map((t) => ({
        id: t.taskId,
        source: 'harness' as const,
        cwd: t.cwdLabel,
        sizeBytes: t.sizeBytes,
        updatedAt: Math.round(t.mtimeMs),
        active: now - t.mtimeMs < ACTIVE_WINDOW_MS,
      })),
    detached: announced.map((a) => ({ id: a.name, source: 'detached' as const, path: a.path, announcedAt: a.ts })),
  });
});

// POST /announce { name, path } — register a detached (setsid) job's log so it's followable too.
activityRouter.post('/announce', async (ctx) => {
  const body = ctx.get('body') as { name?: string; path?: string } | undefined;
  if (!body?.name || !body?.path) return ctx.text('name and path are required', 400);
  const safe = await resolveAllowed(body.path);
  if (!safe) return ctx.text('path is not under an allowed root', 403);
  await mkdir(dirname(ANNOUNCED_PATH), { recursive: true });
  const list = await readAnnounced();
  const next = [{ name: body.name, path: safe, ts: Date.now() }, ...list.filter((a) => a.name !== body.name)].slice(0, 100);
  await writeFile(ANNOUNCED_PATH, JSON.stringify(next));
  return ctx.json({ ok: true, name: body.name, path: safe });
});

// GET /stream?task=<id> | ?path=<abs> — SSE that tails the output file, emitting {kind:'line'|'progress'}.
activityRouter.get('/stream', async (ctx) => {
  const task = ctx.req.query('task');
  const pathQ = ctx.req.query('path');

  let target: string | null = null;
  if (task) {
    target = (await listTaskFiles()).find((f) => f.taskId === task)?.path ?? null;
  } else if (pathQ) {
    target = await resolveAllowed(pathQ);
  }
  if (!target) return ctx.text('a valid task id or path is required', 400);

  const filePath = target;
  const enc = new TextEncoder();

  const stream = new ReadableStream<Uint8Array>({
    async start(controller) {
      let offset = 0;
      let closed = false;
      const send = (obj: unknown) => {
        try {
          controller.enqueue(enc.encode(`data: ${JSON.stringify(obj)}\n\n`));
        } catch {
          /* controller closed */
        }
      };

      const drain = async () => {
        try {
          const st = await stat(filePath);
          if (st.size < offset) offset = 0; // truncated / rotated
          if (st.size > offset) {
            const fh = await open(filePath, 'r');
            const buf = Buffer.alloc(st.size - offset);
            await fh.read(buf, 0, buf.length, offset);
            await fh.close();
            offset = st.size;
            for (const line of buf.toString('utf8').split('\n')) {
              if (line !== '') send(parseTailLine(line));
            }
          }
        } catch {
          /* file may not exist yet — keep polling */
        }
      };

      await drain();
      const poll = setInterval(drain, 800);
      const heartbeat = setInterval(() => {
        try {
          controller.enqueue(enc.encode(': hb\n\n'));
        } catch {
          /* closed */
        }
      }, 15_000);

      const cleanup = () => {
        if (closed) return;
        closed = true;
        clearInterval(poll);
        clearInterval(heartbeat);
        try {
          controller.close();
        } catch {
          /* already closed */
        }
      };
      ctx.req.raw.signal.addEventListener('abort', cleanup);
      setTimeout(cleanup, 60 * 60 * 1000); // safety cap
    },
  });

  return new Response(stream, {
    headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive' },
  });
});