Paralelizar cenas com subagentes sem estourar o limite das APIs

// scripts/pipeline-cenas.ts: o diretor delega três cenas a especialistas, em paralelo.
// Produz especificações JSON; não gera mídia. Execute na raiz video-ia: npx tsx scripts/pipeline-cenas.ts
import { query } from "@anthropic-ai/claude-agent-sdk";

const TETO_TOTAL_USD = 3;   // política do exercício, somando todas as tentativas
const MAX_TENTATIVAS = 2;   // a segunda só acontece se a primeira morrer por limite (429)
const ESPERA_BASE_MS = 5_000;

const PEDIDO = [
  "Prepare três cenas independentes de 5 segundos para um vídeo de receita em pt-BR:",
  "abertura, preparo e fechamento.",
  "Delegue cada cena ao especialista de mesmo nome, em paralelo.",
  "Cada especialista devolve JSON com id, duracao_s, texto_tela e descricao_visual.",
  "Não crie outros agentes. Reúna as três respostas num array JSON sem reescrevê-las.",
  "Se algum especialista falhar, diga qual e pare.",
].join(" ");

// Um especialista por cena; sem ferramentas, ele só pensa e devolve texto.
const especialista = (cena: string, modelo: "sonnet" | "haiku") => ({
  description: `Cria a cena ${cena} de cinco segundos.`,
  prompt:
    `Retorne apenas o JSON da cena ${cena}: id "${cena}", duracao_s 5, ` +
    "texto_tela (até 6 palavras) e descricao_visual (até 40 palavras). " +
    "Não invente medidas culinárias. Não use ferramentas nem delegue.",
  tools: [] as string[],
  model: modelo,
  maxTurns: 3,
});

type Desfecho = "ok" | "limite" | "erro";

async function executarUmaVez(tentativa: number): Promise<Desfecho> {
  const inicio = Date.now();
  let desfecho: Desfecho = "erro";
  try {
    for await (const msg of query({
      prompt: PEDIDO,
      options: {
        model: "claude-opus-5-5",
        tools: ["Agent"],        // disponibilidade: o diretor só enxerga a ferramenta Agent
        allowedTools: ["Agent"], // permissão: aprova a delegação sem perguntar
        maxBudgetUsd: TETO_TOTAL_USD / MAX_TENTATIVAS, // US$ 1,50 por tentativa
        env: {
          ...process.env,
          CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS: "3",
          CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH: "1",
        },
        agents: {
          abertura: especialista("abertura", "sonnet"),
          preparo: especialista("preparo", "sonnet"),
          fechamento: especialista("fechamento", "haiku"),
        },
      },
    })) {
      // Registra tipo e tempo de cada evento, sem imprimir prompt nem credencial.
      const s = ((Date.now() - inicio) / 1000).toFixed(1);
      console.log(`[tentativa ${tentativa} | ${s}s] evento: ${msg.type}`);
      if (msg.type === "rate_limit_event") {
        console.log("Aviso de limite da sessão:", msg.rate_limit_info.status);
      }
      if (msg.type === "result") {
        console.log("Estado:", msg.subtype, "| erro:", msg.is_error, "| custo informado (US$):", msg.total_cost_usd);
        if (msg.subtype === "success" && !msg.is_error) {
          console.log(msg.result);
          desfecho = "ok";
        } else if (msg.subtype === "success" && msg.api_error_status === 429) {
          desfecho = "limite"; // a última chamada ao modelo terminou em 429
        }
        // Sem return: o laço segue até o fim para a exceção posterior chegar ao catch.
      }
    }
  } catch (erro) {
    // Depois de um resultado de erro, o query() de mensagem única lança de propósito.
    console.warn("query() encerrou com erro:", erro instanceof Error ? erro.message : String(erro));
  }
  return desfecho;
}

