El agente ya no se queda mudo ni colgado con archivos grandes

.github/workflows/ci.yml

@@ -53,6 +53,7 @@ jobs:
       - run: npm run demo:preview-caido
       - run: npm run demo:ver-codigo
       - run: npm run demo:sin-espacio
+      - run: npm run demo:stream-vivo
       - run: npm run demo:mensaje-creditos
       # El navegador con el que el agente mira su vista previa. En las salas
       # viene en la imagen; aquí, el de Playwright.

package.json

@@ -58,6 +58,7 @@
     "demo:preview-caido": "npm run demo:preview-caido -w server",
     "demo:ver-codigo": "npm run demo:ver-codigo -w server",
     "demo:sin-espacio": "npm run demo:sin-espacio -w server",
+    "demo:stream-vivo": "npm run demo:stream-vivo -w server",
     "demo:mensaje-creditos": "npm run demo:mensaje-creditos -w server",
     "demo:ver-preview": "npm run demo:ver-preview -w server",
     "demo:internet": "npm run demo:internet -w server",

scripts/medir-stream.mjs

@@ -0,0 +1,108 @@
+// Mide cuánto tarda en llegar el input de una tool grande, sin y con eager_input_streaming.
+// Uso (desde server/, para leer su .env): node ../scripts/medir-stream.mjs [modelo]
+import { readFileSync } from "node:fs";
+
+const env = readFileSync(".env", "utf8");
+const KEY = env.match(/^(?:MULTI_API_KEY|ANTHROPIC_API_KEY)=(.*)$/m)?.[1]?.trim().replace(/^["']|["']$/g, "");
+if (!KEY) throw new Error("no encontré MULTI_API_KEY ni ANTHROPIC_API_KEY en .env");
+const MODELO = process.argv[2] ?? "claude-sonnet-5-5";
+
+const PEDIDO =
+  "Escribe con write_file el archivo src/juego.js: un juego de carreras completo en canvas, " +
+  "de unas 900 líneas, con pista, kart, 3 rivales con IA, física simple, HUD y menú. " +
+  "Todo en ese único archivo, de un jalón. No expliques nada, solo llama write_file.";
+
+async function medir(eager) {
+  const tool = {
+    name: "write_file",
+    description: "Escribe un archivo completo.",
+    input_schema: {
+      type: "object",
+      properties: { path: { type: "string" }, content: { type: "string" } },
+      required: ["path", "content"],
+    },
+    ...(eager ? { eager_input_streaming: true } : {}),
+  };
+  const t0 = Date.now();
+  const res = await fetch("https://api.anthropic.com/v1/messages", {
+    method: "POST",
+    headers: { "content-type": "application/json", "x-api-key": KEY, "anthropic-version": "2023-06-01" },
+    body: JSON.stringify({
+      model: MODELO,
+      max_tokens: 32000,
+      stream: true,
+      tools: [tool],
+      messages: [{ role: "user", content: PEDIDO }],
+    }),
+  });
+  if (!res.ok) throw new Error(`HTTP ${res.status}: ${await res.text()}`);
+
+  const seg = () => ((Date.now() - t0) / 1000).toFixed(1);
+  let buffer = "";
+  let ultimoReal = Date.now(); // último evento que no fue ping
+  let huecoMax = 0;
+  let primerInput = null;
+  let pedazos = 0;
+  let pings = 0;
+  let salida = 0;
+  let textoYPensar = 0;
+  let bytesInput = 0;
+  let ultimoPing = null;
+  // Cada 10 s una línea, para ver en vivo si llega algo o si la conexión está muda.
+  const latido = setInterval(() => {
+    const desdeAlgo = ((Date.now() - ultimoReal) / 1000).toFixed(0);
+    const desdePing = ultimoPing ? ((Date.now() - ultimoPing) / 1000).toFixed(0) + "s" : "nunca";
+    console.log(`  ${seg()}s  pedazos ${pedazos} (${bytesInput} chars)  texto/pensar ${textoYPensar}  pings ${pings}  sin datos hace ${desdeAlgo}s  último ping hace ${desdePing}`);
+  }, 10_000);
+  const tope = setTimeout(() => {
+    console.log(`  ${seg()}s  15 min sin terminar: corto`);
+    reader.cancel().catch(() => {});
+  }, 15 * 60_000);
+  console.log(`\n${eager ? "CON" : "SIN"} eager_input_streaming (respuesta HTTP a los ${seg()}s)`);
+  const dec = new TextDecoder();
+  const reader = res.body.getReader();
+  for (;;) {
+    const { done, value } = await reader.read().catch((e) => {
+      console.log(`  ${seg()}s  la lectura falló: ${e}`);
+      return { done: true };
+    });
+    if (done) break;
+    buffer += dec.decode(value, { stream: true });
+    let i;
+    while ((i = buffer.indexOf("\n\n")) !== -1) {
+      const bloque = buffer.slice(0, i);
+      buffer = buffer.slice(i + 2);
+      const data = bloque.split("\n").find((l) => l.startsWith("data: "));
+      if (!data) continue;
+      const ev = JSON.parse(data.slice(6));
+      if (ev.type === "ping") {
+        pings++;
+        ultimoPing = Date.now();
+        continue;
+      }
+      const ahora = Date.now();
+      huecoMax = Math.max(huecoMax, ahora - ultimoReal);
+      ultimoReal = ahora;
+      if (ev.type === "content_block_delta" && ev.delta?.type === "input_json_delta") {
+        pedazos++;
+        bytesInput += ev.delta.partial_json.length;
+        if (primerInput === null) primerInput = seg();
+      }
+      if (ev.type === "content_block_delta" && ev.delta?.type !== "input_json_delta") textoYPensar++;
+      if (ev.type === "content_block_start") console.log(`  ${seg()}s  empieza bloque ${ev.content_block.type}${ev.content_block.name ? " " + ev.content_block.name : ""}`);
+      if (ev.type === "message_delta") salida = ev.usage?.output_tokens ?? salida;
+      if (ev.type === "error") console.log("  error del stream:", JSON.stringify(ev.error));
+    }
+  }
+  clearInterval(latido);
+  clearTimeout(tope);
+  console.log(`  resultado ${eager ? "CON" : "SIN"} eager:`);
+  console.log(`  primer pedazo del archivo: a los ${primerInput}s`);
+  console.log(`  pedazos del archivo:       ${pedazos}`);
+  console.log(`  silencio más largo:        ${(huecoMax / 1000).toFixed(1)}s (solo pings: ${pings})`);
+  console.log(`  total:                     ${seg()}s, ${salida} tokens de salida`);
+}
+
+console.log(`Modelo: ${MODELO}`);
+await medir(false);
+await medir(true);

server/package.json

@@ -49,6 +49,7 @@
     "demo:preview-caido": "tsx src/demos/preview-caido.ts",
     "demo:ver-codigo": "tsx src/demos/ver-codigo.ts",
     "demo:sin-espacio": "tsx src/demos/sin-espacio.ts",
+    "demo:stream-vivo": "tsx src/demos/stream-vivo.ts",
     "demo:mensaje-creditos": "tsx src/demos/mensaje-creditos.ts",
     "demo:ver-preview": "tsx src/demos/ver-preview.ts",
     "demo:internet": "tsx src/demos/internet.ts",

server/src/agent/loop.ts

@@ -48,6 +48,14 @@ son las únicas que dejan rastro para el historial de la sala y para el aviso en
 así que los cambios de contenido pasan por ahí. Deja bash para lo que es proceso —
 instalar dependencias, git, builds, comandos del framework.
 
+Escribe en archivos chicos, uno por pieza (en un juego: pista, jugador, rivales,
+cámara, controles, HUD, main), y nunca toda la app en un solo archivo. Vale igual para
+un proyecto nuevo que para rehacer o reorganizar uno que ya existe ("pásalo a 3D",
+"cámbiale el motor"): escribe las piezas una por una, no todo de un jalón. Si un archivo
+ya es largo y hay que cambiar mucho, pártelo en módulos.
+Por qué: tu respuesta tiene un tope de largo. Un archivo enorme no cabe, se corta a la
+mitad y se pierde entero, y mientras lo escribes la sala no ve nada durante minutos.
+
 Bash te devuelve el exit code y la salida: LÉELOS. Un comando que falló y das por
 bueno te deja construyendo sobre nada, y quien está en la sala ve una pantalla vacía
 sin saber por qué. Si algo no salió como esperabas, arréglalo antes de seguir.
@@ -104,8 +112,7 @@ no preguntes por dónde empezar.
     3. Los archivos de config del stack y el punto de entrada.
     4. Ya con eso, el código que te pidieron. En proyectos grandes (juegos, páginas
        con mucho contenido) empieza por lo mínimo que funcione y crece en archivos
-       chicos: nunca escribas toda la app en un solo archivo. Una respuesta tuya
-       tiene un tope de largo, y un archivo enorme no cabe.
+       chicos (ver <uso_de_tools>).
   Mira antes qué hay: si un paso ya está hecho, sáltatelo y sigue desde donde se
   quedó. Un turno anterior pudo haberse cortado a la mitad.
   El orden importa porque hasta que no hay manifiesto no hay instalación, sin
@@ -537,21 +544,34 @@ export interface RunResult {
  *
  * Reintentar igual daría lo mismo. Lo que cambia el resultado es pedirle que
  * escriba ya y en pedazos chicos.
- */
-/**
- * El tope de salida del reintento cuando lo que se cortó fue una herramienta.
  *
- * Es lo que recomienda Anthropic para un `tool_use` cortado por `max_tokens`:
- * repetir con más espacio. Solo en ese caso: si se lo gastó pensando, más
- * espacio solo sería pensar más (y pagar más) para terminar igual. Y solo con
- * Anthropic, donde se sabe que el modelo da para esto; en otros proveedores el
- * techo de salida varía por modelo y pedir de más es un 400.
+ * Antes, si lo cortado era un archivo, primero se reintentaba con más tope
+ * (48k). En una sala real eso fue el mismo archivo gigante otra vez: 32k y luego
+ * 48k de salida tirados, unos quince minutos con la sala en "trabajando" sin
+ * enseñar nada, y al final igual no cupo. Ahora se le dice de una qué archivo
+ * no cupo y que lo parta.
  */
-export const TOPE_REINTENTO = 48_000;
-
 export const AVISO_SIN_ESPACIO =
   "Te quedaste sin espacio antes de terminar tu respuesta. No le des más vueltas: escribe ya, archivo por archivo, empezando por lo mínimo que haga funcionar la app. Ningún archivo debe ser enorme: divide en componentes chicos.";
 
+/** El aviso cuando lo que no cupo fue un archivo: se le dice cuál. */
+export function avisoDeArchivoCortado(ruta: string | undefined): string {
+  const cual = ruta ? `\`${ruta}\`` : "El archivo que estabas escribiendo";
+  return (
+    `${cual} no cupo en tu respuesta y no se guardó nada de él. No lo vuelvas a escribir entero: ` +
+    "pártelo en varios archivos chicos (un módulo por pieza) y escríbelos uno por uno."
+  );
+}
+
+/** Cuántas veces por turno se le avisa que se quedó sin espacio antes de rendirse. */
+export const AVISOS_SIN_ESPACIO = 2;
+
+/** Una tool que tarde más que esto queda en el log, para saber dónde se fue el tiempo. */
+const TOOL_LENTA_MS = 30_000;
+
+/** Lo que recibe una tool que seguía corriendo cuando interrumpieron al agente. */
+export const TOOL_INTERRUMPIDA = "interrumpido: alguien de la sala te pidió otra cosa antes de que esto terminara";
+
 /**
  * EL loop del agente. Blueprint: doc oficial de Anthropic + ~/ccx-rs.
  *
@@ -621,8 +641,7 @@ export async function runAgent(opts: {
   // se guarda y se reenvía en cada turno siguiente, y una imagen ahí se pagaría
   // cada vez. El texto de lo que se vio sí se queda.
   const onProgreso = (m: Message[]) => opts.onProgreso?.(paraGuardar(m));
-  // Puede subir una vez si el tope cortó una herramienta (ver TOPE_REINTENTO).
-  let maxTokens = opts.maxTokens;
+  const maxTokens = opts.maxTokens;
 
   const messages: Message[] = [
     ...opts.messages,
@@ -650,8 +669,8 @@ export async function runAgent(opts: {
 
   let finalText = "";
   let turn = 0;
-  /** Si ya se le avisó en este turno que se quedó sin espacio. */
-  let yaSeAviso = false;
+  /** Cuántas veces se le avisó en este turno que se quedó sin espacio. */
+  let avisos = 0;
   /** Cuántas veces Anthropic pausó este turno a media búsqueda (ver pause_turn). */
   let pausas = 0;
   /** Si la vuelta anterior terminó en pausa: lo que llegue sigue el mismo mensaje. */
@@ -735,9 +754,20 @@ export async function runAgent(opts: {
       (c): c is Extract<ContentBlock, { type: "tool_use" }> => c.type === "tool_use",
     );
 
-    // Ejecutar TODAS en paralelo. Cada una produce un tool_result.
-    const salidas = await Promise.all(
-      toolUses.map((tu) => executeTool(tu, toolCtx, callbacks, opts.ve === true)),
+    // Ejecutar TODAS en paralelo. Cada una produce un tool_result. Si alguien
+    // interrumpe mientras corren, no se espera a que acaben: una tool lenta
+    // dejaba al "@agente" sin efecto hasta que terminara (o para siempre). Las
+    // que seguían reciben un resultado de interrumpidas para que cada tool_use
+    // tenga el suyo; el proceso de adentro se muere con su propio timeout.
+    const invalidos = new Set(end.inputsInvalidos ?? []);
+    const salidas = await conInterrupcion(
+      toolUses.map((tu) =>
+        invalidos.has(tu.id)
+          ? Promise.resolve(inputInvalido(tu, callbacks))
+          : executeTool(tu, toolCtx, callbacks, opts.ve === true),
+      ),
+      toolUses,
+      signal,
     );
     const results = salidas.map((s) => s.resultado);
     // Las capturas de ver_preview van como bloques `image` normales, DESPUÉS de
@@ -766,8 +796,8 @@ export async function runAgent(opts: {
     // Sin espacio: las tools que alcanzaron a salir completas ya corrieron
     // arriba, así que lo que haya escrito no se pierde y cada tool_use tiene su
     // resultado. Lo que falta decidir es si se le da otra oportunidad.
-    if (sinEspacio && yaSeAviso) {
-      // Segunda vez en el mismo turno: otro intento serían otros 32k tirados.
+    if (sinEspacio && avisos >= AVISOS_SIN_ESPACIO) {
+      // Ya se le dijo y sigue sin caber: otro intento serían otros 32k tirados.
       console.warn("[agente] sin espacio (max_tokens) otra vez — se rinde");
       onProgreso(messages);
       return { finalText, turns: turn + 1, messages: paraGuardar(messages), sinEspacio: true };
@@ -776,21 +806,18 @@ export async function runAgent(opts: {
       // Primera vez: se le dice qué hacer distinto y se sigue. Va pegado al
       // último mensaje del usuario (el de los resultados, o el pedido mismo si
       // no hubo tools) para no dejar dos mensajes de usuario seguidos.
-      yaSeAviso = true;
-      const masEspacio =
-        end.herramientaCortada && provider.name === "anthropic" && (maxTokens ?? 32_000) < TOPE_REINTENTO;
-      if (masEspacio) maxTokens = TOPE_REINTENTO;
+      avisos++;
+      const aviso = end.herramientaCortada ? avisoDeArchivoCortado(end.archivoCortado) : AVISO_SIN_ESPACIO;
       console.warn(
-        `[agente] sin espacio (max_tokens) — reintento con aviso${masEspacio ? ` y tope de ${TOPE_REINTENTO}` : ""}`,
+        `[agente] sin espacio (max_tokens)${end.herramientaCortada ? ` escribiendo ${end.archivoCortado ?? "un archivo"}` : ""} — aviso ${avisos} de ${AVISOS_SIN_ESPACIO}`,
       );
       const ultimo = messages[messages.length - 1];
       if (esperaAlServidor) {
-        // Ahí no cabe texto (ver arriba). Va sin aviso: el espacio
-        // extra ya es lo que más ayuda.
+        // Ahí no cabe texto (ver arriba). Va sin aviso.
       } else if (ultimo?.role === "user") {
-        ultimo.content = [...ultimo.content, { type: "text", text: AVISO_SIN_ESPACIO }];
+        ultimo.content = [...ultimo.content, { type: "text", text: aviso }];
       } else {
-        messages.push({ role: "user", content: [{ type: "text", text: AVISO_SIN_ESPACIO }] });
+        messages.push({ role: "user", content: [{ type: "text", text: aviso }] });
       }
     }
 
@@ -810,6 +837,47 @@ export async function runAgent(opts: {
 type ToolResult = Extract<ContentBlock, { type: "tool_result" }>;
 type BloqueDeImagen = Extract<ContentBlock, { type: "image" }>;
 
+type SalidaDeTool = { resultado: ToolResult; imagen?: BloqueDeImagen };
+
+/** Espera las tools, o deja de esperarlas si interrumpen (ver arriba). */
+async function conInterrupcion(
+  corriendo: Promise<SalidaDeTool>[],
+  toolUses: Extract<ContentBlock, { type: "tool_use" }>[],
+  signal: AbortSignal | undefined,
+): Promise<SalidaDeTool[]> {
+  if (!signal) return Promise.all(corriendo);
+  const listas: (SalidaDeTool | undefined)[] = new Array(corriendo.length);
+  corriendo.forEach((p, i) => void p.then((s) => (listas[i] = s)));
+  let alAbortar = () => {};
+  const abortada = new Promise<void>((resolve) => {
+    alAbortar = () => resolve();
+    if (signal.aborted) resolve();
+    else signal.addEventListener("abort", alAbortar, { once: true });
+  });
+  try {
+    const todas = await Promise.race([Promise.all(corriendo), abortada.then(() => null)]);
+    if (todas) return todas;
+  } finally {
+    signal.removeEventListener("abort", alAbortar);
+  }
+  // Se deja que terminen las que ya iban de salida (un microtask), y las demás
+  // se dan por interrumpidas.
+  await new Promise((r) => setTimeout(r, 0));
+  return toolUses.map(
+    (tu, i) =>
+      listas[i] ?? {
+        resultado: { type: "tool_result", tool_use_id: tu.id, content: TOOL_INTERRUMPIDA, is_error: true },
+      },
+  );
+}
+
+/** La respuesta a un tool_use cuyo input no llegó como JSON válido. */
+function inputInvalido(tu: Extract<ContentBlock, { type: "tool_use" }>, callbacks: AgentCallbacks): SalidaDeTool {
+  const result = `los argumentos de "${tu.name}" no llegaron como JSON válido y no se ejecutó nada. Vuelve a llamarla.`;
+  callbacks.onToolEnd?.({ id: tu.id, name: tu.name, result, isError: true });
+  return { resultado: { type: "tool_result", tool_use_id: tu.id, content: result, is_error: true } };
+}
+
 /** Ejecuta un tool_use (con gate de permisos) → tool_result, y la imagen si trajo una. */
 async function executeTool(
   tu: Extract<ContentBlock, { type: "tool_use" }>,
@@ -833,8 +901,11 @@ async function executeTool(
   const tool = toolRegistry.get(tu.name);
   if (!tool) return emit(`tool desconocida: ${tu.name}`, true);
 
+  const inicio = Date.now();
   try {
     const result = await tool.run(tu.input, ctx);
+    const tardo = Date.now() - inicio;
+    if (tardo > TOOL_LENTA_MS) console.warn(`[agente] ${tu.name} tardó ${Math.round(tardo / 1000)}s`);
     if (typeof result === "string") return emit(result, false);
     // A un modelo que no ve, una imagen le es un 400: le llega solo el texto.
     const salida = emit(result.texto, false);