feat: add live executor observability

2026-07-17 00:48:52 +08:00
parent 3566a73a03
commit c9d76d8f58
13 changed files with 1546 additions and 28 deletions
@@ -41,6 +41,7 @@ import {
  initWorkflowState,
  nowIso,
  readState,
+  readProgressSidecar,
  releaseLock,
  runDir as makeRunDir,
  writeState,
@@ -179,20 +180,51 @@ function transitionToTerminal(
  releaseLock(state.repoRoot, state.runId);
 }

+/**
+ * Register an attempt log path in the cycle record BEFORE spawning the executor
+ * so that a crash during execution still leaves a discoverable log file.
+ */
+function registerAttemptLog(dir: string, cycleRecord: CycleRecord, logFilePath: string): void {
+  const priorLogs = cycleRecord.executorAttemptLogs
+    ?? (cycleRecord.executorLogFile ? [cycleRecord.executorLogFile] : []);
+  const relativeLog = path.relative(dir, logFilePath);
+  cycleRecord.executorAttemptLogs = [...priorLogs, relativeLog];
+  cycleRecord.executorLogFile = relativeLog;
+}
+
+/**
+ * Finalize an attempt after the executor finishes: set exit code and report slice.
+ * The log path is already registered (by registerAttemptLog), so we never re-derive it.
+ */
 function persistExecutorAttempt(
  dir: string,
  cycleRecord: CycleRecord,
  execResult: ExecutorResult,
+  logFilePath?: string,
 ): void {
-  const priorLogs = cycleRecord.executorAttemptLogs
-    ?? (cycleRecord.executorLogFile ? [cycleRecord.executorLogFile] : []);
-  const attemptIndex = priorLogs.length;
+  // If the log file was created and registered before spawning (even if empty),
+  // keep it and don't overwrite. Empty logs are valid (e.g. immediate exit).
+  if (logFilePath && fs.existsSync(logFilePath)) {
+    const relativeLog = path.relative(dir, logFilePath);
+    cycleRecord.executorLogFile = relativeLog;
+    cycleRecord.executorExitCode = execResult.exitCode ?? undefined;
+    cycleRecord.executorReport = execResult.report.slice(0, 2000);
+    if (!cycleRecord.executorAttemptLogs?.includes(relativeLog)) {
+      cycleRecord.executorAttemptLogs = [...(cycleRecord.executorAttemptLogs ?? []), relativeLog];
+    }
+    return;
+  }
+
+  // Fallback: write the full report if logFilePath wasn't provided (synchronous adapter or legacy path)
+  const attemptIndex = (cycleRecord.executorAttemptLogs?.length ?? 0);
  const execLog = attemptIndex === 0
    ? cycleExecLogFile(dir, cycleRecord.cycleIndex)
    : path.join(dir, `cycle-${cycleRecord.cycleIndex}-exec-retry-${attemptIndex}.log`);
  fs.writeFileSync(execLog, execResult.report, "utf8");
  const relativeLog = path.relative(dir, execLog);
-  cycleRecord.executorAttemptLogs = [...priorLogs, relativeLog];
+  if (!cycleRecord.executorAttemptLogs?.includes(relativeLog)) {
+    cycleRecord.executorAttemptLogs = [...(cycleRecord.executorAttemptLogs ?? []), relativeLog];
+  }
  cycleRecord.executorLogFile = relativeLog;
  cycleRecord.executorExitCode = execResult.exitCode ?? undefined;
  cycleRecord.executorReport = execResult.report.slice(0, 2000);
@@ -318,6 +350,13 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s

  const adapter = createExecutorAdapter(config.executor);
  const { ac, cleanup } = createManagedController(state.timeoutSeconds);
+
+  // Create attempt log file BEFORE spawning the executor
+  const execLogPath = path.join(dir, `cycle-${cycleIndex}-exec.log`);
+  fs.writeFileSync(execLogPath, "", "utf8");
+  registerAttemptLog(dir, cycleRecord, execLogPath);
+  writeState(dir, state);  // persist so retry can discover the log
+
  let execResult;
  try {
    execResult = await adapter.start({
@@ -325,6 +364,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
      plan: typedPlan,
      runDir: dir,
      cycleIndex,
+      logFilePath: execLogPath,
      config: config.executorConfig,
      timeoutSeconds: state.timeoutSeconds,
      signal: ac.signal,
@@ -336,7 +376,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
  }
  cleanup();

-  persistExecutorAttempt(dir, cycleRecord, execResult);
+  persistExecutorAttempt(dir, cycleRecord, execResult, execLogPath);
  if (execResult.sessionHandle) {
    state.sessionHandle = execResult.sessionHandle;
    if (execResult.sessionHandle.kind === "agy") {
@@ -716,6 +756,18 @@ async function resumeExecutorCycle(

  const adapter = createExecutorAdapter(state.executor);
  const { ac, cleanup } = createManagedController(state.timeoutSeconds);
+
+  // Determine attempt index and create log file before spawning
+  const priorLogs = cycleRecord.executorAttemptLogs
+    ?? (cycleRecord.executorLogFile ? [cycleRecord.executorLogFile] : []);
+  const attemptIndex = priorLogs.length;
+  const execLogPath = attemptIndex === 0
+    ? path.join(dir, `cycle-${cycleIndex}-exec.log`)
+    : path.join(dir, `cycle-${cycleIndex}-exec-retry-${attemptIndex}.log`);
+  fs.writeFileSync(execLogPath, "", "utf8");
+  registerAttemptLog(dir, cycleRecord, execLogPath);
+  writeState(dir, state);  // persist so retry can discover the log
+
  let execResult;
  try {
    execResult = await adapter.resume({
@@ -725,6 +777,7 @@ async function resumeExecutorCycle(
      handle: state.sessionHandle,
      runDir: dir,
      cycleIndex,
+      logFilePath: execLogPath,
      config: state.executorConfig ?? {},
      timeoutSeconds: state.timeoutSeconds,
      signal: ac.signal,
@@ -742,7 +795,7 @@ async function resumeExecutorCycle(
  }
  cleanup();

-  persistExecutorAttempt(dir, cycleRecord, execResult);
+  persistExecutorAttempt(dir, cycleRecord, execResult, execLogPath);
  if (execResult.sessionHandle) {
    state.sessionHandle = execResult.sessionHandle;
    if (execResult.sessionHandle.kind === "agy") {
@@ -1088,6 +1141,26 @@ export function statusWorkflow(opts: StatusWorkflowOpts): void {
  const { state, dir } = requireState(opts.runId);
  const cycle = state.cycles.find((c) => c.cycleIndex === state.currentCycle);

+  // Read live progress sidecar for real-time observability
+  const sidecar = readProgressSidecar(dir);
+  const liveLogFile = cycle?.executorLogFile
+    ? path.join(dir, cycle.executorLogFile)
+    : sidecar?.logFilePath
+      ? path.join(dir, sidecar.logFilePath)
+      : undefined;
+  let executionElapsed: string | undefined;
+  if (sidecar?.executionDurationMs !== undefined) {
+    const ms = sidecar.executionDurationMs;
+    const secs = Math.floor(ms / 1000);
+    if (secs < 60) executionElapsed = `${secs}s`;
+    else executionElapsed = `${Math.floor(secs / 60)}m ${secs % 60}s`;
+  } else if (sidecar?.startedAt) {
+    const ms = Date.now() - new Date(sidecar.startedAt).getTime();
+    const secs = Math.floor(ms / 1000);
+    if (secs < 60) executionElapsed = `${secs}s`;
+    else executionElapsed = `${Math.floor(secs / 60)}m ${secs % 60}s`;
+  }
+
  const output: WorkflowStatusOutput = {
    runId: state.runId,
    status: state.status,
@@ -1108,6 +1181,12 @@ export function statusWorkflow(opts: StatusWorkflowOpts): void {
    repoRoot: state.repoRoot,
    createdAt: state.createdAt,
    updatedAt: state.updatedAt,
+    liveLogFile,
+    executionElapsed,
+    lastActivity: sidecar?.lastActivityAt,
+    logBytes: sidecar?.logBytes,
+    executorPid: sidecar?.pid,
+    activitySummary: sidecar?.activitySummary,
  };

  if (opts.json) {
@@ -1133,6 +1212,14 @@ export function statusWorkflow(opts: StatusWorkflowOpts): void {
    lines.push(p("Session handle", kind + extra));
  }

+  // Live execution fields
+  if (output.executorPid) lines.push(p("Executor PID", String(output.executorPid)));
+  if (output.executionElapsed) lines.push(p("Elapsed", output.executionElapsed));
+  if (output.activitySummary) lines.push(p("Activity", output.activitySummary));
+  if (output.lastActivity) lines.push(p("Last activity", output.lastActivity));
+  if (output.logBytes !== undefined) lines.push(p("Log bytes", `${output.logBytes}`));
+  if (output.liveLogFile) lines.push(p("Live log", output.liveLogFile));
+
  if (cycle?.hostReviewFile) lines.push(p("Host review", path.join(dir, cycle.hostReviewFile)));
  if (state.status === "awaiting_scope_resolution" && cycle?.scopeViolations?.length) {
    lines.push(p("Scope violations", String(cycle.scopeViolations.length)));
@@ -1183,3 +1270,158 @@ export function abortWorkflow(opts: AbortWorkflowOpts): void {
  process.stdout.write(`workflow aborted: ${opts.runId}\nreason: ${opts.reason}\n`);
  emitWorkflowMeta({ runId: opts.runId, status: "aborted", reason: opts.reason });
 }
+
+// ---------------------------------------------------------------------------
+// workflow logs
+// ---------------------------------------------------------------------------
+
+export interface LogsWorkflowOpts {
+  runId: string;
+  follow: boolean;
+  tail: number;
+  repoRoot?: string;
+}
+
+export async function logsWorkflow(opts: LogsWorkflowOpts): Promise<void> {
+  const { state, dir } = requireState(opts.runId);
+
+  // Resolve current log path: sidecar is freshest during execution
+  const sidecar = readProgressSidecar(dir);
+  let logPath: string | undefined;
+  if (sidecar?.logFilePath && fs.existsSync(path.join(dir, sidecar.logFilePath))) {
+    logPath = path.join(dir, sidecar.logFilePath);
+  } else {
+    const cycle = state.cycles.find((c) => c.cycleIndex === state.currentCycle);
+    if (cycle?.executorLogFile) {
+      logPath = path.join(dir, cycle.executorLogFile);
+    }
+  }
+
+  // If no log file and state is executing, wait briefly for it to appear
+  const isExecuting = state.status === "executing";
+  if (!logPath && isExecuting) {
+    for (let i = 0; i < 30; i++) {
+      await sleepMs(500);
+      const sc = readProgressSidecar(dir);
+      if (sc?.logFilePath && fs.existsSync(path.join(dir, sc.logFilePath))) {
+        logPath = path.join(dir, sc.logFilePath);
+        break;
+      }
+      const updatedState = readState(dir);
+      if (updatedState) {
+        const uc = updatedState.cycles.find((c) => c.cycleIndex === updatedState.currentCycle);
+        if (uc?.executorLogFile) {
+          logPath = path.join(dir, uc.executorLogFile);
+          break;
+        }
+      }
+    }
+  }
+
+  if (!logPath || !fs.existsSync(logPath)) {
+    process.stdout.write(`No log file found for run '${opts.runId}'.\n`);
+    return;
+  }
+
+  // Tail the last N lines
+  const lines = readTailLines(logPath, opts.tail);
+  for (const line of lines) {
+    process.stdout.write(line + "\n");
+  }
+
+  if (!opts.follow) return;
+
+  // Follow mode: watch for new data and retry rotations
+  let currentPath = logPath;
+  let lastByteOffset = fs.statSync(currentPath).size;
+
+  // States that indicate execution has ceased (no more output expected)
+  const terminalStates = new Set([
+    "completed", "needs_human", "blocked", "budget_exhausted", "aborted",
+    "awaiting_review", "awaiting_host", "awaiting_scope_resolution",
+  ]);
+
+  while (true) {
+    const curState = readState(dir);
+    if (!curState) break;
+
+    // Exit if the workflow has stopped executing
+    if (terminalStates.has(curState.status)) {
+      drainBytes(currentPath, lastByteOffset);
+      return;
+    }
+
+    // Check for log rotation to a newer attempt
+    const sc = readProgressSidecar(dir);
+    let candidatePath: string | undefined;
+    if (sc?.logFilePath && fs.existsSync(path.join(dir, sc.logFilePath))) {
+      candidatePath = path.join(dir, sc.logFilePath);
+    } else {
+      const curCycle = curState.cycles.find((c) => c.cycleIndex === curState.currentCycle);
+      if (curCycle?.executorLogFile) {
+        candidatePath = path.join(dir, curCycle.executorLogFile);
+      }
+    }
+
+    if (candidatePath && candidatePath !== currentPath) {
+      drainBytes(currentPath, lastByteOffset);
+      currentPath = candidatePath;
+      lastByteOffset = fs.statSync(currentPath).size;
+      process.stdout.write(`\n--- log rotated to ${path.basename(currentPath)} ---\n`);
+      const headLines = readTailLines(currentPath, Math.min(opts.tail, 10));
+      for (const line of headLines) {
+        process.stdout.write(line + "\n");
+      }
+      continue;
+    }
+
+    // Drain new bytes since lastByteOffset using byte-accurate reading
+    try {
+      const stats = fs.statSync(currentPath);
+      if (stats.size > lastByteOffset) {
+        drainBytes(currentPath, lastByteOffset);
+        lastByteOffset = stats.size;
+      }
+    } catch {
+      // File may have been deleted or rotated
+    }
+
+    await sleepMs(500);
+  }
+}
+
+function sleepMs(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function readTailLines(filePath: string, n: number): string[] {
+  try {
+    const content = fs.readFileSync(filePath, "utf8");
+    const lines = content.split("\n");
+    return lines.slice(Math.max(0, lines.length - n));
+  } catch {
+    return [];
+  }
+}
+
+/**
+ * Drain all bytes starting at fromByteOffset using proper byte-level I/O.
+ */
+function drainBytes(filePath: string, fromByteOffset: number): void {
+  try {
+    const fd = fs.openSync(filePath, "r");
+    try {
+      const stats = fs.fstatSync(fd);
+      const remaining = stats.size - fromByteOffset;
+      if (remaining > 0) {
+        const buf = Buffer.alloc(remaining);
+        fs.readSync(fd, buf, 0, remaining, fromByteOffset);
+        process.stdout.write(buf.toString("utf8"));
+      }
+    } finally {
+      fs.closeSync(fd);
+    }
+  } catch {
+    // ignore
+  }
+}