async function main(): Promise<void> {
  if (!process.env.ANTHROPIC_API_KEY) throw new Error("Defina ANTHROPIC_API_KEY no ambiente.");
  for (let tentativa = 1; tentativa <= MAX_TENTATIVAS; tentativa++) {
    const desfecho = await executarUmaVez(tentativa);
    if (desfecho === "ok") return;
    // Qualquer outro desfecho para na hora: repetir às cegas só multiplica a conta.
    if (desfecho === "erro" || tentativa === MAX_TENTATIVAS) {
      throw new Error(`Execução sem resultado válido na tentativa ${tentativa}.`);
    }
    const espera = ESPERA_BASE_MS * 2 ** (tentativa - 1) + Math.random() * 1_000;
    console.warn(`Limite de requisições; nova tentativa em ${(espera / 1000).toFixed(1)} s.`);
    await new Promise((r) => setTimeout(r, espera));
  }
}

main().catch((erro) => {
  console.error("Execução interrompida:", erro instanceof Error ? erro.message : String(erro));
  process.exitCode = 1;
});

"""scripts/pipeline_cenas.py: o diretor delega três cenas a especialistas, em paralelo.

Produz especificações JSON; não gera mídia. Execute na raiz video-ia: python3 scripts/pipeline_cenas.py
"""
import asyncio
import os
import random

from claude_agent_sdk import (
    AgentDefinition,
    ClaudeAgentOptions,
    RateLimitEvent,
    ResultMessage,
    query,
)

TETO_TOTAL_USD = 3.0   # política do exercício, somando todas as tentativas
MAX_TENTATIVAS = 2     # a segunda só acontece se a primeira morrer por limite (429)
ESPERA_BASE_S = 5.0

PEDIDO = (
    "Prepare três cenas independentes de 5 segundos para um vídeo de receita em pt-BR: "
    "abertura, preparo e fechamento. Delegue cada cena ao especialista de mesmo nome, "
    "em paralelo. Cada especialista devolve JSON com id, duracao_s, texto_tela e "
    "descricao_visual. Não crie outros agentes. Reúna as três respostas num array JSON "
    "sem reescrevê-las. Se algum especialista falhar, diga qual e pare."
)


def especialista(cena: str, modelo: str) -> AgentDefinition:
    """Um especialista por cena; sem ferramentas, ele só pensa e devolve texto."""
    return AgentDefinition(
        description=f"Cria a cena {cena} de cinco segundos.",
        prompt=(
            f'Retorne apenas o JSON da cena {cena}: id "{cena}", duracao_s 5, '
            "texto_tela (até 6 palavras) e descricao_visual (até 40 palavras). "
            "Não invente medidas culinárias. Não use ferramentas nem delegue."
        ),
        tools=[],
        model=modelo,
    )


def opcoes() -> ClaudeAgentOptions:
    return ClaudeAgentOptions(
        model="claude-opus-5-5",
        tools=["Agent"],           # disponibilidade: o diretor só enxerga a ferramenta Agent
        allowed_tools=["Agent"],   # permissão: aprova a delegação sem perguntar
        max_budget_usd=TETO_TOTAL_USD / MAX_TENTATIVAS,   # US$ 1,50 por tentativa
        env={
            "CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS": "3",
            "CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH": "1",
        },
        agents={
            "abertura": especialista("abertura", "sonnet"),
            "preparo": especialista("preparo", "sonnet"),
            "fechamento": especialista("fechamento", "haiku"),
        },
    )


async def executar_uma_vez(tentativa: int) -> str:
    """Devolve "ok", "limite" (a execução terminou em 429) ou "erro"."""
    desfecho = "erro"
    try:
        async for msg in query(prompt=PEDIDO, options=opcoes()):
            print(f"[tentativa {tentativa}] evento: {type(msg).__name__}")
            if isinstance(msg, RateLimitEvent):
                print("Aviso de limite da sessão:", msg.rate_limit_info.status)
            if isinstance(msg, ResultMessage):
                print("Estado:", msg.subtype, "| erro:", msg.is_error,
                      "| custo informado (US$):", msg.total_cost_usd)
                if msg.subtype == "success" and not msg.is_error:
                    print(msg.result)
                    desfecho = "ok"
                elif msg.subtype == "success" and msg.api_error_status == 429:
                    desfecho = "limite"   # a última chamada ao modelo terminou em 429
                # Sem return: o laço segue até o fim para a exceção posterior aparecer.
    except Exception as erro:
        # Depois de um resultado de erro, o query() de mensagem única levanta de propósito.
        print(f"query() encerrou com erro: {erro}")
    return desfecho


