dmux · Oct 27, 2026 · Oct 26, 2026 · Oct 26, 2026 · Oct 26, 2026 · Oct 26, 2026
diff --git a/src/minilake/services/jobs.py b/src/minilake/services/jobs.py
@@ -189,6 +189,26 @@ async def delete_job(req: DeleteJobRequest) -> dict:
 # ============================================================================
 
 
+def _read_sql_task_file(sql_path, display_path: str) -> str:
+    """Read workspace SQL and report file and decoding failures."""
+    try:
+        if not sql_path.is_file():
+            raise DatabricksError(
+                error_code="INVALID_PARAMETER_VALUE",
+                message=f"Workspace SQL path '{display_path}' is not a file",
+                status_code=400,
+            )
+        return sql_path.read_text(encoding="utf-8")
+    except DatabricksError:
+        raise
+    except (OSError, UnicodeError) as e:
+        raise DatabricksError(
+            error_code="INVALID_PARAMETER_VALUE",
+            message=f"Unable to read workspace SQL file '{display_path}': {e}",
+            status_code=400,
+        ) from e
+
+
 def _resolve_sql_task_text(sql_task: SqlTask) -> str:
     """Return the SQL a sql_task should run, from whichever source it names.
 
@@ -210,7 +230,7 @@ def _resolve_sql_task_text(sql_task: SqlTask) -> str:
                 message=f"Workspace file '{sql_task.file.path}' not found",
                 status_code=404,
             )
-        return sql_path.read_text()
+        return _read_sql_task_file(sql_path, sql_task.file.path)
 
     if sql_task.query is not None:
         return saved_queries.resolve_query_text(sql_task.query.query_id)
@@ -268,6 +288,7 @@ async def _execute_sql_task(task: Task, task_run_id: int, now_ms: int) -> RunTas
     try:
         sql_text = _resolve_sql_task_text(sql_task)
     except DatabricksError as e:
+        _state["task_outputs"][task_run_id] = {"logs": None, "error": e.message}
         return RunTaskInfo(
             task_key=task.task_key,
             run_id=task_run_id,
@@ -535,10 +556,32 @@ async def _execute_run(run_id: int, argv: List[str]) -> None:
     Each task's run_if is evaluated against its *direct* dependencies' real
     outcomes before it runs; a task whose condition isn't met is SKIPPED.
     """
-    run = _state["runs"][run_id]
+    run = _state["runs"].get(run_id)
+    if run is None:
+        return
     run["state"] = {"life_cycle_state": "RUNNING", "result_state": None, "state_message": None}
     run["start_time"] = int(time.time() * 1000)
 
+    try:
+        await _execute_run_tasks(run_id, argv)
+    except Exception as e:
+        import traceback
+
+        diagnostic = traceback.format_exc()
+        message = f"Run execution failed: {type(e).__name__}: {e}"
+        logger.error("Run %s failed unexpectedly: %s", run_id, diagnostic)
+        run["end_time"] = int(time.time() * 1000)
+        run["state"] = {
+            "life_cycle_state": "TERMINATED",
+            "result_state": RunResultState.FAILED.value,
+            "state_message": message,
+        }
+        _state.setdefault("task_outputs", {})[run_id] = {"logs": None, "error": diagnostic}
+
+
+async def _execute_run_tasks(run_id: int, argv: List[str]) -> None:
+    """Run the shared DAG scheduler after the outer run boundary is established."""
+    run = _state["runs"][run_id]
     # A submitted run carries its own tasks (no job_id); a triggered run reads them
     # from the job. Everything downstream — the DAG, run_if, cancellation — is shared.
     job = _state["jobs"].get(run["job_id"])
@@ -577,7 +620,12 @@ async def run_one(task_key: str) -> None:
                     state=RunState(life_cycle_state="SKIPPED", state_message="Unresolvable dependency or cycle"),
                 )
             break
-        await asyncio.gather(*(run_one(k) for k in ready))
+        outcomes = await asyncio.gather(*(run_one(k) for k in ready), return_exceptions=True)
+        for outcome in outcomes:
+            if isinstance(outcome, asyncio.CancelledError):
+                raise RuntimeError("A job task was cancelled unexpectedly") from outcome
+            if isinstance(outcome, Exception):
+                raise outcome
         pending -= set(ready)
 
     task_results = [results[t.task_key] for t in tasks]

