diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 488cd95..b323015 100644 --- a/.github/workflows/ci.yml +++ b/.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. diff --git a/package.json b/package.json index e1737c3..afeb9bd 100644 --- a/package.json +++ b/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", diff --git a/scripts/medir-stream.mjs b/scripts/medir-stream.mjs new file mode 100644 index 0000000..e7b19ca --- /dev/null +++ b/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); diff --git a/server/package.json b/server/package.json index 8d20e73..b1a59ff 100644 --- a/server/package.json +++ b/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", diff --git a/server/src/agent/loop.ts b/server/src/agent/loop.ts index 99b1234..ea2f461 100644 --- a/server/src/agent/loop.ts +++ b/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 ). 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 => 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; type BloqueDeImagen = Extract; +type SalidaDeTool = { resultado: ToolResult; imagen?: BloqueDeImagen }; + +/** Espera las tools, o deja de esperarlas si interrumpen (ver arriba). */ +async function conInterrupcion( + corriendo: Promise[], + toolUses: Extract[], + signal: AbortSignal | undefined, +): Promise { + 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((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, 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, @@ -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); diff --git a/server/src/agent/providers/anthropic.ts b/server/src/agent/providers/anthropic.ts index 5da2faa..9a9565c 100644 --- a/server/src/agent/providers/anthropic.ts +++ b/server/src/agent/providers/anthropic.ts @@ -8,6 +8,7 @@ import { type StopReason, ProviderError, } from "./types.js"; +import { rutaEnJsonParcial } from "./json-parcial.js"; const API_URL = "https://api.anthropic.com/v1/messages"; const ANTHROPIC_VERSION = "2023-06-01"; @@ -46,6 +47,19 @@ const DEFAULT_MAX_TOKENS = 32_000; // Retry: 429 (rate limit) y 529 (overloaded). Config estilo CCX. const RETRY = { maxRetries: 5, baseMs: 1000, capMs: 60_000 }; +/** + * Cuánto puede pasar sin que llegue NADA por el stream antes de darlo por muerto. + * + * `fetch` no tiene límite entre pedazo y pedazo: si la conexión se queda abierta + * pero ya no llega nada, `reader.read()` espera para siempre, y la sala se queda + * en "trabajando" hasta que alguien reinicia el server (pasó). Anthropic manda + * pings mientras el modelo piensa o escribe, así que un silencio de minuto y + * medio no es un modelo lento: es una conexión muerta. Se corta y se reintenta. + * + * Se lee en cada request para que la demo lo pueda bajar a un segundo. + */ +const silencioMaximoMs = () => Number(process.env.MULTI_STREAM_SILENCIO_MS) || 90_000; + /** * Proveedor Claude — cliente HTTP directo a la API de Mensajes de Anthropic. * Sin SDK. Parsea el stream SSE a mano y lo traduce a StreamEvent neutrales. @@ -71,6 +85,10 @@ export class AnthropicProvider implements ModelProvider { try { return await this.streamOnce(params, callbacks); } catch (err) { + // Alguien de la sala interrumpió: cortar la conexión no es una falla de + // red y no se reintenta. Antes se reintentaba cinco veces, y el "@agente" + // tardaba medio minuto en soltar al agente. + if (params.signal?.aborted) throw err; const pe = err instanceof ProviderError ? err : new ProviderError(String(err), "network", { cause: err }); if (!pe.retryable || attempt >= RETRY.maxRetries) throw pe; attempt++; @@ -87,32 +105,47 @@ export class AnthropicProvider implements ModelProvider { ): Promise> { const body = buildRequestBody(params, this.defaultModel); - let res: Response; + // Un corte propio, para poder soltar una conexión muda sin tocar la señal de + // la sala. Si la sala interrumpe, también corta este. + const corte = new AbortController(); + const interrumpir = () => corte.abort(); + if (params.signal?.aborted) corte.abort(); + params.signal?.addEventListener("abort", interrumpir, { once: true }); + const vigilante = vigilarSilencio(corte); + try { - res = await fetch(API_URL, { - method: "POST", - headers: { - "content-type": "application/json", - "x-api-key": this.apiKey, - "anthropic-version": ANTHROPIC_VERSION, - }, - body: JSON.stringify(body), - signal: params.signal, - }); - } catch (err) { - throw new ProviderError("fallo de red al llamar la API", "network", { cause: err }); - } + let res: Response; + try { + res = await fetch(API_URL, { + method: "POST", + headers: { + "content-type": "application/json", + "x-api-key": this.apiKey, + "anthropic-version": ANTHROPIC_VERSION, + }, + body: JSON.stringify(body), + signal: corte.signal, + }); + } catch (err) { + if (vigilante.mudo()) throw vigilante.error(); + throw new ProviderError("fallo de red al llamar la API", "network", { cause: err }); + } - if (!res.ok) throw await httpError(res); - if (!res.body) throw new ProviderError("respuesta sin body", "api", { status: res.status }); + if (!res.ok) throw await httpError(res); + if (!res.body) throw new ProviderError("respuesta sin body", "api", { status: res.status }); - return this.consumeStream(res.body, callbacks); + return await this.consumeStream(res.body, callbacks, vigilante); + } finally { + vigilante.parar(); + params.signal?.removeEventListener("abort", interrumpir); + } } /** Lee el stream SSE byte a byte, reensambla bloques, emite el evento `end`. */ private async consumeStream( body: ReadableStream, callbacks: StreamCallbacks, + vigilante: ReturnType, ): Promise> { const reader = body.getReader(); const decoder = new TextDecoder(); @@ -125,6 +158,7 @@ export class AnthropicProvider implements ModelProvider { for (;;) { const { done, value } = await reader.read(); if (done) break; + vigilante.latido(); buffer += decoder.decode(value, { stream: true }); // Los eventos SSE se separan por línea en blanco (\n\n). @@ -136,6 +170,7 @@ export class AnthropicProvider implements ModelProvider { } } } catch (err) { + if (vigilante.mudo()) throw vigilante.error(); throw new ProviderError("error leyendo el stream", "network", { cause: err }); } finally { reader.releaseLock(); @@ -153,6 +188,8 @@ export class AnthropicProvider implements ModelProvider { message, usage: acc.usage, herramientaCortada: acc.herramientaCortada, + archivoCortado: acc.archivoCortado, + inputsInvalidos: acc.inputsInvalidos.length ? acc.inputsInvalidos : undefined, }; } @@ -228,13 +265,22 @@ class MessageAccumulator { stopReason: StopReason | null = null; /** Si `buildMessage` descartó un tool_use que el tope dejó a medias. */ herramientaCortada = false; + /** La ruta del archivo que el tope dejó a medias, si se alcanzó a leer. */ + archivoCortado: string | undefined; + /** + * Los tool_use cuyo input llegó completo pero no es JSON válido. Con + * `eager_input_streaming` la API ya no valida el JSON antes de mandarlo, y la + * doc avisa que puede pasar: se le contesta a esa tool con un error en vez de + * tumbar el turno entero. + */ + inputsInvalidos: string[] = []; usage: Extract["usage"]; // Bloques por índice, en construcción. private blocks = new Map< number, | { type: "text"; text: string } - | { type: "tool_use"; id: string; name: string; jsonBuf: string } + | { type: "tool_use"; id: string; name: string; jsonBuf: string; anunciada?: boolean } | { type: "thinking"; text: string } /** * Lo que corre del lado de Anthropic (ver `crudo` en types.ts). Los @@ -284,6 +330,16 @@ class MessageAccumulator { if (block.type === "tool_use" || block.type === "crudo") { block.jsonBuf += delta.partial_json ?? ""; } + // En cuanto se sabe a qué archivo va, se anuncia: escribir uno grande + // tarda minutos. La ruta va al principio, así que solo se busca en los + // primeros caracteres (y no se reescanea un archivo entero en cada delta). + if (block.type === "tool_use" && !block.anunciada && block.jsonBuf.length < 2_000) { + const path = rutaEnJsonParcial(block.jsonBuf); + if (path) { + block.anunciada = true; + callbacks.onToolPreparing?.({ id: block.id, name: block.name, input: { path } }); + } + } break; case "thinking_delta": if (block.type === "thinking") { @@ -351,6 +407,15 @@ class MessageAccumulator { // AVISO_SIN_ESPACIO en loop.ts). Antes esto culpaba a los créditos. if (this.stopReason === "max_tokens") { this.herramientaCortada = true; + this.archivoCortado ??= rutaEnJsonParcial(raw); + continue; + } + // El modelo terminó de escribirlo y pidió correrlo, pero no parsea. + // Va con input vacío (el historial necesita uno) y el loop le + // contesta con un error para que lo repita. + if (this.stopReason === "tool_use") { + this.inputsInvalidos.push(b.id); + content.push({ type: "tool_use", id: b.id, name: b.name, input: {} }); continue; } // Casi siempre es un stream que se cortó a media generación: el JSON @@ -409,6 +474,12 @@ function buildRequestBody(params: StreamParams, defaultModel: string): Record { // ── Utilidades de retry ───────────────────────────────────────────────────── +/** + * El timer de `silencioMaximoMs`: se rearma con cada pedazo que llega (pings + * incluidos) y, si vence, corta la conexión. + */ +function vigilarSilencio(corte: AbortController) { + const limite = silencioMaximoMs(); + const inicio = Date.now(); + let vencio = false; + let timer: ReturnType; + const armar = () => { + clearTimeout(timer); + timer = setTimeout(() => { + vencio = true; + console.warn( + `[agente] el stream lleva ${Math.round(limite / 1000)}s sin datos (a los ${Math.round((Date.now() - inicio) / 1000)}s del request) — se corta y se reintenta`, + ); + corte.abort(); + }, limite); + }; + armar(); + return { + latido: armar, + parar: () => clearTimeout(timer), + mudo: () => vencio, + error: () => new ProviderError(`el stream se quedó ${Math.round(limite / 1000)}s sin datos`, "network"), + }; +} + function backoff(attempt: number): number { // Exponencial con tope y jitter. const exp = Math.min(RETRY.baseMs * 2 ** (attempt - 1), RETRY.capMs); diff --git a/server/src/agent/providers/json-parcial.ts b/server/src/agent/providers/json-parcial.ts new file mode 100644 index 0000000..f2bd18c --- /dev/null +++ b/server/src/agent/providers/json-parcial.ts @@ -0,0 +1,17 @@ +/** + * La ruta de un `write_file`/`edit_file` sacada del JSON a medio llegar. + * + * Los argumentos de una herramienta llegan en pedazos. La ruta va primero y es + * cortita, así que aparece mucho antes que el contenido: con ella la sala puede + * decir "Escribiendo juego.js" desde el principio, y el aviso de "no cupo" puede + * decir cuál archivo fue. + */ +export function rutaEnJsonParcial(json: string): string | undefined { + const m = /"path"\s*:\s*"((?:[^"\\]|\\.)*)"/.exec(json); + if (!m) return undefined; + try { + return JSON.parse(`"${m[1]}"`) as string; + } catch { + return undefined; + } +} diff --git a/server/src/agent/providers/openai-compat.ts b/server/src/agent/providers/openai-compat.ts index f1dae27..78a5fb5 100644 --- a/server/src/agent/providers/openai-compat.ts +++ b/server/src/agent/providers/openai-compat.ts @@ -8,6 +8,7 @@ import { type StopReason, ProviderError, } from "./types.js"; +import { rutaEnJsonParcial } from "./json-parcial.js"; /** * Proveedor para cualquier API que hable el formato de OpenAI. @@ -247,10 +248,12 @@ export class OpenAICompatProvider implements ModelProvider { // raro) y el turno se marca como sin espacio, igual que en anthropic.ts. const sinEspacio = finishReason === "length"; let herramientaCortada = false; + let archivoCortado: string | undefined; for (const [i, tc] of [...toolCalls.entries()].sort((a, b) => a[0] - b[0])) { if (sinEspacio && !jsonCompleto(tc.args)) { herramientaCortada = true; + archivoCortado ??= rutaEnJsonParcial(tc.args); continue; } content.push({ @@ -277,6 +280,7 @@ export class OpenAICompatProvider implements ModelProvider { message: { role: "assistant", content }, usage, herramientaCortada, + archivoCortado, }; } } diff --git a/server/src/agent/providers/types.ts b/server/src/agent/providers/types.ts index a948fa5..838564d 100644 --- a/server/src/agent/providers/types.ts +++ b/server/src/agent/providers/types.ts @@ -95,6 +95,10 @@ export type StreamEvent = * acabó pensando": lo primero sí se arregla con más espacio. */ herramientaCortada?: boolean; + /** La ruta del archivo que se cortó, si se alcanzó a leer. Va en el aviso. */ + archivoCortado?: string; + /** Ids de tool_use cuyo input no era JSON válido: se contestan con error. */ + inputsInvalidos?: string[]; }; export type StopReason = @@ -130,6 +134,12 @@ export interface StreamCallbacks { * leer una página). No hay nada que ejecutar: es para enseñarlo en la sala. */ onServerTool?: (info: { id: string; name: string; input: Record }) => void; + /** + * Una herramienta empezó a escribirse y ya se sabe a qué archivo va, aunque + * sus argumentos sigan llegando. Un archivo grande tarda minutos en llegar + * completo: sin esto la sala no enseñaba nada en todo ese rato. + */ + onToolPreparing?: (info: { id: string; name: string; input: Record }) => void; } // ── La interfaz del proveedor ────────────────────────────────────────────── diff --git a/server/src/agent/tools/bash.ts b/server/src/agent/tools/bash.ts index 8aba1fe..cf96f4d 100644 --- a/server/src/agent/tools/bash.ts +++ b/server/src/agent/tools/bash.ts @@ -1,6 +1,12 @@ import { type Tool, ToolError, reqString } from "./base.js"; const DEFAULT_TIMEOUT_MS = 120_000; +/** + * Lo más que puede pedir el modelo. Sin tope, un `timeout_ms` enorme sobre un + * comando que no termina (un dev server, un watch) dejaba el turno esperando + * horas con la sala en "trabajando". + */ +const MAX_TIMEOUT_MS = 10 * 60_000; const MAX_OUTPUT = 30_000; // truncar salidas enormes para no reventar el contexto /** @@ -21,14 +27,15 @@ export const bashTool: Tool = { type: "object", properties: { command: { type: "string", description: "Comando de shell a ejecutar" }, - timeout_ms: { type: "number", description: "Timeout en ms (default 120000)" }, + timeout_ms: { type: "number", description: "Timeout en ms (default 120000, máximo 600000)" }, }, required: ["command"], }, }, async run(input, ctx) { const command = reqString(input, "command"); - const timeoutMs = typeof input.timeout_ms === "number" ? input.timeout_ms : DEFAULT_TIMEOUT_MS; + const pedido = typeof input.timeout_ms === "number" && input.timeout_ms > 0 ? input.timeout_ms : DEFAULT_TIMEOUT_MS; + const timeoutMs = Math.min(pedido, MAX_TIMEOUT_MS); ctx.emit?.({ type: "tool:bash", command }); diff --git a/server/src/demos/sin-espacio.ts b/server/src/demos/sin-espacio.ts index a1646dc..c97c941 100644 --- a/server/src/demos/sin-espacio.ts +++ b/server/src/demos/sin-espacio.ts @@ -1,7 +1,7 @@ import { mkdtemp, readFile, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { runAgent, AVISO_SIN_ESPACIO, TOPE_REINTENTO } from "../agent/loop.js"; +import { runAgent, AVISO_SIN_ESPACIO, AVISOS_SIN_ESPACIO, avisoDeArchivoCortado } from "../agent/loop.js"; import { localRunner } from "../engine/runner.js"; import { AnthropicProvider } from "../agent/providers/anthropic.js"; import type { ContentBlock, Message, ModelProvider, StopReason, StreamEvent } from "../agent/providers/types.js"; @@ -32,7 +32,7 @@ function check(name: string, ok: boolean, detail = ""): void { } } -type Vuelta = { stop: StopReason; content: ContentBlock[]; herramientaCortada?: boolean }; +type Vuelta = { stop: StopReason; content: ContentBlock[]; herramientaCortada?: boolean; archivoCortado?: string }; /** Proveedor falso: una respuesta por vuelta, en orden. Guarda lo que se le mandó. */ function proveedor( @@ -56,6 +56,7 @@ function proveedor( message: { role: "assistant", content: v.content }, usage: { inputTokens: 0, outputTokens: v.stop === "max_tokens" ? 32000 : 10 }, herramientaCortada: v.herramientaCortada, + archivoCortado: v.archivoCortado, }; }, }; @@ -69,9 +70,9 @@ function hayAssistantVacio(msgs: Message[]): boolean { return msgs.some((m) => m.role === "assistant" && m.content.length === 0); } -function llevaElAviso(msgs: Message[]): boolean { +function llevaElAviso(msgs: Message[], aviso = AVISO_SIN_ESPACIO): boolean { const ultimo = msgs[msgs.length - 1]; - return ultimo?.role === "user" && ultimo.content.some((c) => c.type === "text" && c.text === AVISO_SIN_ESPACIO); + return ultimo?.role === "user" && ultimo.content.some((c) => c.type === "text" && c.text === aviso); } /** Cada tool_use del historial tiene su tool_result. Es lo que la API exige. */ @@ -135,11 +136,12 @@ async function main() { check("ni en lo que se le mandó al modelo", !llamadas.some(hayAssistantVacio)); } - console.log("\n2. Se lo gasta dos veces: se rinde y lo dice, sin un tercer intento"); + const intentos = AVISOS_SIN_ESPACIO + 1; + console.log(`\n2. Se lo gasta ${intentos} veces: se rinde y lo dice, sin otro intento`); { - const { provider, llamadas } = proveedor([VACIO, VACIO, listo("no debería llegar aquí")]); + const { provider, llamadas } = proveedor([...Array(intentos).fill(VACIO), listo("no debería llegar aquí")]); const r = await runAgent({ ...base, provider }); - check("solo dos llamadas (no se queman otros 32k)", llamadas.length === 2, `${llamadas.length} llamadas`); + check(`solo ${intentos} llamadas (no se queman otros 32k)`, llamadas.length === intentos, `${llamadas.length} llamadas`); check("sale marcado como sin espacio", r.sinEspacio === true); check("el texto viene vacío (lo pone index.ts)", r.finalText === "", r.finalText); check("el historial no tiene ningún assistant vacío", !hayAssistantVacio(r.messages)); @@ -164,21 +166,19 @@ async function main() { check("termina normal", r.finalText === "Listo, ya hay App.tsx." && !r.sinEspacio, r.finalText); } - console.log("\n4. Si lo que se cortó fue una herramienta, el reintento lleva más espacio"); + console.log("\n4. Si lo que no cupo fue un archivo, se le dice cuál y que lo parta (sin subir el tope)"); { - const cortada: Vuelta = { stop: "max_tokens", content: [], herramientaCortada: true }; + const cortada: Vuelta = { stop: "max_tokens", content: [], herramientaCortada: true, archivoCortado: "src/juego.js" }; const a = proveedor([cortada, listo("ya")], "anthropic"); await runAgent({ ...base, provider: a.provider }); - check(`con Anthropic el reintento pide ${TOPE_REINTENTO}`, a.topes[1] === TOPE_REINTENTO, String(a.topes[1])); - check("y el primer intento no", a.topes[0] === undefined, String(a.topes[0])); - - const pensando = proveedor([VACIO, listo("ya")], "anthropic"); - await runAgent({ ...base, provider: pensando.provider }); - check("si se lo gastó pensando, el tope no sube", pensando.topes[1] === undefined, String(pensando.topes[1])); - - const otro = proveedor([cortada, listo("ya")], "openrouter"); - await runAgent({ ...base, provider: otro.provider }); - check("con otro proveedor no se sube (su techo varía por modelo)", otro.topes[1] === undefined, String(otro.topes[1])); + const aviso = avisoDeArchivoCortado("src/juego.js"); + check("el aviso nombra el archivo", llevaElAviso(a.llamadas[1] ?? [], aviso) && aviso.includes("src/juego.js")); + check("y le pide partirlo", aviso.includes("pártelo")); + check("el reintento NO sube el tope (eran otros 48k tirados)", a.topes[1] === undefined, String(a.topes[1])); + + const sinRuta = proveedor([{ ...cortada, archivoCortado: undefined }, listo("ya")], "anthropic"); + await runAgent({ ...base, provider: sinRuta.provider }); + check("sin ruta también avisa", llevaElAviso(sinRuta.llamadas[1] ?? [], avisoDeArchivoCortado(undefined))); } console.log("\n5. Anthropic: un tool_use truncado por el tope no es un error"); @@ -191,16 +191,16 @@ async function main() { check("stop_reason max_tokens", fin.stopReason === "max_tokens", fin.stopReason); check("el bloque truncado se descarta", !fin.message.content.some((c) => c.type === "tool_use")); check("y avisa que fue una herramienta", fin.herramientaCortada === true); + check("con la ruta del archivo", fin.archivoCortado === "src/App.tsx", String(fin.archivoCortado)); check("el texto de antes se queda", fin.message.content.some((c) => c.type === "text" && c.text === "Va el archivo.")); + // Con eager_input_streaming la API ya no valida el JSON: si el modelo + // pidió correr la tool y no parsea, se le contesta con error a esa tool. globalThis.fetch = fetchConSse(sseConToolTruncado("tool_use")); - let lanzo = ""; - try { - await new AnthropicProvider("sk-falsa").stream({ messages: [] }); - } catch (err) { - lanzo = String(err); - } - check("con otra stop_reason sigue siendo un error", lanzo.includes("se cortó a media escritura"), lanzo.slice(0, 80)); + const invalido = await new AnthropicProvider("sk-falsa").stream({ messages: [] }); + const tu = invalido.message.content.find((c) => c.type === "tool_use"); + check("con stop tool_use no lanza: la tool va con input vacío", tu?.type === "tool_use" && Object.keys(tu.input).length === 0); + check("y queda marcada como inválida", invalido.inputsInvalidos?.includes("tu_1") === true, JSON.stringify(invalido.inputsInvalidos)); } finally { globalThis.fetch = original; } diff --git a/server/src/demos/stream-vivo.ts b/server/src/demos/stream-vivo.ts new file mode 100644 index 0000000..a9e4183 --- /dev/null +++ b/server/src/demos/stream-vivo.ts @@ -0,0 +1,328 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { runAgent, TOOL_INTERRUMPIDA } from "../agent/loop.js"; +import { localRunner, type Runner } from "../engine/runner.js"; +import { AnthropicProvider } from "../agent/providers/anthropic.js"; +import { bashTool } from "../agent/tools/bash.js"; +import type { ContentBlock, Message, ModelProvider, StreamEvent, StopReason } from "../agent/providers/types.js"; +import type { ToolContext } from "../agent/tools/base.js"; + +/** + * Demo: el agente no se queda mudo ni colgado. + * Uso: npm run demo:stream-vivo + * + * Lo que la motivó, en una sala real: al pedir "rehazlo en 3D" el agente quiso + * escribir todo el juego en un archivo. Sin `eager_input_streaming` Anthropic + * junta el input completo antes de mandarlo, así que fueron minutos sin un solo + * byte: la sala en "trabajando" sin enseñar nada, y un "@agente" que no lo + * soltaba. Se terminó reiniciando el server. + * + * Lo que se revisa: + * 1. Las tools van con `eager_input_streaming` (las de web no). + * 2. La sala sabe qué archivo se está escribiendo antes de que termine. + * 3. Un stream que se queda callado se corta y se reintenta solo. + * 4. Interrumpir corta al instante, sin reintentos. + * 5. Interrumpir no espera a una tool lenta, y el historial queda válido. + * 6. Un input que no parsea se contesta con error, sin tumbar el turno. + * 7. bash no acepta un timeout de horas. + * + * Sin red ni API key: `fetch` falso que manda el SSE a mano, a su ritmo. + */ + +let pass = 0; +let fail = 0; +function check(name: string, ok: boolean, detail = ""): void { + if (ok) { + pass++; + console.log(` [ok] ${name}`); + } else { + fail++; + console.log(` [X] ${name} ${detail}`); + } +} + +const enc = new TextEncoder(); +const sse = (eventos: object[]) => eventos.map((e) => `event: x\ndata: ${JSON.stringify(e)}\n\n`).join(""); +const espera = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +const INICIO = { type: "message_start", message: { usage: { input_tokens: 4, output_tokens: 0 } } }; +const FIN = (stop: StopReason) => [ + { type: "message_delta", delta: { stop_reason: stop }, usage: { output_tokens: 10 } }, + { type: "message_stop" }, +]; +const RESPUESTA_CORTA = [ + INICIO, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "hola" } }, + { type: "content_block_stop", index: 0 }, + ...FIN("end_turn"), +]; + +type Pedazo = { despuesDe: number; texto: string }; + +/** + * Un `fetch` falso que manda los pedazos con pausas, y con `callarse` deja la + * conexión abierta sin mandar nada más. Respeta el abort como el de verdad. + */ +function fetchFalso(guiones: { pedazos: Pedazo[]; callarse?: boolean }[], registro: { bodies: any[] }): typeof fetch { + return (async (_url: unknown, init?: RequestInit) => { + const signal = init?.signal ?? undefined; + if (signal?.aborted) throw new DOMException("aborted", "AbortError"); + registro.bodies.push(JSON.parse(String(init?.body))); + const guion = guiones[Math.min(registro.bodies.length - 1, guiones.length - 1)]; + const body = new ReadableStream({ + async start(c) { + let cortado = false; + signal?.addEventListener("abort", () => { + cortado = true; + try { + c.error(new DOMException("aborted", "AbortError")); + } catch { + // Ya estaba cerrado. + } + }); + for (const p of guion.pedazos) { + await espera(p.despuesDe); + if (cortado) return; + c.enqueue(enc.encode(p.texto)); + } + if (!guion.callarse && !cortado) c.close(); + }, + }); + return new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }); + }) as typeof fetch; +} + +/** Proveedor falso con guion, para las pruebas del loop. */ +function proveedor( + vueltas: { stop: StopReason; content: ContentBlock[]; inputsInvalidos?: string[] }[], +): { provider: ModelProvider; llamadas: Message[][] } { + const llamadas: Message[][] = []; + const provider: ModelProvider = { + name: "falso", + defaultModel: "falso/modelo", + async stream(params): Promise> { + llamadas.push(JSON.parse(JSON.stringify(params.messages))); + const v = vueltas[llamadas.length - 1]; + if (!v) throw new Error(`el guion no tiene vuelta ${llamadas.length - 1}`); + return { + type: "end", + stopReason: v.stop, + message: { role: "assistant", content: v.content }, + usage: { inputTokens: 0, outputTokens: 10 }, + inputsInvalidos: v.inputsInvalidos, + }; + }, + }; + return { provider, llamadas }; +} + +function toolsEmparejadas(messages: Message[]): boolean { + const usados = new Set(); + const resueltos = new Set(); + for (const m of messages) { + for (const c of m.content) { + if (c.type === "tool_use") usados.add(c.id); + if (c.type === "tool_result") resueltos.add(c.tool_use_id); + } + } + for (const id of usados) if (!resueltos.has(id)) return false; + return true; +} + +function resultadoDe(messages: Message[], id: string): string | undefined { + for (const m of messages) { + for (const c of m.content) { + if (c.type === "tool_result" && c.tool_use_id === id) return typeof c.content === "string" ? c.content : JSON.stringify(c.content); + } + } + return undefined; +} + +const TOOL_WRITE = { + name: "write_file", + description: "escribe", + input_schema: { type: "object" as const, properties: { path: { type: "string" }, content: { type: "string" } } }, +}; + +async function main() { + console.log("=== El agente no se queda mudo ni colgado ===\n"); + const original = globalThis.fetch; + const dir = await mkdtemp(join(tmpdir(), "multi-stream-vivo-")); + + try { + console.log("1. Las tools van con eager_input_streaming"); + { + const registro = { bodies: [] as any[] }; + globalThis.fetch = fetchFalso([{ pedazos: [{ despuesDe: 0, texto: sse(RESPUESTA_CORTA) }] }], registro); + await new AnthropicProvider("sk-falsa").stream({ messages: [], tools: [TOOL_WRITE], model: "claude-sonnet-5-5" }); + const tools = registro.bodies[0]?.tools ?? []; + const nuestra = tools.find((t: any) => t.name === "write_file"); + const web = tools.filter((t: any) => t.name !== "write_file"); + check("write_file lleva eager_input_streaming", nuestra?.eager_input_streaming === true, JSON.stringify(nuestra)); + check("las de internet no (no son nuestras)", web.length > 0 && web.every((t: any) => t.eager_input_streaming === undefined)); + } + + console.log("\n2. La sala sabe qué archivo se escribe antes de que termine"); + { + const registro = { bodies: [] as any[] }; + const t0 = Date.now(); + globalThis.fetch = fetchFalso( + [ + { + pedazos: [ + { + despuesDe: 0, + texto: sse([ + INICIO, + { type: "content_block_start", index: 0, content_block: { type: "tool_use", id: "tu_1", name: "write_file" } }, + { type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: '{"path": "src/jue' } }, + { type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: 'go.js", "content": "// mucho código' } }, + ]), + }, + { + despuesDe: 400, + texto: sse([ + { type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: ' más"}' } }, + { type: "content_block_stop", index: 0 }, + ...FIN("tool_use"), + ]), + }, + ], + }, + ], + registro, + ); + const anuncios: { path: unknown; ms: number }[] = []; + const fin = await new AnthropicProvider("sk-falsa").stream( + { messages: [], tools: [TOOL_WRITE] }, + { onToolPreparing: ({ input }) => anuncios.push({ path: input.path, ms: Date.now() - t0 }) }, + ); + const total = Date.now() - t0; + check("se anuncia una sola vez", anuncios.length === 1, JSON.stringify(anuncios)); + check("con la ruta completa", anuncios[0]?.path === "src/juego.js", String(anuncios[0]?.path)); + check("mucho antes de que termine el archivo", (anuncios[0]?.ms ?? total) < 200 && total >= 400, `${anuncios[0]?.ms}ms de ${total}ms`); + const tu = fin.message.content.find((c) => c.type === "tool_use"); + check("y el input completo llega igual", tu?.type === "tool_use" && tu.input.content === "// mucho código más"); + } + + console.log("\n3. Un stream que se queda callado se corta y se reintenta solo"); + { + process.env.MULTI_STREAM_SILENCIO_MS = "300"; + const registro = { bodies: [] as any[] }; + globalThis.fetch = fetchFalso( + [ + { pedazos: [{ despuesDe: 0, texto: sse([INICIO]) }], callarse: true }, + { pedazos: [{ despuesDe: 0, texto: sse(RESPUESTA_CORTA) }] }, + ], + registro, + ); + const reintentos: string[] = []; + const t0 = Date.now(); + const fin = await new AnthropicProvider("sk-falsa").stream( + { messages: [] }, + { onRetry: ({ reason }) => reintentos.push(reason) }, + ); + check("se reintentó una vez", registro.bodies.length === 2 && reintentos.length === 1, `${registro.bodies.length} requests`); + check("como falla de red", reintentos[0] === "network", reintentos[0]); + check("y el segundo sí contestó", fin.stopReason === "end_turn"); + check("sin quedarse esperando para siempre", Date.now() - t0 < 5_000, `${Date.now() - t0}ms`); + delete process.env.MULTI_STREAM_SILENCIO_MS; + } + + console.log("\n4. Interrumpir corta al instante, sin reintentos"); + { + const registro = { bodies: [] as any[] }; + globalThis.fetch = fetchFalso([{ pedazos: [{ despuesDe: 0, texto: sse([INICIO]) }], callarse: true }], registro); + const corte = new AbortController(); + setTimeout(() => corte.abort(), 100); + const t0 = Date.now(); + let error = ""; + try { + await new AnthropicProvider("sk-falsa").stream({ messages: [], signal: corte.signal }); + } catch (err) { + error = String(err); + } + const tardo = Date.now() - t0; + check("lanza", error.length > 0); + check("en menos de un segundo", tardo < 1_000, `${tardo}ms`); + check("sin volver a llamar a la API", registro.bodies.length === 1, `${registro.bodies.length} requests`); + } + + console.log("\n5. Interrumpir no espera a una tool lenta"); + { + const { provider } = proveedor([ + { + stop: "tool_use", + content: [ + { type: "tool_use", id: "tu_lenta", name: "bash", input: { command: "sleep 5" } }, + { type: "tool_use", id: "tu_rapida", name: "bash", input: { command: "echo listo" } }, + ], + }, + ]); + const corte = new AbortController(); + setTimeout(() => corte.abort(), 500); + const t0 = Date.now(); + const r = await runAgent({ + provider, + workspaceDir: dir, + runner: localRunner(dir), + messages: [], + userMessage: "corre esto", + signal: corte.signal, + }); + const tardo = Date.now() - t0; + check("el turno se suelta casi al instante", tardo < 2_000, `${tardo}ms`); + check("sale marcado como interrumpido", r.interrumpido === true); + check("cada tool_use tiene su tool_result", toolsEmparejadas(r.messages)); + check("la lenta dice que la interrumpieron", resultadoDe(r.messages, "tu_lenta") === TOOL_INTERRUMPIDA, resultadoDe(r.messages, "tu_lenta")); + check("la rápida conserva su resultado", resultadoDe(r.messages, "tu_rapida")?.includes("listo") === true, resultadoDe(r.messages, "tu_rapida")); + } + + console.log("\n6. Un input que no parsea se contesta con error y el turno sigue"); + { + const { provider, llamadas } = proveedor([ + { + stop: "tool_use", + content: [{ type: "tool_use", id: "tu_rota", name: "write_file", input: {} }], + inputsInvalidos: ["tu_rota"], + }, + { stop: "end_turn", content: [{ type: "text", text: "ya lo repetí" }] }, + ]); + const r = await runAgent({ provider, workspaceDir: dir, runner: localRunner(dir), messages: [], userMessage: "escribe" }); + const resultado = resultadoDe(r.messages, "tu_rota") ?? ""; + check("no se ejecutó con el input vacío", resultado.includes("no llegaron como JSON válido"), resultado); + check("el modelo recibió el error y siguió", llamadas.length === 2 && r.finalText === "ya lo repetí", r.finalText); + check("historial válido", toolsEmparejadas(r.messages)); + } + + console.log("\n7. bash no acepta un timeout de horas"); + { + let pedido = 0; + const runner: Runner = { + isolated: false, + async exec(_cmd, opts) { + pedido = opts.timeoutMs; + return { stdout: "", stderr: "", code: 0, timedOut: false }; + }, + } as Runner; + const ctx = { workspaceDir: dir, runner } as ToolContext; + await bashTool.run({ command: "npm run dev", timeout_ms: 10 * 60 * 60_000 }, ctx); + check("se topa en 10 minutos", pedido === 600_000, String(pedido)); + await bashTool.run({ command: "ls" }, ctx); + check("sin pedir nada, el default de siempre", pedido === 120_000, String(pedido)); + } + } finally { + globalThis.fetch = original; + await rm(dir, { recursive: true, force: true }); + } + + console.log(`\n${pass} pasaron, ${fail} fallaron`); + process.exit(fail > 0 ? 1 : 0); +} + +main().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/server/src/engine/container.ts b/server/src/engine/container.ts index 26cedd3..b7d1ae1 100644 --- a/server/src/engine/container.ts +++ b/server/src/engine/container.ts @@ -236,6 +236,10 @@ export async function startContainer( "--cpus", CPU_LIMIT, // Sin privilegios extra y sin poder ganarlos (un sudo dentro no escala). "--security-opt", "no-new-privileges", + // Un init de verdad como PID 1. El proceso principal es `sleep infinity`, + // que no recoge a los hijos que mueren: cada captura de ver_preview dejaba + // sus Chromium como zombies () hasta que se recreaba el contenedor. + "--init", // Si el server de Multi muere, el contenedor no se queda huérfano para siempre. "--label", "multi.room=" + roomId, IMAGE_TAG, diff --git a/server/src/index.ts b/server/src/index.ts index fbd9075..fdb91df 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -2857,6 +2857,16 @@ async function runAgentTurn( io.to(room.id).emit("agent:tool", { agentId, name: b.name, accion, summary: summarizeTool(b.name, b.input) }); } }, + // Un archivo grande tarda minutos en llegar completo. En cuanto se sabe + // a cuál va, la sala lo dice con la misma línea de actividad; cuando + // llega completo, onToolStart la repite con lo demás. + onToolPreparing: ({ name, input }) => { + room.agents.touch(agentId); + const accion = accionDeTool(name, input); + if (!ultimaAccion.has(room.id)) ultimaAccion.set(room.id, new Map()); + ultimaAccion.get(room.id)!.set(agentId, accion); + io.to(room.id).emit("agent:tool", { agentId, name, accion, summary: summarizeTool(name, input) }); + }, onToolStart: ({ name, input }) => { room.agents.touch(agentId); const accion = accionDeTool(name, input);