Files
researchowl/tests/test_shortspec.py
T
ChemaVXandClaude Fable 5 cfe4a6d754
Build & Deploy ResearchOwl / build-and-push (push) Successful in 10s
feat(short): el ejemplo del prompt habla, y por eso los specs vuelven a hablar
La primera generación con narración viva narró 5 de 7 planos. La siguiente, ya
con la sección de presupuestos de ancho en el prompt, narró 1 de 8.

La causa no era la sección 3b sino la 5: el ejemplo trabajado — un spec completo
de 8 planos presentado como "el que produjo un buen vídeo" — no llevaba ni una
narración, porque es anterior a la voz. Contra esa señal de formato, una regla
en prosa que además dice "no todos los planos necesitan una" pierde siempre.

Ahora el ejemplo narra 6 de sus 8 y calla en los dos que dibujan una cita, donde
la voz sólo competiría con palabras que ya están en pantalla. Sus duraciones
suben para pagar lo que habla: un plano que se queda corto para su propia voz
enseñaría a infradeclarar, y el render no corta la voz, alarga el plano. La
ventana de silencio se recoloca porque seguía a document_quote por tiempo.

El aviso de ancho ya no cita el hueco de 16 caracteres de quote_a: shortsmith
envuelve esas citas en dos líneas y el presupuesto vivo es 31.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-06 21:32:49 +00:00

283 lines
10 KiB
Python

