openclaw/src/agents/bash-tools.process.ts

/**
 * Process-control tool factory.
 * Lists, polls, logs, writes to, sends keys to, pastes into, kills, clears,
 * and removes background exec sessions.
 */
import { createAbortError as createNamedAbortError } from "../infra/abort-signal.js";
import { formatDurationCompact } from "../infra/format-time/format-duration.ts";
import { getDiagnosticSessionState } from "../logging/diagnostic-session-state.js";
import { getProcessSupervisor } from "../process/supervisor/index.js";
import { cancelBackgroundExecSession } from "./bash-process-control.js";
import {
  acknowledgeNotifyOnExit,
  type ProcessSession,
  deleteSession,
  drainFinishedSession,
  drainSession,
  getFinishedSession,
  getFinishedSessionForProcess,
  getSession,
  listFinishedSessions,
  listRunningSessions,
  markTerminalPollObserved,
  setJobTtlMs,
} from "./bash-process-registry.js";
import { describeProcessTool } from "./bash-tools.descriptions.js";
import { appendExecTimeoutRetryGuidance, renderExecExitLabel } from "./bash-tools.exec-output.js";
import {
  handleProcessSendKeys,
  type WritableStdin,
  writeProcessStdin,
} from "./bash-tools.process-send-keys.js";
import { processSchema } from "./bash-tools.schemas.js";
import {
  clampWithDefault,
  deriveSessionName,
  padProcessStatus,
  readEnvInt,
  sliceLogLines,
  truncateMiddle,
} from "./bash-tools.shared.js";
import { recordCommandPoll, resetCommandPollCount } from "./command-poll-backoff.js";
import { encodePaste } from "./pty-keys.js";
import type { AgentToolResult } from "./runtime/index.js";
import { PROCESS_TOOL_DISPLAY_SUMMARY } from "./tool-description-presets.js";
import type { AgentToolWithMeta } from "./tools/common.js";

/** Defaults injected by tests, agent scopes, and scoped process registries. */
export type ProcessToolDefaults = {
  cleanupMs?: number;
  hasCronTool?: boolean;
  inputWaitIdleMs?: number;
  scopeKey?: string;
};

const DEFAULT_LOG_TAIL_LINES = 200;
const DEFAULT_INPUT_WAIT_IDLE_MS = 15_000;
const MIN_INPUT_WAIT_IDLE_MS = 1_000;
const MAX_INPUT_WAIT_IDLE_MS = 10 * 60 * 1000;
const PROCESS_TOOL_ACTIONS = (
  processSchema.properties.action as typeof processSchema.properties.action & {
    enum: readonly string[];
  }
).enum;
type ProcessToolAction = (typeof PROCESS_TOOL_ACTIONS)[number];

function resolveLogSliceWindow(offset?: number, limit?: number) {
  const usingDefaultTail = offset === undefined && limit === undefined;
  const effectiveLimit =
    typeof limit === "number" && Number.isFinite(limit)
      ? limit
      : usingDefaultTail
        ? DEFAULT_LOG_TAIL_LINES
        : undefined;
  return { effectiveOffset: offset, effectiveLimit, usingDefaultTail };
}

function defaultTailNote(totalLines: number, usingDefaultTail: boolean) {
  if (!usingDefaultTail || totalLines <= DEFAULT_LOG_TAIL_LINES) {
    return "";
  }
  return `\n\n[showing last ${DEFAULT_LOG_TAIL_LINES} of ${totalLines} lines; pass offset/limit to page]`;
}

function retentionCapNote(session: Pick<ProcessSession, "totalOutputChars" | "aggregated">) {
  return session.totalOutputChars > session.aggregated.length
    ? "\n\n[earlier output was discarded at the retention cap and cannot be recovered]"
    : "";
}

const MAX_POLL_WAIT_MS = 30_000;

type RunningSessionRuntime = {
  stdinWritable: boolean;
  waitingForInput: boolean;
  idleMs: number;
  lastOutputAt: number;
};

function resolveSessionStdin(session: ProcessSession): WritableStdin | undefined {
  return session.stdin as WritableStdin | undefined;
}

