fix(planner): attributable step failures, safe narration, headless exit prompt

conftest.py

@@ -0,0 +1,27 @@
+"""Make the tests import the working tree, not an installed snapshot.
+
+``pyproject.toml`` force-includes ``main.py`` and ``config.py`` as top-level
+modules in the wheel, so a ``uv sync`` / ``pip install`` of this project drops
+copies of both into ``site-packages``. Under pytest those copies won every
+import: ``import config`` in ``tests/config_roundtrip_test.py`` resolved to
+``.venv/lib/python3.11/site-packages/config.py`` — a snapshot frozen at install
+time — while ``sources/`` (not shipped as a top-level module) correctly
+resolved to the tree.
+
+The result was a test suite that could pass green against code the repository
+no longer contained: a config field added in the working tree was simply
+invisible to the round-trip test that exists to check exactly that.
+
+Putting the repo root at the FRONT of ``sys.path`` makes the tree win.
+"""
+
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent
+
+# Front, not append: site-packages is already on the path and would otherwise
+# keep winning for `main` and `config`.
+if str(REPO_ROOT) in sys.path:
+    sys.path.remove(str(REPO_ROOT))
+sys.path.insert(0, str(REPO_ROOT))

main.py

@@ -124,6 +124,16 @@ async def papers_mode(args, config):
                                   learning=args.learn,
                                   single_agent_mode=args.single_agent
                                  )
+    # Per-row failures are caught so the batch continues, but they must still
+    # reach the exit code — otherwise a run where every task failed is
+    # indistinguishable from a clean one to any wrapping harness.
+    failed = papers.errored_rows
+    if failed:
+        print_err(
+            f"{len(failed)} of {len(papers.execution_history)} CSV row(s) failed; "
+            "exiting non-zero."
+        )
+        sys.exit(1)
 
 async def science_bench_papers_mode(args, config):
     # Use concurrent evaluation by default for science_agent_bench

sources/benchmark_evaluation/csv_mode.py