"""Escritura del shot spec: prompt, bucle de reintento y fallback.
El LLM entra como un callable, así que aquí se prueba el bucle, no a Haiku:
respuesta buena, respuesta malformada, typo en una prop, y las tres seguidas.
"""
import json
from pathlib import Path
import pytest
from src.generator.shortspec import (
MAX_ATTEMPTS, ShortSpecWriter, SpecWriteFailed, extract_json,
)
from tests.test_spec_contract import TEMPLATES
EXAMPLE = Path(__file__).resolve().parents[1] / "src/generator/examples/jal1628.json"
GOOD = {
"version": 1,
"meta": {"id": "caso", "title": "Un caso"},
"shots": [
{"template": "radar_sweep", "duration": 15.0, "props": {"headline": "3 RADARS"}},
{"template": "scale_bars", "duration": 15.0, "props": {
"headline": "REPORTED SCALE",
"bars": [{"label": "BOEING 747", "value": 232, "unit": "FT"}]}},
],
}
class FakeLLM:
"""Devuelve respuestas de una cola y guarda los prompts que recibió."""
def __init__(self, *responses):
self.responses = list(responses)
self.prompts: list[str] = []
self.systems: list[str] = []
async def __call__(self, system, prompt):
self.systems.append(system)
self.prompts.append(prompt)
return self.responses.pop(0) if len(self.responses) > 1 else self.responses[0]
def writer(*responses, **kw):
llm = FakeLLM(*responses)
return ShortSpecWriter(llm, TEMPLATES, **kw), llm
# --- parseo -----------------------------------------------------------------
def test_extract_json_survives_markdown_fences():
assert extract_json('```json\n{"a": 1}\n```') == {"a": 1}
assert extract_json('Here you go:\n{"a": 1}\nHope that helps') == {"a": 1}
assert extract_json('{"a": 1}') == {"a": 1}
def test_extract_json_complains_when_there_is_no_object():
with pytest.raises(ValueError):
extract_json("I'm afraid I can't do that")
def test_straight_quotes_inside_a_string_are_repaired():
"""Cómo falla esto en la vida real (sesión de Bélgica, 2026-08-01): el
modelo escribe la cita con comillas rectas, que cierran la cadena JSON antes
de tiempo. Se arregla aquí porque además es lo que se quiere dibujar."""
broken = '{"quote_a": ""CREDIBLE PEOPLE. THEY TOLD WHAT THEY SAW."", "n": 1}'
assert extract_json(broken) == {
"quote_a": "“CREDIBLE PEOPLE. THEY TOLD WHAT THEY SAW.”", "n": 1}
def test_the_repair_leaves_correct_json_alone():
good = {"a": 'texto con “tipográficas” dentro', "b": [1, 2], "c": {"d": "e"}}
assert extract_json(json.dumps(good, ensure_ascii=False)) == good
def test_the_repair_does_not_eat_escaped_quotes():
assert extract_json(r'{"a": "dijo \"hola\" y se fue"}') == {"a": 'dijo "hola" y se fue'}
def test_an_unrepairable_response_reports_the_offending_fragment():
"""El modelo no ve su salida numerada: "line 189 column 22" no le sirve; el
trozo sí."""
with pytest.raises(ValueError) as exc:
extract_json('{"a": 1, "b": [1, 2,,,], "c": 3}')
assert "aquí:" in str(exc.value)
# --- el prompt --------------------------------------------------------------
def test_prompt_carries_the_fetched_contract_not_a_copy():
w, _ = writer("{}")
prompt = w.build_prompt("Caso X", "material", None, "THEEXCLUSIONZONE.COM")
assert "radar_sweep:" in prompt and "contact_bearing_deg" in prompt
assert "1-3 elementos" in prompt # los límites de longitud, del esquema
assert "ink, amber, amber_dark" in prompt # la paleta, también del esquema
def test_prompt_states_the_editorial_constraints():
w, _ = writer("{}")
prompt = w.build_prompt("Caso X", "material", "https://x.test/post", "X.TEST")
assert "20-45 seconds" in prompt
assert "never hex" in prompt
assert "https://x.test/post" in prompt
assert "X.TEST" in prompt
assert "material" in prompt
def test_prompt_includes_the_worked_example_in_full():
w, _ = writer("{}")
prompt = w.build_prompt("Caso X", "material", None, "X.TEST")
example = json.loads(EXAMPLE.read_text())
assert example["meta"]["title"] in prompt
assert "counter_close" in prompt
def test_the_worked_example_narrates_most_of_its_shots():
"""El ejemplo es la señal de formato más fuerte del prompt, más que cualquier
regla en prosa. Sin narración en él, el modelo escribía specs mudos aunque la
sección 3b le dijera lo contrario: en la primera generación tras añadir la voz
narró 1 de 8 planos. Si alguien vuelve a dejar el ejemplo mudo, esto salta.
"""
example = json.loads(EXAMPLE.read_text())
shots = example["shots"]
spoken = [s for s in shots if s.get("narration")]
assert len(spoken) >= len(shots) * 0.6, "el ejemplo enseña a no narrar"
# Y el silencio del ejemplo es una decisión, no un olvido: calla justo donde
# la plantilla ya dibuja una cita.
silent = {s["template"] for s in shots if not s.get("narration")}
assert silent == {"scale_bars", "document_quote"}
def test_the_worked_example_declares_time_for_its_own_narration():
"""Un plano que se queda corto para su propia voz enseña a infradeclarar: el
render no corta la voz, alarga el plano, y el total se va del objetivo."""
from src.generator.spec_contract import NARRATION_CHARS_PER_SECOND
example = json.loads(EXAMPLE.read_text())
for i, shot in enumerate(example["shots"]):
narration = shot.get("narration", "")
if not narration:
continue
needs = len(narration) / NARRATION_CHARS_PER_SECOND
assert shot["duration"] >= needs, f"shot {i} declara menos de lo que habla"
def test_prompt_says_out_loud_that_there_is_no_article_yet():
w, _ = writer("{}")
assert "No article URL yet" in w.build_prompt("X", "m", None, "X.TEST")
def test_prompt_offers_the_live_audio_palette():
"""La mitad de audio del contrato vivo: los presets y sus notas de mood
vienen de GET /audio, no de este repo."""
palette = {"sonar": "the case-file mood", "pulse": "tension, built for debunks"}
w, _ = writer("{}", presets=palette)
prompt = w.build_prompt("X", "m", None, "X.TEST")
assert '"pulse" — tension, built for debunks' in prompt
assert "whose mood fits the shape" in prompt
def test_prompt_without_a_palette_only_offers_the_baseline():
w, _ = writer("{}")
prompt = w.build_prompt("X", "m", None, "X.TEST")
assert '"sonar"' in prompt and '"none"' in prompt
assert '"pulse"' not in prompt
# --- bucle ------------------------------------------------------------------
@pytest.mark.asyncio
async def test_a_good_response_validates_on_the_first_attempt():
w, llm = writer(json.dumps(GOOD))
result = await w.write("Caso X", "material")
assert result.attempts == 1
assert result.spec["meta"]["id"] == "caso"
assert len(llm.prompts) == 1
@pytest.mark.asyncio
async def test_malformed_json_triggers_a_retry():
w, llm = writer("no soy JSON", json.dumps(GOOD))
result = await w.write("Caso X", "material")
assert result.attempts == 2
assert "no es un objeto JSON válido" in result.history[0][0]
assert "previous attempt was rejected" in llm.prompts[1]
@pytest.mark.asyncio
async def test_the_exact_error_paths_are_fed_back_verbatim():
"""La ruta que devuelve la validación es lo más útil que se le puede dar al
modelo: se le pasa tal cual, sin parafrasear."""
bad = json.loads(json.dumps(GOOD))
bad["shots"][0]["props"]["sweeeps"] = 2
w, llm = writer(json.dumps(bad), json.dumps(GOOD))
result = await w.write("Caso X", "material")
assert result.attempts == 2
assert "shots.0.radar_sweep.props.sweeeps" in llm.prompts[1]
assert "sweeps" in llm.prompts[1] # y cuáles sí valen
@pytest.mark.asyncio
async def test_three_failures_raise_but_keep_the_last_attempt():
"""La parte cara es la generación, no el render: el último intento viaja en
la excepción para poder editarlo a mano y reenviarlo."""
bad = json.loads(json.dumps(GOOD))
bad["meta"]["id"] = "Caso Con Espacios"
w, _ = writer(json.dumps(bad))
with pytest.raises(SpecWriteFailed) as exc:
await w.write("Caso X", "material")
assert exc.value.attempts == MAX_ATTEMPTS
assert exc.value.last_spec["meta"]["id"] == "Caso Con Espacios"
assert any("meta.id" in e for e in exc.value.errors)
@pytest.mark.asyncio
async def test_an_off_target_duration_is_commented_once_then_accepted():
"""70 s cumple el contrato pero no el encargo: se comenta y, si el modelo
insiste, se renderiza igual antes que tirar la generación."""
long_spec = json.loads(json.dumps(GOOD))
long_spec["shots"][0]["duration"] = 55.0 # 70 s en total
w, llm = writer(json.dumps(long_spec))
result = await w.write("Caso X", "material")
assert result.attempts == MAX_ATTEMPTS
assert result.notes and "recorta" in result.notes[0]
assert "off-brief" in llm.prompts[1]
@pytest.mark.asyncio
async def test_a_valid_attempt_is_not_thrown_away_by_a_worse_one():
long_spec = json.loads(json.dumps(GOOD))
long_spec["shots"][0]["duration"] = 55.0
w, _ = writer(json.dumps(long_spec), "esto ya no es JSON", "tampoco")
result = await w.write("Caso X", "material")
assert result.spec["shots"][0]["duration"] == 55.0
assert result.notes
@pytest.mark.asyncio
async def test_the_contract_is_refetched_after_a_validation_failure():
"""Si el renderizador se actualizó a mitad de la run, la plantilla nueva
entra en el segundo intento."""
new_template = {"type": "object", "additionalProperties": False,
"required": ["title"],
"properties": {"title": {"type": "string", "minLength": 1}}}
refreshed = {**TEMPLATES, "holo_scan": new_template}
async def refresh():
return refreshed
with_new = json.loads(json.dumps(GOOD))
with_new["shots"][1] = {"template": "holo_scan", "duration": 15.0,
"props": {"title": "X"}}
w, llm = writer(json.dumps(with_new), json.dumps(with_new),
refresh_templates=refresh)
result = await w.write("Caso X", "material")
assert result.attempts == 2
assert "holo_scan" in llm.prompts[1]
assert result.spec["shots"][1]["template"] == "holo_scan"
@pytest.mark.asyncio
async def test_progress_is_reported_only_when_it_retries():
seen = []
async def on_progress(text):
seen.append(text)
w, _ = writer("no JSON", json.dumps(GOOD))
await w.write("Caso X", "material", on_progress=on_progress)
assert len(seen) == 1 and "attempt 2/3" in seen[0]