function isWritableStdin(stdin: WritableStdin | undefined): stdin is WritableStdin {
  if (!stdin || stdin.destroyed) {
    return false;
  }
  if (stdin.writable === false || stdin.writableEnded === true || stdin.writableFinished === true) {
    return false;
  }
  return true;
}

function runningSessionInputDetails(runtime: RunningSessionRuntime) {
  return {
    stdinWritable: runtime.stdinWritable,
    waitingForInput: runtime.waitingForInput,
    idleMs: runtime.idleMs,
    lastOutputAt: runtime.lastOutputAt,
  };
}

function resolvePollWaitMs(value: unknown) {
  if (typeof value === "number" && Number.isFinite(value)) {
    return Math.max(0, Math.min(MAX_POLL_WAIT_MS, Math.floor(value)));
  }
  if (typeof value === "string" && /^[+-]?\d+$/.test(value.trim())) {
    const parsed = Number(value.trim());
    if (Number.isSafeInteger(parsed)) {
      return Math.max(0, Math.min(MAX_POLL_WAIT_MS, parsed));
    }
  }
  return 0;
}

function failText(text: string): AgentToolResult<unknown> {
  return {
    content: [
      {
        type: "text",
        text,
      },
    ],
    details: { status: "failed" },
  };
}

function recordPollRetrySuggestion(sessionId: string, hasNewOutput: boolean): number | undefined {
  try {
    const sessionState = getDiagnosticSessionState({ sessionId });
    return recordCommandPoll(sessionState, sessionId, hasNewOutput);
  } catch {
    return undefined;
  }
}

function resetPollRetrySuggestion(sessionId: string): void {
  try {
    const sessionState = getDiagnosticSessionState({ sessionId });
    resetCommandPollCount(sessionState, sessionId);
  } catch {
    // Ignore diagnostics state failures for process tool behavior.
  }
}

type FinishedSession = NonNullable<ReturnType<typeof getFinishedSession>>;

function finishedPollResult(
  sessionId: string,
  finished: FinishedSession,
): AgentToolResult<unknown> {
  resetPollRetrySuggestion(sessionId);
  acknowledgeNotifyOnExit(finished);
  const { stdout, stderr, outputDropped } = drainFinishedSession(finished);
  const output = [stdout.trimEnd(), stderr.trimEnd()].filter(Boolean).join("\n").trim();
  // Omitted retained output is pageable only while this public id still owns
  // the exact snapshot; a reused slug must never point the model at successor logs.
  const retainedOutputNote = outputDropped
    ? getFinishedSession(sessionId) === finished
      ? "\n\n[earlier output is omitted from this poll; use action=log with offset and limit to inspect retained output]"
      : "\n\n[earlier output is omitted from this poll; omitted output is no longer available through action=log]"
    : "";
  return {
    content: [
      {
        type: "text",
        text: appendExecTimeoutRetryGuidance(
          (output || "(no new output)") +
            retentionCapNote(finished) +
            retainedOutputNote +
            `\n\nProcess exited with ${renderExecExitLabel(finished)}.`,
          finished.exitReason,
        ),
      },
    ],
    details: {
      status: finished.status === "completed" ? "completed" : "failed",
      sessionId,
      exitCode: finished.exitCode ?? undefined,
      ...(finished.exitSignal != null ? { exitSignal: finished.exitSignal } : {}),
      ...(finished.exitReason
        ? {
            exitReason: finished.exitReason,
            timedOut:
              finished.exitReason === "overall-timeout" ||
              finished.exitReason === "no-output-timeout",
          }
        : {}),
      ...(finished.noOutputTimedOut !== undefined
        ? { noOutputTimedOut: finished.noOutputTimedOut }
        : {}),
      aggregated: finished.aggregated,
      name: deriveSessionName(finished.command),
    },
  };
}

function createAbortError(reason: unknown): Error {
  if (reason instanceof Error) {
    return reason;
  }
  return createNamedAbortError(typeof reason === "string" ? reason : "Aborted");
}