async def main() -> None:
    if not os.environ.get("ANTHROPIC_API_KEY"):
        raise SystemExit("Defina ANTHROPIC_API_KEY no ambiente.")
    for tentativa in range(1, MAX_TENTATIVAS + 1):
        desfecho = await executar_uma_vez(tentativa)
        if desfecho == "ok":
            return
        # Qualquer outro desfecho para na hora: repetir às cegas só multiplica a conta.
        if desfecho == "erro" or tentativa == MAX_TENTATIVAS:
            raise SystemExit(f"Execução sem resultado válido na tentativa {tentativa}.")
        espera = ESPERA_BASE_S * 2 ** (tentativa - 1) + random.uniform(0, 1)
        print(f"Limite de requisições; nova tentativa em {espera:.1f} s.")
        await asyncio.sleep(espera)


if __name__ == "__main__":
    asyncio.run(main())

"""scripts/despachar_cenas.py: limita a concorrência, espaça os inícios e espera em 429.

Simula um serviço externo (voz, imagem ou vídeo) sem gastar crédito.
Execute na raiz video-ia: python3 scripts/despachar_cenas.py
"""
import asyncio
import random
import time

CONCORRENCIA = 3      # trabalhos ativos ao mesmo tempo (política do exercício)
INTERVALO_S = 0.5     # espaço mínimo entre dois inícios (política do exercício, não cota)
MAX_TENTATIVAS = 4
ESPERA_BASE_S = 1.0
ESPERA_MAX_S = 20.0
INICIO = time.monotonic()

# Falhas de mentira para o teste: (cena, tentativa) -> espera pedida pelo serviço, em s.
FALHAS_SIMULADAS = {("preparo", 1): 1.0}


class Erro429(Exception):
    """Resposta Too Many Requests; retry_after é a espera que o serviço pediu, se pediu."""

    def __init__(self, retry_after=None):
        super().__init__("429 Too Many Requests")
        self.retry_after = retry_after


class Espacador:
    """Garante INTERVALO_S entre dois inícios, venham de qual cena vierem."""

    def __init__(self, intervalo):
        self.intervalo = intervalo
        self.proxima = 0.0
        self.lock = asyncio.Lock()

    async def esperar(self):
        async with self.lock:
            agora = time.monotonic()
            if self.proxima > agora:
                await asyncio.sleep(self.proxima - agora)
            self.proxima = time.monotonic() + self.intervalo


def espera_backoff(tentativa, retry_after):
    """Obedece ao retry_after do serviço; sem ele, dobra a espera e soma um sorteio."""
    if retry_after is not None:
        return float(retry_after)
    base = min(ESPERA_MAX_S, ESPERA_BASE_S * 2 ** (tentativa - 1))
    return base + random.uniform(0, base / 2)   # jitter: as cenas não voltam juntas


def agora():
    return f"{time.monotonic() - INICIO:5.2f}s"


async def chamar_servico(cena, tentativa):
    """Troque este corpo pela chamada real quando a cota da sua conta estiver anotada.

    Um 429 recusa o pedido antes do trabalho e pode ser repetido. Um timeout depois
    de enviar uma geração paga deixa dúvida: consulte o job conhecido antes de repetir.
    """
    print(f"{agora()}  início  {cena} (tentativa {tentativa})")
    await asyncio.sleep(1.2)   # duração simulada do trabalho
    if (cena, tentativa) in FALHAS_SIMULADAS:
        raise Erro429(retry_after=FALHAS_SIMULADAS[(cena, tentativa)])
    return {"cena": cena, "arquivo": f"storyboard/cenas/{cena}.json", "tentativas": tentativa}


async def executar(cena, semaforo, espacador, estado):
    for tentativa in range(1, MAX_TENTATIVAS + 1):
        try:
            async with semaforo:            # vaga entre as CONCORRENCIA disponíveis
                await espacador.esperar()   # e o espaço mínimo entre inícios
                estado["ativos"] += 1
                estado["pico"] = max(estado["pico"], estado["ativos"])
                try:
                    return await chamar_servico(cena, tentativa)
                finally:
                    estado["ativos"] -= 1
        except Erro429 as erro:
            if tentativa == MAX_TENTATIVAS:
                raise
            espera = espera_backoff(tentativa, erro.retry_after)
            # A espera acontece fora do semáforo: a vaga fica livre para outra cena.
            print(f"{agora()}  429     {cena}: aguardando {espera:.1f}s")
            await asyncio.sleep(espera)


