feat: YouTube fallback strategy + dataset file logging + BLAS thread safety

- Add YouTube fallback strategy to agent prompt: try transcript first, then search for video ID/title/dialogue if blocked
- Log all GAIA dataset file operations: missing rows, downloads, local copies, failures
- Scoring client: fallback to dataset on 404 and invalid JSON payloads
- Force BLAS thread env defaults in Python sandbox to prevent threading conflicts
- Move sys.path.append before imports and apply safe thread env

app.py

@@ -3,12 +3,16 @@ import sys
 import traceback
 from pathlib import Path
 
+sys.path.append(str(Path(__file__).parent / "src"))
+
 from dotenv import load_dotenv
 
 env_path = Path(__file__).parent / ".env"
 load_dotenv(dotenv_path=env_path, override=True)
 
-sys.path.append(str(Path(__file__).parent / "src"))
+from lilith_agent.runtime_env import apply_safe_thread_env
+
+apply_safe_thread_env()
 
 import gradio as gr
 import pandas as pd

src/lilith_agent/app.py

@@ -595,8 +595,15 @@ def build_react_agent(cfg: Config):
             "If `find_files` returns 'No files found' for an exact filename, do NOT escalate to `find_files(root='/')` — "
             "broaden instead: `grep` for a substring, `find_files` with a shorter/partial name, or `ls` the parent "
             "directory to see what's actually there. User filename references may be casual or imprecise (e.g. `.lol` in chat is often laughter, not an extension).\n"
-            "8. MATHEMATICAL PRECISION: If the question requires math, double-check your algebraic calculations carefully. If a specific decimal precision or rounding is asked for (e.g., 'to 2 decimal places', 'nearest tenth'), you MUST calculate precisely and round STRICTLY AT THE VERY END. Do NOT prematurely round intermediate numbers.\n"
-            "9. FINAL ANSWER FORMAT: When you have the final answer, output ONLY the value itself. Do not say 'The answer is...', do not provide explanations in your final output. Just output the bare minimum exact string, number, or list."
+            "8. YOUTUBE FALLBACK STRATEGY: First try `youtube_transcript` for spoken captions. If it is blocked, unavailable, "
+            "or says YouTube is blocking requests, do not repeatedly retry transcript or yt-dlp. Extract the video ID and "
+            "search for `\"<video id>\" transcript`, `\"<video id>\"`, the exact video title, and distinctive quoted dialogue "
+            "or on-screen phrases. Use search snippets, cached transcripts in Hugging Face Spaces/datasets, and reliable web "
+            "pages as evidence. For visual questions, try `youtube_frame_at` only when a timestamp is needed; if video download "
+            "is blocked, pivot to title/time/object searches and answer from the strongest available evidence."
+            "directory to see what's actually there. User filename references may be casual or imprecise (e.g. `.lol` in chat is often laughter, not an extension).\n"
+            "9. MATHEMATICAL PRECISION: If the question requires math, double-check your algebraic calculations carefully. If a specific decimal precision or rounding is asked for (e.g., 'to 2 decimal places', 'nearest tenth'), you MUST calculate precisely and round STRICTLY AT THE VERY END. Do NOT prematurely round intermediate numbers.\n"
+            "10. FINAL ANSWER FORMAT: When you have the final answer, output ONLY the value itself. Do not say 'The answer is...', do not provide explanations in your final output. Just output the bare minimum exact string, number, or list."
         )
         
         if memory_context:

src/lilith_agent/gaia_dataset.py

@@ -58,10 +58,18 @@ class GaiaDatasetClient:
         """Copy the dataset's local file_path into dest_dir."""
         row = next((r for r in self._rows if r["task_id"] == task_id), None)
         if row is None:
+            print(
+                f"[gaia_dataset] file row missing task={task_id} split={self.split} config={self.config}",
+                flush=True,
+            )
             return None
         
         file_name = row.get("file_name")
         if not file_name:
+            print(
+                f"[gaia_dataset] file_name missing task={task_id} split={self.split} config={self.config}",
+                flush=True,
+            )
             return None
 
         dest_root = Path(dest_dir)
@@ -69,16 +77,16 @@ class GaiaDatasetClient:
         dest = dest_root / file_name
 
         if dest.exists() and dest.stat().st_size > 300:
+             print(f"[gaia_dataset] file already exists task={task_id} path={dest}", flush=True)
              return dest
 
-        # 1. Try local data dir fallback
         level = row.get("Level", "1")
         local_data_path = Path("data") / f"gaia_level{level}" / "files" / task_id
         if local_data_path.exists() and local_data_path.stat().st_size > 300:
              shutil.copy(local_data_path, dest)
+             print(f"[gaia_dataset] copied local file task={task_id} source={local_data_path} dest={dest}", flush=True)
              return dest
 
-        # 2. Direct download
         year = "2023"
         if hasattr(self, "config") and "_" in self.config:
             year = self.config.split("_")[0]
@@ -96,8 +104,20 @@ class GaiaDatasetClient:
             if response.status_code == 200:
                 with open(dest, "wb") as f:
                     shutil.copyfileobj(response.raw, f)
+                print(f"[gaia_dataset] downloaded hf file task={task_id} path={dest} url={hf_url}", flush=True)
                 return dest
-        except Exception:
-            pass
+            print(
+                f"[gaia_dataset] hf download failed task={task_id} status={response.status_code} url={hf_url}",
+                flush=True,
+            )
+        except Exception as exc:
+            print(
+                f"[gaia_dataset] hf download exception task={task_id} type={type(exc).__name__} error={exc} url={hf_url}",
+                flush=True,
+            )
             
+        print(
+            f"[gaia_dataset] file unavailable task={task_id} file_name={file_name} split={self.split} config={self.config}",
+            flush=True,
+        )
         return None

src/lilith_agent/runtime_env.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+import os
+
+SAFE_THREAD_ENV = {
+    "OPENBLAS_NUM_THREADS": "1",
+    "OMP_NUM_THREADS": "1",
+    "MKL_NUM_THREADS": "1",
+    "NUMEXPR_NUM_THREADS": "1",
+    "VECLIB_MAXIMUM_THREADS": "1",
+    "TOKENIZERS_PARALLELISM": "false",
+}
+
+
+def apply_safe_thread_env() -> None:
+    for key, value in SAFE_THREAD_ENV.items():
+        os.environ[key] = value

src/lilith_agent/scoring_client.py

@@ -48,18 +48,51 @@ class ScoringApiClient:
             response = self.session.get(f"{self.api_url}/files/{task_id}", timeout=60)
             response.raise_for_status()
         except requests.RequestException as exc:
-            fallback = self._fallback_dataset_client_for(exc, action=f"download file for {task_id}")
+            fallback = self._fallback_dataset_client_for(
+                exc,
+                action=f"download file for {task_id}",
+                missing_is_fallback=True,
+            )
             if fallback is None:
+                print(
+                    f"[scoring] file download failed without fallback task={task_id} error={exc.__class__.__name__}",
+                    flush=True,
+                )
                 return None
-            return fallback.download_file(task_id, dest_dir)
+            path = fallback.download_file(task_id, dest_dir)
+            print(f"[scoring] fallback file result task={task_id} path={path}", flush=True)
+            return path
 
         filename = task_id
         cd = response.headers.get("content-disposition", "")
         if "filename=" in cd:
             filename = cd.split("filename=")[-1].strip().strip('"')
+        content_type = response.headers.get("content-type", "")
+        if self._is_invalid_file_payload(response.content, content_type):
+            print(
+                f"[scoring] invalid file payload task={task_id} status={response.status_code} "
+                f"bytes={len(response.content)} content_type={content_type!r}; trying dataset fallback",
+                flush=True,
+            )
+            try:
+                fallback = self._get_dataset_client()
+            except Exception as exc:
+                print(
+                    f"[scoring] dataset fallback unavailable task={task_id} type={type(exc).__name__} error={exc}",
+                    flush=True,
+                )
+                return None
+            path = fallback.download_file(task_id, dest_dir)
+            print(f"[scoring] fallback file result task={task_id} path={path}", flush=True)
+            return path
         out = dest_dir / filename
         out.write_bytes(response.content)
         self.last_warning = None
+        print(
+            f"[scoring] file downloaded task={task_id} path={out} bytes={len(response.content)} "
+            f"content_type={response.headers.get('content-type', '')!r}",
+            flush=True,
+        )
         return out
 
     def _fallback_dataset_client_for(
@@ -67,8 +100,9 @@ class ScoringApiClient:
         exc: requests.RequestException,
         *,
         action: str,
+        missing_is_fallback: bool = False,
     ) -> Any | None:
-        if not self._should_use_dataset_fallback(exc):
+        if not self._should_use_dataset_fallback(exc, missing_is_fallback=missing_is_fallback):
             return None
         try:
             dataset_client = self._get_dataset_client()
@@ -99,8 +133,14 @@ class ScoringApiClient:
         return self._dataset_client
 
     @staticmethod
-    def _should_use_dataset_fallback(exc: requests.RequestException) -> bool:
+    def _should_use_dataset_fallback(exc: requests.RequestException, *, missing_is_fallback: bool = False) -> bool:
         if isinstance(exc, (requests.Timeout, requests.ConnectionError)):
             return True
         status_code = getattr(getattr(exc, "response", None), "status_code", None)