async function sleepPollInterval(ms: number, signal?: AbortSignal): Promise<void> {
  if (signal?.aborted) {
    throw createAbortError(signal.reason);
  }
  await new Promise<void>((resolve, reject) => {
    const cleanup = () => {
      if (timer) {
        clearTimeout(timer);
      }
      if (onAbort) {
        signal?.removeEventListener("abort", onAbort);
      }
    };
    const onResolve = () => {
      cleanup();
      resolve();
    };
    const onAbort: (() => void) | undefined = () => {
      cleanup();
      reject(createAbortError(signal?.reason));
    };
    const timer: ReturnType<typeof setTimeout> | undefined = setTimeout(onResolve, ms);
    timer.unref?.();
    signal?.addEventListener("abort", onAbort, { once: true });
  });
}

/** Build the process-control tool with optional cleanup, scope, and input-idle defaults. */
export function createProcessTool(
  defaults?: ProcessToolDefaults,
): AgentToolWithMeta<typeof processSchema, unknown> {
  if (defaults?.cleanupMs !== undefined) {
    setJobTtlMs(defaults.cleanupMs);
  }
  const scopeKey = defaults?.scopeKey;
  const supervisor = getProcessSupervisor();
  const inputWaitIdleMs = clampWithDefault(
    defaults?.inputWaitIdleMs ?? readEnvInt("OPENCLAW_PROCESS_INPUT_WAIT_IDLE_MS"),
    DEFAULT_INPUT_WAIT_IDLE_MS,
    MIN_INPUT_WAIT_IDLE_MS,
    MAX_INPUT_WAIT_IDLE_MS,
  );
  const isInScope = (session?: { scopeKey?: string } | null) =>
    !scopeKey || session?.scopeKey === scopeKey;

  const describeRunningSession = (session: ProcessSession): RunningSessionRuntime => {
    const record = supervisor.getRecord(session.id);
    const lastOutputAt = record?.lastOutputAtMs ?? session.startedAt;
    const idleMs = Math.max(0, Date.now() - lastOutputAt);
    const stdinWritable = isWritableStdin(resolveSessionStdin(session));
    return {
      stdinWritable,
      waitingForInput: stdinWritable && idleMs >= inputWaitIdleMs,
      idleMs,
      lastOutputAt,
    };
  };

  const buildInputWaitHint = (runtime: RunningSessionRuntime | undefined) => {
    if (!runtime?.waitingForInput) {
      return "";
    }
    const idle = formatDurationCompact(runtime.idleMs) ?? `${runtime.idleMs}ms`;
    return `\n\nNo new output for ${idle}; this session may be waiting for input. Use process write, send-keys, submit, or paste to provide input.`;
  };

  return {
    name: "process",
    label: "process",
    displaySummary: PROCESS_TOOL_DISPLAY_SUMMARY,
    description: describeProcessTool({ hasCronTool: defaults?.hasCronTool === true }),
    parameters: processSchema,
    execute: async (_toolCallId, args, signal, _onUpdate): Promise<AgentToolResult<unknown>> => {
      const action = (args as { action?: unknown }).action;
      if (!PROCESS_TOOL_ACTIONS.includes(action as ProcessToolAction)) {
        return failText(
          `Invalid process action. Expected one of: ${PROCESS_TOOL_ACTIONS.join(", ")}`,
        );
      }
      const params = args as {
        action: ProcessToolAction;
        sessionId?: string;
        data?: string;
        keys?: string[];
        hex?: string[];
        literal?: string;
        text?: string;
        bracketed?: boolean;
        eof?: boolean;
        offset?: number;
        limit?: number;
        timeout?: unknown;
      };

      if (params.action === "list") {
        const running = listRunningSessions()
          .filter((s) => isInScope(s))
          .map((s) => {
            const runtime = describeRunningSession(s);
            return {
              sessionId: s.id,
              status: "running",
              pid: s.pid ?? undefined,
              startedAt: s.startedAt,
              runtimeMs: Date.now() - s.startedAt,
              cwd: s.cwd,
              command: s.command,
              name: deriveSessionName(s.command),
              tail: s.tail,
              truncated: s.truncated,
              stdinWritable: runtime.stdinWritable,
              waitingForInput: runtime.waitingForInput,
              idleMs: runtime.idleMs,
              lastOutputAt: runtime.lastOutputAt,
            };
          });
        const finished = listFinishedSessions()
          .filter((s) => isInScope(s))
          .map((s) => ({
            sessionId: s.id,
            status: s.status,
            startedAt: s.startedAt,
            endedAt: s.endedAt,
            runtimeMs: s.endedAt - s.startedAt,
            cwd: s.cwd,
            command: s.command,
            name: deriveSessionName(s.command),
            tail: s.tail,
            truncated: s.truncated,
            exitCode: s.exitCode ?? undefined,
            exitSignal: s.exitSignal ?? undefined,
          }));
        const lines = [...running, ...finished]
          .toSorted((a, b) => b.startedAt - a.startedAt)
          .map((s) => {
            const label = s.name ? truncateMiddle(s.name, 80) : truncateMiddle(s.command, 120);
            const marker = "waitingForInput" in s && s.waitingForInput ? " [input-wait]" : "";
            return `${s.sessionId} ${padProcessStatus(s.status, 9)} ${
              formatDurationCompact(s.runtimeMs) ?? "n/a"
            }${marker} :: ${label}`;
          });
        return {
          content: [
            {
              type: "text",
              text: lines.join("\n") || "No running or recent sessions.",
            },
          ],
          details: { status: "completed", sessions: [...running, ...finished] },
        };
      }

      if (!params.sessionId) {
        return {
          content: [{ type: "text", text: "sessionId is required for this action." }],
          details: { status: "failed" },
        };
      }

      const session = getSession(params.sessionId);
      const finished = getFinishedSession(params.sessionId);
      const scopedSession = isInScope(session) ? session : undefined;
      const scopedFinished = isInScope(finished) ? finished : undefined;

      const failedResult = (text: string): AgentToolResult<unknown> => ({
        content: [{ type: "text", text }],
        details: { status: "failed" },
      });

      const resolveBackgroundedWritableStdin = () => {
        if (!scopedSession) {
          return {
            ok: false as const,
            result: failedResult(`No active session found for ${params.sessionId}`),
          };
        }
        if (!scopedSession.backgrounded) {
          return {
            ok: false as const,
            result: failedResult(`Session ${params.sessionId} is not backgrounded.`),
          };
        }
        if (scopedSession.finalizing) {
          return {
            ok: false as const,
            result: failedResult(`Session ${params.sessionId} is finalizing.`),
          };
        }
        const stdin = resolveSessionStdin(scopedSession);
        if (!isWritableStdin(stdin)) {
          return {
            ok: false as const,
            result: failedResult(`Session ${params.sessionId} stdin is not writable.`),
          };
        }
        return { ok: true as const, session: scopedSession, stdin };
      };

      const runningSessionResult = (
        sessionLocal: ProcessSession,
        text: string,
      ): AgentToolResult<unknown> => ({
        content: [{ type: "text", text }],
        details: {
          status: "running",
          sessionId: params.sessionId,
          name: deriveSessionName(sessionLocal.command),
        },
      });

      switch (params.action) {
        case "poll": {
          if (!scopedSession) {
            if (scopedFinished) {
              return finishedPollResult(params.sessionId, scopedFinished);
            }
            resetPollRetrySuggestion(params.sessionId);
            return failText(`No session found for ${params.sessionId}`);
          }
          if (!scopedSession.backgrounded) {
            return failText(`Session ${params.sessionId} is not backgrounded.`);
          }
          const pollWaitMs = resolvePollWaitMs(params.timeout);
          if (pollWaitMs > 0 && !scopedSession.exited) {
            const deadline = Date.now() + pollWaitMs;
            while (!scopedSession.exited && Date.now() < deadline) {
              await sleepPollInterval(Math.max(0, Math.min(250, deadline - Date.now())), signal);
            }
          }
          if (scopedSession.exited) {
            markTerminalPollObserved(scopedSession);
            // Exit finalization owns the terminal transition. Re-read by process
            // object because the public id may already index a successor.
            const finishedAfterWait = getFinishedSessionForProcess(scopedSession);
            if (finishedAfterWait && isInScope(finishedAfterWait)) {
              return finishedPollResult(params.sessionId, finishedAfterWait);
            }
            resetPollRetrySuggestion(params.sessionId);
            return failText(`No session found for ${params.sessionId}`);
          }
          const { stdout, stderr, outputDropped } = drainSession(scopedSession);
          const output = [stdout.trimEnd(), stderr.trimEnd()].filter(Boolean).join("\n").trim();
          const aggregateOutputNote = retentionCapNote(scopedSession);
          const retainedOutputNote = outputDropped
            ? "\n\n[earlier output is omitted from this poll; use action=log with offset and limit to inspect retained output]"
            : "";
          const hasNewOutput = output.length > 0;
          const retryInMs = recordPollRetrySuggestion(params.sessionId, hasNewOutput);
          const runtime = describeRunningSession(scopedSession);
          return {
            content: [
              {
                type: "text",
                text: appendExecTimeoutRetryGuidance(
                  (output || "(no new output)") +
                    aggregateOutputNote +
                    retainedOutputNote +
                    (buildInputWaitHint(runtime) || "\n\nProcess still running."),
                  undefined,
                ),
              },
            ],
            details: {
              status: "running",
              sessionId: params.sessionId,
              aggregated: scopedSession.aggregated,
              name: deriveSessionName(scopedSession.command),
              ...runningSessionInputDetails(runtime),
              ...(typeof retryInMs === "number" ? { retryInMs } : {}),
            },
          };
        }

        case "log": {
          if (scopedSession) {
            if (!scopedSession.backgrounded) {
              return {
                content: [
                  {
                    type: "text",
                    text: `Session ${params.sessionId} is not backgrounded.`,
                  },
                ],
                details: { status: "failed" },
              };
            }
            const window = resolveLogSliceWindow(params.offset, params.limit);
            const { slice, totalLines, totalChars } = sliceLogLines(
              scopedSession.aggregated,
              window.effectiveOffset,
              window.effectiveLimit,
            );
            const runtime = describeRunningSession(scopedSession);
            const logDefaultTailNote = defaultTailNote(totalLines, window.usingDefaultTail);
            return {
              content: [
                {
                  type: "text",
                  text:
                    (slice || "(no output yet)") +
                    logDefaultTailNote +
                    retentionCapNote(scopedSession) +
                    buildInputWaitHint(runtime),
                },
              ],
              details: {
                status: scopedSession.exited ? "completed" : "running",
                sessionId: params.sessionId,
                total: totalLines,
                totalLines,
                totalChars,
                truncated: scopedSession.truncated,
                name: deriveSessionName(scopedSession.command),
                ...runningSessionInputDetails(runtime),
              },
            };
          }
          if (scopedFinished) {
            const window = resolveLogSliceWindow(params.offset, params.limit);
            const { slice, totalLines, totalChars } = sliceLogLines(
              scopedFinished.aggregated,
              window.effectiveOffset,
              window.effectiveLimit,
            );
            const status = scopedFinished.status === "completed" ? "completed" : "failed";
            const logDefaultTailNote = defaultTailNote(totalLines, window.usingDefaultTail);
            return {
              content: [
                {
                  type: "text",
                  text:
                    (slice || "(no output recorded)") +
                    logDefaultTailNote +
                    retentionCapNote(scopedFinished),
                },
              ],
              details: {
                status,
                sessionId: params.sessionId,
                total: totalLines,
                totalLines,
                totalChars,
                truncated: scopedFinished.truncated,
                exitCode: scopedFinished.exitCode ?? undefined,
                exitSignal: scopedFinished.exitSignal ?? undefined,
                name: deriveSessionName(scopedFinished.command),
              },
            };
          }
          return {
            content: [
              {
                type: "text",
                text: `No session found for ${params.sessionId}`,
              },
            ],
            details: { status: "failed" },
          };
        }

        case "write": {
          const resolved = resolveBackgroundedWritableStdin();
          if (!resolved.ok) {
            return resolved.result;
          }
          await writeProcessStdin(resolved.stdin, params.data ?? "");
          if (params.eof) {
            resolved.stdin.end();
          }
          return runningSessionResult(
            resolved.session,
            `Wrote ${Buffer.byteLength(params.data ?? "", "utf8")} bytes to session ${params.sessionId}${
              params.eof ? " (stdin closed)" : ""
            }.`,
          );
        }

        case "send-keys": {
          const resolved = resolveBackgroundedWritableStdin();
          if (!resolved.ok) {
            return resolved.result;
          }
          return await handleProcessSendKeys({
            sessionId: params.sessionId,
            session: resolved.session,
            stdin: resolved.stdin,
            keys: params.keys,
            hex: params.hex,
            literal: params.literal,
          });
        }

        case "submit": {
          const resolved = resolveBackgroundedWritableStdin();
          if (!resolved.ok) {
            return resolved.result;
          }
          await writeProcessStdin(resolved.stdin, "\r");
          return runningSessionResult(
            resolved.session,
            `Submitted session ${params.sessionId} (sent CR).`,
          );
        }

        case "paste": {
          const resolved = resolveBackgroundedWritableStdin();
          if (!resolved.ok) {
            return resolved.result;
          }
          const payload = encodePaste(params.text ?? "", params.bracketed !== false);
          if (!payload) {
            return {
              content: [
                {
                  type: "text",
                  text: "No paste text provided.",
                },
              ],
              details: { status: "failed" },
            };
          }
          await writeProcessStdin(resolved.stdin, payload);
          return runningSessionResult(
            resolved.session,
            `Pasted ${params.text?.length ?? 0} chars to session ${params.sessionId}.`,
          );
        }

        case "kill": {
          if (!scopedSession) {
            return failText(`No active session found for ${params.sessionId}`);
          }
          if (!scopedSession.backgrounded) {
            return failText(`Session ${params.sessionId} is not backgrounded.`);
          }
          if (scopedSession.finalizing) {
            return failText(`Session ${params.sessionId} is finalizing.`);
          }
          if (!cancelBackgroundExecSession(scopedSession.id)) {
            return failText(
              `Unable to terminate session ${params.sessionId}: no active supervisor cancellation handle. Use process poll to check whether it is already exiting.`,
            );
          }
          resetPollRetrySuggestion(params.sessionId);
          return {
            content: [
              {
                type: "text",
                text: `Termination requested for session ${params.sessionId}.`,
              },
            ],
            details: {
              status: "failed",
              name: scopedSession ? deriveSessionName(scopedSession.command) : undefined,
            },
          };
        }

        case "clear": {
          if (scopedFinished) {
            resetPollRetrySuggestion(params.sessionId);
            deleteSession(params.sessionId);
            return {
              content: [{ type: "text", text: `Cleared session ${params.sessionId}.` }],
              details: { status: "completed" },
            };
          }
          return {
            content: [
              {
                type: "text",
                text: `No finished session found for ${params.sessionId}`,
              },
            ],
            details: { status: "failed" },
          };
        }

        case "remove": {
          if (scopedSession) {
            if (!scopedSession.backgrounded) {
              return failText(`Session ${params.sessionId} is not backgrounded.`);
            }
            if (scopedSession.finalizing) {
              return failText(`Session ${params.sessionId} is finalizing.`);
            }
            if (!cancelBackgroundExecSession(scopedSession.id)) {
              return failText(
                `Unable to remove session ${params.sessionId}: no active supervisor cancellation handle. Use process poll to check whether it is already exiting.`,
              );
            }
            // Keep remove semantics deterministic: drop from process registry now.
            scopedSession.backgrounded = false;
            deleteSession(params.sessionId);
            resetPollRetrySuggestion(params.sessionId);
            return {
              content: [
                {
                  type: "text",
                  text: `Removed session ${params.sessionId} (termination requested).`,
                },
              ],
              details: {
                status: "failed",
                name: scopedSession ? deriveSessionName(scopedSession.command) : undefined,
              },
            };
          }
          if (scopedFinished) {
            resetPollRetrySuggestion(params.sessionId);
            deleteSession(params.sessionId);
            return {
              content: [{ type: "text", text: `Removed session ${params.sessionId}.` }],
              details: { status: "completed" },
            };
          }
          return {
            content: [
              {
                type: "text",
                text: `No session found for ${params.sessionId}`,
              },
            ],
            details: { status: "failed" },
          };
        }
      }

      return {
        content: [{ type: "text", text: `Unknown action ${params.action as string}` }],
        details: { status: "failed" },
      };
    },
  };
}

/** Shared process-control tool instance used by the default Bash tool barrel. */
export const processTool = createProcessTool();
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */