automaton/plugins/automaton-guard-pi/guard.ts

/**
 * Automaton Guard — Pi Dev Harness Edition
 *
 * Pre-edit guard that calls `status.py --can-edit` before every file
 * modification. Blocks edits when no automaton task is in implement or
 * doc_review phase.
 *
 * Install:
 *   pi install ~/.automaton/plugins/automaton-guard-pi
 */

import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";

const STATUS_SCRIPT = `${process.env.HOME}/.automaton/scripts/status.py`;
const STALE_THRESHOLD_MINUTES = 30;

interface CanEditResult {
  allowed: boolean;
  reason: string;
  task?: string;
  stale_task?: string;
  stale_minutes?: number;
}

async function checkCanEdit(file: string): Promise<CanEditResult> {
  const cmd = [
    "python3",
    STATUS_SCRIPT,
    "--can-edit",
    `--project`,
    process.cwd(),
    `--file`,
    file,
    "--json",
  ];

  try {
    const { stdout, code } = await (await import("@earendil-works/pi-coding-agent")).default?.exec
      ? {} as any
      : { stdout: "", code: 1 };

    // Use child_process directly since pi.exec is only available on ExtensionAPI
    const { execSync } = await import("child_process");
    const output = execSync(cmd.join(" "), { encoding: "utf-8", timeout: 5000 });
    const lines = output.trim().split("\n");
    const jsonLine = lines[lines.length - 1];
    const result = JSON.parse(jsonLine);
    return {
      allowed: result.allowed,
      reason: result.reason,
      task: result.primary_task?.task,
      stale_task: result.stale_task,
      stale_minutes: result.stale_minutes,
    };
  } catch (e: any) {
    if (e.status !== undefined && e.status !== 0) {
      const text = (e.stdout || e.stderr || "").trim();
      const lines = text.split("\n");
      const jsonLine = lines[lines.length - 1];
      try {
        const result = JSON.parse(jsonLine);
        return {
          allowed: false,
          reason: result.reason,
          stale_task: result.stale_task,
          stale_minutes: result.stale_minutes,
        };
      } catch {
        return { allowed: false, reason: text || "Denied by automaton" };
      }
    }
    return { allowed: true, reason: "status.py not available, allowing edit" };
  }
}

export default function (pi: ExtensionAPI) {
  const EDIT_TOOLS = new Set(["edit", "write", "bash"]);

  pi.on("tool_call", async (event, ctx) => {
    if (!EDIT_TOOLS.has(event.toolName)) return undefined;

    const filePath =
      event.input?.file_path ||
      event.input?.filePath ||
      event.input?.path ||
      event.input?.target ||
      "";

    // For bash commands, skip unless it's a write/redirect operation
    if (event.toolName === "bash") {
      const cmd = event.input?.command || "";
      if (!/\b(write|tee|cp|mv|sed\b.*-i|dd\b.*of=)\b/.test(cmd)) {
        return undefined;
      }
    }

    if (!filePath) return undefined;

    const result = await checkCanEdit(filePath);

    if (result.allowed) return undefined;

    if (result.reason === "stale_task") {
      if (ctx.hasUI) {
        ctx.ui.notify(
          `BLOCKED: Task '${result.stale_task}' stale for ${result.stale_minutes ?? "?"} min. Create a new task or touch it.`,
          "warning",
        );
      }
      return {
        block: true,
        reason:
          `[AUTOMATON GUARD] Task '${result.stale_task || "unknown"}' has been in edit phase for ${result.stale_minutes || "?"} minutes (stale).\n\n` +
          `To continue: python ~/.automaton/scripts/status.py --touch --task ${result.stale_task || "<task>"}\n` +
          `Or create a new task: python ~/.automaton/scripts/status.py --create-task <name>`,
      };
    }

    const msg =
      result.reason === "no_edit_tasks"
        ? "No task in implement or doc_review phase. Create or transition a task first."
        : result.reason === "out_of_scope"
          ? "File is outside the project scope."
          : result.reason === "wrong_phase"
            ? "Current task is not in an edit-allowed phase."
            : `Edit denied: ${result.reason}`;

    if (ctx.hasUI) {
      ctx.ui.notify(`BLOCKED: ${msg}`, "warning");
    }

    return {
      block: true,
      reason:
        `[AUTOMATON GUARD] ${msg}\n\n` +
        `To proceed:\n` +
        `1. Create a task: python ~/.automaton/scripts/status.py --create-task <name>\n` +
        `2. Transition it: python ~/.automaton/scripts/status.py --transition implement --task <name>`,
    };
  });
}