+        if missing_is_fallback and status_code == 404:
+            return True
         return status_code in {429, 502, 503, 504}
+
+    @staticmethod
+    def _is_invalid_file_payload(content: bytes, content_type: str) -> bool:
+        return "application/json" in content_type.lower() and len(content) < 4096

src/lilith_agent/tools/python_exec.py

@@ -27,9 +27,12 @@ import sys
 import tempfile
 import textwrap
 
+from lilith_agent.runtime_env import SAFE_THREAD_ENV
+
 _OUTPUT_CAP_CHARS = 200_000
 _DOCKER_IMAGE_DEFAULT = "lilith-pysandbox:latest"
 _DOCKER_STARTUP_HEADROOM_S = 5
+BLAS_THREAD_ENV_DEFAULTS = SAFE_THREAD_ENV
 
 # Env vars that are safe (or necessary) to forward. Everything else is dropped.
 _ENV_ALLOWLIST = frozenset({
@@ -46,6 +49,12 @@ _ENV_ALLOWLIST = frozenset({
     "PYTHONPATH",
     "PYTHONHOME",
     "PYTHONIOENCODING",
+    "OPENBLAS_NUM_THREADS",
+    "OMP_NUM_THREADS",
+    "MKL_NUM_THREADS",
+    "NUMEXPR_NUM_THREADS",
+    "VECLIB_MAXIMUM_THREADS",
+    "TOKENIZERS_PARALLELISM",
     # Proxy config is routinely needed for scraping; not a secret.
     "HTTP_PROXY",
     "HTTPS_PROXY",
@@ -102,7 +111,9 @@ _RUNNER_SCRIPT = textwrap.dedent(
 
 def _scrubbed_env() -> dict[str, str]:
     """Return a fresh env dict containing only allowlisted keys from os.environ."""
-    return {k: v for k, v in os.environ.items() if k in _ENV_ALLOWLIST}
+    env = {k: v for k, v in os.environ.items() if k in _ENV_ALLOWLIST}
+    env.update(BLAS_THREAD_ENV_DEFAULTS)
+    return env
 
 
 def _truncate_output(text: str) -> str:

tests/test_gaia_dataset.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from unittest.mock import Mock, patch
+
+from lilith_agent.gaia_dataset import GaiaDatasetClient
+
+
+def _client_with_rows(rows):
+    with patch("lilith_agent.gaia_dataset.load_dataset", return_value=rows):
+        return GaiaDatasetClient(config="2023_all", split="test", level=None, token="hf-token")
+
+
+def test_download_file_prints_when_task_row_is_missing(tmp_path):
+    client = _client_with_rows([])
+
+    with patch("builtins.print") as printed:
+        path = client.download_file("missing-task", tmp_path)
+
+    assert path is None
+    printed.assert_any_call(
+        "[gaia_dataset] file row missing task=missing-task split=test config=2023_all",
+        flush=True,
+    )
+
+
+def test_download_file_prints_hf_download_status_on_failure(tmp_path):
+    client = _client_with_rows([
+        {
+            "task_id": "task-img",
+            "Question": "Q",
+            "Level": "1",
+            "file_name": "task-img.png",
+            "Final answer": "",
+        }
+    ])
+    response = Mock(status_code=404)
+
+    with patch("lilith_agent.gaia_dataset.requests.get", return_value=response) as get, patch("builtins.print") as printed:
+        path = client.download_file("task-img", tmp_path)
+
+    assert path is None
+    assert get.call_args.args[0] == "https://huggingface.co/datasets/gaia-benchmark/GAIA/resolve/main/2023/test/task-img.png"
+    printed.assert_any_call(
+        "[gaia_dataset] hf download failed task=task-img status=404 url=https://huggingface.co/datasets/gaia-benchmark/GAIA/resolve/main/2023/test/task-img.png",
+        flush=True,
+    )

tests/test_graph.py

@@ -91,6 +91,40 @@ def test_build_react_agent_prints_effective_recursion_limit(monkeypatch, tmp_pat
     assert "[graph] effective_recursion_limit=79 logical_recursion_limit=50 budget_hard_cap=25 headroom=4" in captured
 
 
+def test_model_prompt_includes_youtube_fallback_strategy(monkeypatch, tmp_path):
+    class FakeModel:
+        def __init__(self):
+            self.system_prompt = ""
+
+        def bind_tools(self, tools):
+            return self
+
+        def invoke(self, messages):
+            self.system_prompt = str(messages[0].content)
+            return AIMessage(content="Final Answer: inspected")
+
+    fake_model = FakeModel()
+    cfg = Config.from_env()
+    cfg.compact_summarize = False
+    monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith"))
+    monkeypatch.setattr("lilith_agent.app.get_extra_strong_model", lambda cfg: fake_model)
+    monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: fake_model)
+    monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool])
+    monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None)
+
+    graph = build_react_agent(cfg)
+    graph.invoke(
+        {"messages": [HumanMessage(content="What happens in https://www.youtube.com/watch?v=abcdefghijk?")], "iterations": 0, "todos": []},
+        {"configurable": {"thread_id": "youtube-fallback-prompt-test"}},
+    )
+
+    prompt = fake_model.system_prompt.lower()
+    assert "youtube fallback strategy" in prompt
+    assert "video id" in prompt
+    assert "transcript" in prompt
+    assert "do not repeatedly retry" in prompt
+
+
 def test_fail_safe_uses_unbound_model_to_prevent_more_tool_calls(monkeypatch, tmp_path):
     class FakeBoundModel:
         def __init__(self):

tests/test_python_sandbox.py

@@ -75,6 +75,18 @@ def test_allowlisted_env_passes_through():
     assert "HAS_PATH=True" in result
 
 
+def test_blas_thread_env_defaults_are_forced_in_sandbox(monkeypatch):
+    from lilith_agent.tools import python_exec as pe
+
+    for key in pe.BLAS_THREAD_ENV_DEFAULTS:
+        monkeypatch.delenv(key, raising=False)
+
+    env = pe._scrubbed_env()
+
+    for key, value in pe.BLAS_THREAD_ENV_DEFAULTS.items():
+        assert env[key] == value
+
+
 def test_cwd_is_not_repo_root():
     """Subprocess runs in a scratch dir so relative-path writes cannot hit the repo."""
     repo_root = str(Path(__file__).resolve().parent.parent)

tests/test_runtime_env.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+
+def test_apply_safe_thread_env_overrides_existing_values(monkeypatch):
+    from lilith_agent.runtime_env import SAFE_THREAD_ENV, apply_safe_thread_env
+
+    for key in SAFE_THREAD_ENV:
+        monkeypatch.setenv(key, "99")
+
+    apply_safe_thread_env()
+
+    for key, value in SAFE_THREAD_ENV.items():
+        assert __import__("os").environ[key] == value

tests/test_scoring_client.py

@@ -99,6 +99,67 @@ class ScoringApiClientTests(unittest.TestCase):
         self.assertIn("429", client.last_warning or "")
         dataset_client.download_file.assert_called_once_with("task-123", Path(tmpdir))
 
+    def test_download_file_falls_back_to_dataset_on_404(self) -> None:
+        response = Mock(status_code=404, headers={})
+        error = requests.HTTPError("Not Found", response=response)
+        response.raise_for_status.side_effect = error
+
+        session = Mock()
+        session.get.return_value = response
+        dataset_client = Mock()
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            fallback_path = Path(tmpdir) / "from-dataset.png"
+            fallback_path.write_bytes(b"image-bytes")
+            dataset_client.download_file.return_value = fallback_path
+
+            client = ScoringApiClient(
+                api_url="https://example.com",
+                session=session,
+                dataset_client=dataset_client,
+            )
+
+            with patch("builtins.print") as printed:
+                payload = client.download_file("task-404", tmpdir)
+
+        self.assertEqual(payload, fallback_path)
+        self.assertIn("404", client.last_warning or "")
+        dataset_client.download_file.assert_called_once_with("task-404", Path(tmpdir))
+        printed.assert_any_call(
+            "Scoring API unavailable while trying to download file for task-404 (status=404); falling back to GAIA dataset.",
+            flush=True,
+        )
+
+    def test_download_file_falls_back_when_api_returns_tiny_json_payload(self) -> None:
+        response = Mock(status_code=200, headers={"content-type": "application/json"})
+        response.content = b'{"detail":"file not found"}'
+        response.raise_for_status.return_value = None
+
+        session = Mock()
+        session.get.return_value = response
+        dataset_client = Mock()
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            fallback_path = Path(tmpdir) / "from-dataset.png"
+            fallback_path.write_bytes(b"image-bytes")
+            dataset_client.download_file.return_value = fallback_path
+
+            client = ScoringApiClient(
+                api_url="https://example.com",
+                session=session,
+                dataset_client=dataset_client,
+            )
+
+            with patch("builtins.print") as printed:
+                payload = client.download_file("task-json", tmpdir)
+
+        self.assertEqual(payload, fallback_path)
+        dataset_client.download_file.assert_called_once_with("task-json", Path(tmpdir))
+        printed.assert_any_call(
+            "[scoring] invalid file payload task=task-json status=200 bytes=27 content_type='application/json'; trying dataset fallback",
+            flush=True,
+        )
+
     def test_default_dataset_client_is_built_lazily(self) -> None:
         session = Mock()
         response = Mock(status_code=429, headers={})