async def main():
    semaforo = asyncio.Semaphore(CONCORRENCIA)
    espacador = Espacador(INTERVALO_S)
    estado = {"ativos": 0, "pico": 0}
    cenas = ["abertura", "preparo", "fechamento"]
    resultados = await asyncio.gather(
        *(executar(c, semaforo, espacador, estado) for c in cenas),
        return_exceptions=True,   # uma cena com falha não derruba as outras
    )
    for cena, r in zip(cenas, resultados):
        print(f"{cena:<11} {'erro: ' + str(r) if isinstance(r, Exception) else 'ok'}")
    assert estado["pico"] <= CONCORRENCIA, "concorrência acima do limite"
    print("Pico simultâneo:", estado["pico"])


if __name__ == "__main__":
    asyncio.run(main())

---
name: cenarista-haiku
description: Versão barata do cenarista para cenas curtas com briefing completo. Use quando roteiro e duração já estiverem aprovados.
tools: Read
model: haiku
effort: low
---
Mesma entrada, mesma entrega e mesmas regras do cenarista, em português do Brasil
com acentuação completa: devolva só o objeto JSON com id, duracao_s, texto_tela
(até 6 palavras) e descricao_visual (até 40 palavras).

- Não invente medidas, não gere mídia, não escreva arquivos e não delegue.
- Se o briefing estiver incompleto, devolva {"erro": "motivo"} e pare.

---
name: pesquisador-de-modelos
description: Pesquisa e cota modelos de vídeo na fal para uma cena aprovada. Use quando a cena precisa de b-roll gerado e ainda não há modelo escolhido.
tools: Read, mcp__fal__search_models, mcp__fal__get_model_schema, mcp__fal__get_pricing
mcpServers:
  - fal
model: haiku
effort: low
maxTurns: 4
background: true
color: orange
---
Você cota modelos para UMA cena, em português do Brasil com acentuação completa.

Entrada: storyboard/cenas/<cena>.json, que você pode ler.
Saída: tabela com fornecedor, ID exato do modelo, duração, preço e unidade de cobrança,
no máximo cinco linhas, chamando get_pricing uma vez com todos os IDs.

- Não gere mídia e não peça para gerar: quem decide é a pessoa.
- Descarte preview com data de saída anunciada.
- Se faltar preço, escreva [conferir].

"""Hook SubagentStop do cenarista. Saída 2 devolve o stderr ao subagente como próxima instrução."""
import json
import sys

CAMPOS = {"id", "duracao_s", "texto_tela", "descricao_visual"}

dados = json.load(sys.stdin)
if dados.get("stop_hook_active"):
    sys.exit(0)  # já foi devolvido uma vez: deixa o diretor decidir, sem laço

texto = (dados.get("last_assistant_message") or "").strip()
try:
    cena = json.loads(texto)
except ValueError:
    print("Devolva só o objeto JSON da cena, sem texto em volta nem cercas Markdown.", file=sys.stderr)
    sys.exit(2)

if isinstance(cena, dict) and "erro" in cena:
    sys.exit(0)  # o próprio contrato permite devolver {"erro": "motivo"}
faltando = CAMPOS - set(cena) if isinstance(cena, dict) else CAMPOS
if faltando:
    print(f"Faltam os campos {sorted(faltando)}. Devolva o JSON completo.", file=sys.stderr)
    sys.exit(2)
if len(str(cena["texto_tela"]).split()) > 6:
    print("texto_tela passou de 6 palavras. Encurte e devolva o JSON.", file=sys.stderr)
    sys.exit(2)
sys.exit(0)

{
  "hooks": {
    "SubagentStop": [
      {
        "matcher": "cenarista|cenarista-haiku",
        "hooks": [
          {
            "type": "command",
            "command": "python3 \"$CLAUDE_PROJECT_DIR/.claude/hooks/validar_cena.py\"",
            "timeout": 10
          }
        ]
      }
    ]
  }
}