diff --git a/tests/test_jobs_submit.py b/tests/test_jobs_submit.py
@@ -6,20 +6,25 @@
 run must be visible to `runs/get` / `runs/list` like any other.
 """
 
+import asyncio
 from datetime import timedelta
+from uuid import uuid4
 
 import pytest
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.errors import DatabricksError
 from databricks.sdk.service.jobs import (
     RunIf,
     SqlTask,
+    SqlTaskFile,
     SqlTaskQuery,
     SubmitTask,
     TaskDependency,
 )
 from databricks.sdk.service.sql import CreateQueryRequestQuery
 
+from minilake.services import jobs
+
 RUN_TIMEOUT = timedelta(minutes=5)
 
 
@@ -43,6 +48,16 @@ def _task(task_key: str, warehouse: str, query_id: str, **kwargs) -> SubmitTask:
     )
 
 
+def _import_bytes(client: WorkspaceClient, path: str, content: bytes) -> None:
+    client.api_client.do(
+        "POST",
+        f"/api/2.0/workspace-files/import-file{path}",
+        query={"overwrite": "true"},
+        headers={"Content-Type": "application/octet-stream"},
+        data=content,
+    )
+
+
 @pytest.mark.serial
 @pytest.mark.workflow
 def test_submit_runs_without_a_job(workspace_client: WorkspaceClient, warehouse: str, query_id: str):
@@ -104,6 +119,93 @@ def test_submit_honours_task_dependencies(workspace_client: WorkspaceClient, war
     assert by_key["never"].state.life_cycle_state.value == "SKIPPED"
 
 
+@pytest.mark.serial
+@pytest.mark.workflow
+def test_submit_sql_task_directory_fails_terminally(workspace_client: WorkspaceClient, warehouse: str):
+    path = f"/Shared/sql_task_dir_{uuid4().hex[:8]}"
+    workspace_client.workspace.mkdirs(path=path)
+
+    run = workspace_client.jobs.submit(
+        run_name="sql-directory",
+        tasks=[SubmitTask(task_key="directory", sql_task=SqlTask(warehouse_id=warehouse, file=SqlTaskFile(path=path)))],
+    ).result(timeout=RUN_TIMEOUT)
+
+    assert run.state.life_cycle_state.value == "TERMINATED"
+    assert run.state.result_state.value == "FAILED"
+    assert "not a file" in (run.tasks[0].state.state_message or "")
+    output = workspace_client.jobs.get_run_output(run_id=run.run_id)
+    assert "not a file" in (output.error or "")
+
+
+@pytest.mark.serial
+@pytest.mark.workflow
+def test_submit_sql_task_invalid_utf8_fails_terminally(workspace_client: WorkspaceClient, warehouse: str):
+    path = f"/Shared/sql_task_invalid_utf8_{uuid4().hex[:8]}.sql"
+    _import_bytes(workspace_client, path, b"SELECT \xff")
+
+    run = workspace_client.jobs.submit(
+        run_name="sql-invalid-utf8",
+        tasks=[
+            SubmitTask(task_key="invalid-utf8", sql_task=SqlTask(warehouse_id=warehouse, file=SqlTaskFile(path=path)))
+        ],
+    ).result(timeout=RUN_TIMEOUT)
+
+    assert run.state.life_cycle_state.value == "TERMINATED"
+    assert run.state.result_state.value == "FAILED"
+    output = workspace_client.jobs.get_run_output(run_id=run.run_id)
+    assert "Unable to read workspace SQL file" in (output.error or "")
+    assert "codec can't decode" in (output.error or "")
+
+
+def test_run_executor_exception_marks_run_failed_with_diagnostics(monkeypatch):
+    run_id = 987654321
+    run = {
+        "run_id": run_id,
+        "job_id": None,
+        "state": {"life_cycle_state": "PENDING", "result_state": None, "state_message": None},
+        "_tasks": [{"task_key": "unexpected"}],
+    }
+    jobs._state["runs"][run_id] = run
+    sibling_started = asyncio.Event()
+    release_sibling = asyncio.Event()
+    sibling_finished = asyncio.Event()
+
+    async def fail_executor(task, argv):
+        if task.task_key == "unexpected":
+            await sibling_started.wait()
+            raise RuntimeError("executor exploded")
+        sibling_started.set()
+        try:
+            await release_sibling.wait()
+        finally:
+            sibling_finished.set()
+        return jobs.RunTaskInfo(task_key=task.task_key, state=jobs.RunState(life_cycle_state="TERMINATED"))
+
+    run["_tasks"].append({"task_key": "sibling"})
+
+    monkeypatch.setattr(jobs, "_execute_task", fail_executor)
+    try:
+
+        async def run_with_sibling():
+            running = asyncio.create_task(jobs._execute_run(run_id, []))
+            await asyncio.wait_for(sibling_started.wait(), timeout=1)
+            await asyncio.sleep(0.01)
+            assert not running.done()
+            assert run["state"]["life_cycle_state"] == "RUNNING"
+            release_sibling.set()
+            await running
+
+        asyncio.run(run_with_sibling())
+        assert sibling_finished.is_set()
+        assert run["state"]["life_cycle_state"] == "TERMINATED"
+        assert run["state"]["result_state"] == "FAILED"
+        assert "executor exploded" in run["state"]["state_message"]
+        assert "RuntimeError: executor exploded" in jobs._state["task_outputs"][run_id]["error"]
+    finally:
+        jobs._state["runs"].pop(run_id, None)
+        jobs._state.get("task_outputs", {}).pop(run_id, None)
+
+
 @pytest.mark.serial
 def test_submit_idempotency_token_reuses_the_run(workspace_client: WorkspaceClient, warehouse: str, query_id: str):
     """Re-submitting with the same token returns the original run, not a second one."""