@@ -1296,6 +1296,18 @@ def _build_summary_rows(self) -> tuple[list[tuple[str, str]], list[dict], list[d
             ]
         return rows, current_runs, sab_runs
 
+    @property
+    def errored_rows(self) -> list[dict]:
+        """Rows whose evaluation raised, recorded as ``success_level == "Error"``.
+
+        The loop catches per-row exceptions and continues, which is right for a
+        batch — one bad row should not abandon the rest. But nothing downstream
+        reflected it, so a run in which every row failed still returned exit 0
+        and any harness reading the exit code scored it as a clean pass.
+        """
+        return [d for d in self.execution_history
+                if d.get("success_level") == "Error"]
+
     def _print_final_summary(self) -> None:
         """Print a summary of all autonomous executions."""
         rows, current_runs, sab_runs = self._build_summary_rows()

sources/core/planner.py

@@ -1,10 +1,11 @@
 import json
+import logging
 import os
 import re
 import sys
 import threading
 import time
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from typing import Any
 
 from sources.cli.pretty_print import (
@@ -34,6 +35,15 @@
 from .workflow_selection import WorkflowSelector
 
 
+class UserInterventionRequired(Exception):
+    """A decision needs a human, and no human is reachable.
+
+    Raised instead of blocking on ``input()`` when stdin is not a TTY, so an
+    unattended run fails with the question it could not ask rather than with
+    ``EOF when reading a line``.
+    """
+
+
 class PlanValidationError(Exception):
     """Exception raised when plan validation fails."""
     pass
@@ -66,6 +76,7 @@ def __init__(self, config: "Config", enable_tts: bool = True) -> None:
             raise ValueError("❌ Planner: Configuration cannot be None")
 
         self.config = config
+        self.logger = logging.getLogger(__name__)
         self.workspace_path = config.workspace_dir
         self.evolve = EvolutionEngine(config)
         self.task_history: list[Task] = []
@@ -635,6 +646,17 @@ def _verify_expected_outputs(self, step: PlanStep) -> tuple[bool, list[str]]:
         workspace_files = self._get_workspace_files()
 
         for expected_output in step.expected_outputs:
+            # A plan may declare a *directory* as an output ("/workspace/data/").
+            # The workspace scan yields files only, so such an output could never
+            # be matched and the step stayed permanently "missing outputs" —
+            # which then blocked every dependent step. Observed on p_iimn
+            # 2026-08-22: data_acquisition wrote nine files under data/ and was
+            # still reported as missing /workspace/data/.
+            if str(expected_output).rstrip().endswith(("/", "\\")):
+                if self._directory_output_satisfied(expected_output, workspace_files):
+                    continue
+                missing_outputs.append(expected_output)
+                continue
             # Normalise expected path to forward slashes for cross-platform comparison
             normalised_expected = Path(expected_output).as_posix()
             # Use Path.stem to strip the extension in a platform-agnostic way
@@ -650,6 +672,22 @@ def _verify_expected_outputs(self, step: PlanStep) -> tuple[bool, list[str]]:
 
         return len(missing_outputs) == 0, missing_outputs
 
+    @staticmethod
+    def _directory_output_satisfied(expected_output: str, workspace_files: list[str]) -> bool:
+        """True when any workspace file sits inside the declared directory.
+
+        Matches on the trailing directory name rather than the full path: plans
+        declare workspace-absolute paths ("/workspace/data/") while the scan
+        returns paths relative to the workspace root ("data/features.csv").
+        """
+        name = PurePosixPath(str(expected_output).replace("\\", "/").rstrip("/")).name.lower()
+        if not name:
+            return False
+        return any(
+            name in [part.lower() for part in PurePosixPath(actual).parent.parts]
+            for actual in workspace_files
+        )
+
     def _can_execute_step(self, step: PlanStep) -> tuple[bool, list[str]]:
         """
         Check if a step can be executed based on its dependencies.
@@ -675,7 +713,19 @@ def _can_execute_step(self, step: PlanStep) -> tuple[bool, list[str]]:
         return len(missing_deps) == 0, missing_deps
 
     def request_user_exit(self, msg: str) -> None:
-        """Send a notification and prompt the user to continue or exit.
+        """Ask whether to continue — but only when someone can answer.
+
+        On a non-TTY this raises :class:`UserInterventionRequired` instead of
+        reading stdin. The prompt was the last blocking ``input()`` on the
+        benchmark path: on the p_iimn run of 2026-08-22 the planner reached
+        step 4 of 6, asked "Continue ? (y/n)" into a redirected stdout, and
+        died with ``EOF when reading a line`` — a message that names neither
+        the question nor the step it was asked about.
+
+        Raising rather than ``exit(1)`` is deliberate: the CSV harness counts
+        the row as failed and still prints its summary, which a ``SystemExit``
+        from inside the planner would skip. The same rule is already applied in
+        ``pricing.py`` and ``csv_mode._prompt_with_default``.
 
         Args:
             msg: Message shown both in the Pushover notification body and
@@ -686,6 +736,13 @@ def request_user_exit(self, msg: str) -> None:
             title="Mimosa exit request."
         )
         print(msg)
+
+        if not sys.stdin.isatty():
+            self.logger.error("Intervention needed but stdin is not a TTY: %s", msg)
+            raise UserInterventionRequired(
+                f"{msg}\n(stdin is not a TTY — cannot ask whether to continue)"
+            )
+
         choice = input("\nContinue ? (y(yes)/n(no))")
         if choice.lower() == "y" or choice.lower() == "yes":
             return
@@ -895,13 +952,51 @@ async def run_attempts(
 
         step.cost = attempt_cost
         step.score = attempt_score
-        if self.tts:
-            answer = '. '.join([x[:128] for x in final_answers if x]) if final_answers else "No answers produced."
+        self._narrate_step_completion(step_name, attempt_score, attempt_cost, final_answers)
+        return step
+
+    def _narrate_step_completion(
+        self,
+        step_name: str,
+        attempt_score: float,
+        attempt_cost: float,
+        final_answers: list[Any],
+    ) -> None:
+        """Speak a step's outcome, without ever being able to fail the step.
+
+        Two defects met here on a real run and cost it everything it had
+        produced.
+
+        ``final_answers`` is annotated ``list[str]`` but agents answer with a
+        structured object: every entry of that run's ``state_result.json`` is a
+        dict (``{"status": ..., "approach": ...}``). Slicing one raised
+        ``TypeError: unhashable type: 'slice'`` on Python 3.11 — and on 3.12+,
+        where slices became hashable, the same line degrades to a ``KeyError``
+        instead. Every other consumer already coerces first
+        (``planner.py`` line ~397, ``evolution_engine.py`` line ~245); this one
+        did not.
+
+        And the narration sat inside the step body, so a cosmetic summary
+        propagated out as "Critical error in step execution" — reported after
+        the step had already written its deliverable and its ASTRA capsule, and
+        turning a scored run into a 0% success rate and a non-zero exit. What
+        is spoken aloud must never decide whether the work counts.
+        """
+        if not self.tts:
+            return
+        try:
+            answer = (
+                '. '.join([str(x)[:128] for x in final_answers if x])
+                if final_answers else "No answers produced."
+            )
             tts_text = f"""
             Task completed. Score: {attempt_score}, Cost: {attempt_cost}. {answer}
             """
             self.tts.speak(tts_text, voice_index=0)
-        return step
+        except Exception:
+            # Loud, but not fatal: the operator still learns narration broke.
+            self.logger.exception("TTS narration failed for step '%s'", step_name)
+            print_warn(f"Could not narrate completion of step '{step_name}'")
 
     async def start_planner(
         self,
@@ -976,7 +1071,19 @@ async def start_planner(
                 except Exception as e:
                     step.status = TaskStatus.FAILED
                     self._update_visualization(total_cost)  # Update to show failed status
-                    raise Exception(f"❌ Critical error in step execution: {str(e)}") from e
+                    # Log the traceback before re-raising. `from e` preserves the
+                    # chain for a Python caller, but the operator only ever sees
+                    # the formatted message — so a bare TypeError like
+                    # "unhashable type: 'slice'" arrives with no file or line and
+                    # is effectively unattributable. Observed on a real run that
+                    # had already produced its deliverable.
+                    self.logger.exception(
+                        "Step '%s' (%d/%d) failed", step_name, step_idx + 1,
+                        len(self.current_plan.steps),
+                    )
+                    raise Exception(
+                        f"❌ Critical error in step execution: {type(e).__name__}: {e}"
+                    ) from e
                 lst_step = step
 
                 if step.status != TaskStatus.COMPLETED:

sources/core/schema.py

@@ -5,6 +5,7 @@
 from dataclasses import dataclass, field
 from enum import Enum
 from datetime import datetime
+from typing import Any
 
 class TaskStatus(Enum):
     """Enumeration for task execution status."""
@@ -59,7 +60,10 @@ class IndividualRun:
     workflow_template: str | None = None
     scenario_rubric: str | None = None
     eval_type: str | None = None
-    answers: list[str] | None = None
+    # Agents answer with a structured object, not a string: every entry of
+    # a real run's state_result.json is a dict. Annotated list[str] until a
+    # consumer sliced one and raised "unhashable type: 'slice'".
+    answers: list[Any] | None = None
     state_result: dict | None = None
     plot: str | None = ""
     original_task: str | None = None  # Original unwrapped task for similarity matching
@@ -151,7 +155,7 @@ class Task:
     description: str
     run_id: int = 0
     evolve_runs: list[IndividualRun] = field(default_factory=list) # evolution run result for task
-    final_answers: list[str] = field(default_factory=list) # last evolution run answers
+    final_answers: list[Any] = field(default_factory=list) # last evolution run answers; entries are usually dicts
     cost: float = 0
     final_uuid: str | None = None # last evolution run uuid
     workflow_uuid: str | None = None # last workflow uuid

sources/modules/smolagent_factory.py

@@ -435,15 +435,22 @@ def _run_agent():
                 try:
                     result['response'] = self.agent.run(instructions)
                     result['completed'] = True
+                    # Drop any earlier retry's exception — this attempt worked,
+                    # and the caller re-raises whatever is left here.
+                    result['exception'] = None
                     error = False
                     warning = True
                 except Exception as e:
                     print(str(e))
                     print("retrying...")
                     error = True
                     count += 1
-                #result['exception'] = e
-                #result['completed'] = True
+                    # Keep the last failure. Without it an agent that exhausts
+                    # its retries leaves completed=False with no exception, the
+                    # thread exits, join() returns at once, and the caller below
+                    # reports a timeout that never happened — discarding the
+                    # real cause.
+                    result['exception'] = e
 
         agent_thread = threading.Thread(target=_run_agent, daemon=True)
         agent_thread.start()
@@ -452,7 +459,18 @@ def _run_agent():
         try:
             if not result['completed']:
                 # no save here: the except branch below saves once for all failures
-                raise TimeoutError(f"Agent '{self.name}' execution timed out after {timeout_seconds} seconds")
+                if agent_thread.is_alive():
+                    # Still running past the deadline: a genuine timeout.
+                    raise TimeoutError(
+                        f"Agent '{self.name}' execution timed out after {timeout_seconds} seconds"
+                    )
+                # Finished without completing: every retry raised. Surface the
+                # last real exception instead of inventing a timeout.
+                if result['exception'] is not None:
+                    raise result['exception']
+                raise RuntimeError(
+                    f"Agent '{self.name}' exhausted its retries without producing a response"
+                )
             if result['exception']:
                 raise result['exception']
             self.save_memories(workflow_uuid=workflow_uuid)