diff --git a/src/live_mem/config.py b/src/live_mem/config.py
index 2547950..36a058f 100644
--- a/src/live_mem/config.py
+++ b/src/live_mem/config.py
@@ -215,6 +215,16 @@ def _validate_config(self) -> "Settings":
                 "and LLMAAS_API_KEY or neither"
             )
 
+        # LLM budget coherence (issue #32) : an output budget that consumes
+        # the whole context window can never produce a valid call — the
+        # backend rejects input + max_tokens > window before inference.
+        if self.llmaas_max_tokens >= self.llmaas_context_window:
+            errors.append(
+                f"LLMAAS_MAX_TOKENS={self.llmaas_max_tokens} must be strictly "
+                f"less than LLMAAS_CONTEXT_WINDOW={self.llmaas_context_window} "
+                "(output budget cannot consume the whole model context window)"
+            )
+
         # Consolidation ranges
         if self.consolidation_timeout < 10:
             errors.append(
diff --git a/src/live_mem/core/consolidation_queue.py b/src/live_mem/core/consolidation_queue.py
index 82c4833..345087b 100644
--- a/src/live_mem/core/consolidation_queue.py
+++ b/src/live_mem/core/consolidation_queue.py
@@ -201,7 +201,12 @@ async def progress_callback(progress: dict) -> None:
                     job.result = result
                     job.progress.update(
                         {
-                            "phase": "done",
+                            # Issue #32 — keep the lane phase consistent with
+                            # the real outcome (a failed consolidation must
+                            # not display a "done" phase).
+                            "phase": (
+                                "done" if result.get("status") == "ok" else "failed"
+                            ),
                             "batch_size": result.get(
                                 "batch_size", job.progress.get("batch_size")
                             ),
diff --git a/src/live_mem/core/consolidator.py b/src/live_mem/core/consolidator.py
index 369ffc3..4fcb4c0 100644
--- a/src/live_mem/core/consolidator.py
+++ b/src/live_mem/core/consolidator.py
@@ -596,6 +596,11 @@ async def emit_progress(payload: dict) -> None:
         total_prompt_tokens = 0
         total_completion_tokens = 0
         batches_completed = 0
+        # Issue #32 — a failed batch must surface in the final status.
+        # `batch_failure` keeps the upstream error message, `failed_batch`
+        # the 1-based index of the batch that stopped the pipeline.
+        batch_failure: str | None = None
+        failed_batch = 0
         last_synthesis_size = 0
         # Issue #17 — post-pass validation, accumulated over all batches
         validation_unattributed = 0
@@ -677,11 +682,13 @@ async def emit_progress(payload: dict) -> None:
             # Appeler le LLM
             llm_result = await self._call_llm(messages)
             if llm_result.get("status") == "error":
+                batch_failure = llm_result.get("message", "LLM call failed")
+                failed_batch = batch_idx
                 logger.error(
                     "Batch %d/%d LLM failed: %s — stopping (previous batches OK)",
                     batch_idx,
                     batch_count,
-                    llm_result.get("message"),
+                    batch_failure,
                 )
                 break
 
@@ -698,11 +705,13 @@ async def emit_progress(payload: dict) -> None:
             )
 
             if write_result.get("status") != "ok":
+                batch_failure = write_result.get("message", "Bank write failed")
+                failed_batch = batch_idx
                 logger.error(
                     "Batch %d/%d write failed: %s — stopping",
                     batch_idx,
                     batch_count,
-                    write_result.get("message"),
+                    batch_failure,
                 )
                 break
 
@@ -812,10 +821,25 @@ async def emit_progress(payload: dict) -> None:
         bank_objects = await storage.list_objects(f"{space_id}/bank/")
         total_bank = len([o for o in bank_objects if not o["Key"].endswith(".keep")])
 
+        # Issue #32 — the final status must reflect what actually happened.
+        # A batch failure with zero completed batches is an error, not a
+        # success: reporting "ok" here made the queue expose `succeeded`
+        # jobs while 100% of the notes were left unconsolidated.
+        if batch_failure is None:
+            status = "ok"
+            final_phase = "done"
+        elif batches_completed == 0:
+            status = "error"
+            final_phase = "failed"
+        else:
+            status = "partial"
+            final_phase = "failed"
+
         duration = round(time.monotonic() - t0, 1)
         logger.info(
-            "Consolidation done — space=%s agent=%s notes=%d batches=%d/%d "
+            "Consolidation %s — space=%s agent=%s notes=%d batches=%d/%d "
             "created=%d updated=%d tokens=%d duration=%.1fs",
+            status,
             space_id,
             agent_label,
             total_notes,
@@ -828,7 +852,7 @@ async def emit_progress(payload: dict) -> None:
         )
 
         result = {
-            "status": "ok",
+            "status": status,
             "space_id": space_id,
             "notes_processed": total_notes,
             "bank_files_updated": total_updated,
@@ -845,9 +869,17 @@ async def emit_progress(payload: dict) -> None:
             "batch_size": batch_size,
             "duration_seconds": duration,
         }
+        if batch_failure is not None:
+            result["failed_batch"] = failed_batch
+            remaining_notes = len(all_notes) - total_notes
+            result["message"] = (
+                f"Batch {failed_batch}/{batch_count} failed: {batch_failure} — "
+                f"{batches_completed} batch(es) applied, "
+                f"{remaining_notes} note(s) left in live/"
+            )
         await emit_progress(
             {
-                "phase": "done",
+                "phase": final_phase,
                 "batch_size": batch_size,
                 "notes_total": len(all_notes),
                 "notes_done": total_notes,
diff --git a/tests/test_config.py b/tests/test_config.py
index 4643b13..b38d9e8 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -143,6 +143,28 @@ def test_batch_size_zero(self):
             _make_settings(consolidation_batch_size=0)
 
 
+class TestLlmBudgetCoherence:
+    """Issue #32 — max_tokens must never consume the whole context window."""
+
+    def test_max_tokens_equal_to_window_rejected(self):
+        with pytest.raises(ValueError, match="LLMAAS_MAX_TOKENS"):
+            _make_settings(llmaas_max_tokens=131072, llmaas_context_window=131072)
+
+    def test_max_tokens_above_window_rejected(self):
+        # Incident config shape: output budget bigger than a real backend
+        # window makes every large-prompt call impossible.
+        with pytest.raises(ValueError, match="LLMAAS_MAX_TOKENS"):
+            _make_settings(llmaas_max_tokens=200000, llmaas_context_window=131072)
+
+    def test_defaults_are_coherent(self):
+        s = _make_settings()
+        assert s.llmaas_max_tokens < s.llmaas_context_window
+
+    def test_sane_override_accepted(self):
+        s = _make_settings(llmaas_max_tokens=16384, llmaas_context_window=1048576)
+        assert s.llmaas_max_tokens == 16384
+
+
 class TestTemperatureValidation:
     def test_temperature_out_of_range(self):
         with pytest.raises(ValueError, match="LLMAAS_TEMPERATURE"):
diff --git a/tests/test_consolidation_batch_failure.py b/tests/test_consolidation_batch_failure.py
new file mode 100644
index 0000000..4503a03
--- /dev/null
+++ b/tests/test_consolidation_batch_failure.py
@@ -0,0 +1,332 @@
+# -*- coding: utf-8 -*-
+"""
+Tests for issue #32 — a failed batch must never produce a fake success.
+
+Production incident (2026-07-08, space `terraform-provider`): every LLM
+call was rejected by the backend, `consolidate()` still returned
+`status="ok"` with `notes_processed=0`, and the queue exposed the job as
+`succeeded`. The failure stayed invisible for 6 days.
+
+These tests are deliberately adversarial: they replay the exact incident
+conditions (first batch rejected with the real 262144 error message) and
+assert on every observable side effect (status, metrics, `_meta.json`
+untouched, no bank write attempted), not just on the happy path.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from live_mem.core.consolidation_queue import (
+    ConsolidationQueueService,
+    reset_consolidation_queue_for_tests,
+)
+from live_mem.core.consolidator import ConsolidatorService
+
+# Real error template returned by the LLMaaS backend during the incident.
+INCIDENT_LLM_ERROR = (
+    "LLM call failed: Error code: 400 — This model's maximum context length "
+    "is 262144 tokens. However, you requested 200000 output tokens and your "
+    "prompt contains at least 62145 input tokens, for a total of at least "
+    "262145 tokens."
+)
+
+
+class FakeStorage:
+    """Minimal S3 stand-in tracking every mutating call."""
+
+    def __init__(self, notes: list[dict]):
+        self._notes = notes
+        self.put_json_calls: list[tuple[str, dict]] = []
+
+    async def get_json(self, key: str):
+        return {"space_id": "sp", "consolidation_count": 3}
+
+    async def put_json(self, key: str, value: dict):
+        self.put_json_calls.append((key, value))
+
+    async def get(self, key: str):
+        return None  # no rules / no synthesis
+
+    async def list_and_get(self, prefix: str):
+        if prefix.endswith("/live/"):
+            return list(self._notes)
+        return []  # empty bank
+
+    async def list_objects(self, prefix: str):
+        return []
+
+
+def _notes(count: int) -> list[dict]:
+    return [
+        {
+            "key": f"sp/live/20260708T10000{i}_CLR_observation_{i:04x}.md",
+            "content": f"note {i}",
+        }
+        for i in range(count)
+    ]
+
+
+def _consolidator(batch_size: int = 2) -> ConsolidatorService:
+    """
+    Build a ConsolidatorService without touching network or env.
+
+    __init__ instantiates an AsyncOpenAI client (requires an API key and
+    spawns httpx machinery), so the pipeline attributes are set manually.
+    Every LLM/storage interaction is mocked per-test.
+    """
+    svc = object.__new__(ConsolidatorService)
+    svc._model = "test-model"
+    svc._context_window = 131072
+    svc._max_tokens = 16384
+    svc._temperature = 0.3
+    svc._max_notes = 200
+    svc._batch_size = batch_size
+    svc._cooldown_seconds = 0
+    svc._compact_threshold = 0.6
+    svc._bank_file_max_size = 15360
+    svc._validation_enabled = False
+    svc._validation_max_examples = 20
+    return svc
+
+
+def _ok_llm_result() -> dict:
+    return {
+        "status": "ok",
+        "data": {"file_edits": [], "synthesis": "s"},
+        "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+    }
+
+
+def _ok_write_result(notes_count: int) -> dict:
+    return {
+        "status": "ok",
+        "notes_processed": notes_count,
+        "bank_files_created": 0,
+        "bank_files_updated": 1,
+        "operations_applied": 1,
+        "operations_failed": 0,
+        "llm_tokens_used": 15,
+        "llm_prompt_tokens": 10,
+        "llm_completion_tokens": 5,
+        "synthesis_size": 1,
+    }
+
+
+@pytest.mark.asyncio
+class TestFirstBatchFailure:
+    """Replay of the incident: batch 1 rejected → the job must NOT succeed."""
+
+    async def test_first_batch_llm_failure_returns_error_not_ok(self):
+        storage = FakeStorage(_notes(4))  # 4 notes, batch_size=2 → 2 batches
+        svc = _consolidator(batch_size=2)
+        write_results = AsyncMock()
+
+        with patch(
+            "live_mem.core.consolidator.get_storage", return_value=storage
+        ), patch.object(
+            svc,
+            "_call_llm",
+            new=AsyncMock(
+                return_value={"status": "error", "message": INCIDENT_LLM_ERROR}
+            ),
+        ), patch.object(svc, "_write_results", new=write_results):
+            result = await svc.consolidate("sp", agent="CLR", enforce_cooldown=False)
+
+        assert result["status"] == "error", (
+            "A consolidation whose FIRST batch failed must not report ok "
+            f"(got {result['status']!r})"
+        )
+        assert result["notes_processed"] == 0
+        assert result["batches_completed"] == 0
+        assert result["batches_total"] == 2
+        assert result["failed_batch"] == 1
+        # The upstream error must be exploitable by the caller/UI.
+        assert "262144" in result["message"]
+        assert "4 note(s) left in live/" in result["message"]
+        # No bank write must have been attempted.
+        write_results.assert_not_awaited()
+        # _meta.json must NOT be touched: no fake last_consolidation.
+        assert storage.put_json_calls == []
+
+    async def test_first_batch_write_failure_returns_error(self):
+        storage = FakeStorage(_notes(2))
+        svc = _consolidator(batch_size=2)
+
+        with patch(
+            "live_mem.core.consolidator.get_storage", return_value=storage
+        ), patch.object(
+            svc, "_call_llm", new=AsyncMock(return_value=_ok_llm_result())
+        ), patch.object(
+            svc,
+            "_write_results",
+            new=AsyncMock(
+                return_value={"status": "error", "message": "S3 write failed"}
+            ),
+        ):
+            result = await svc.consolidate("sp", agent="CLR", enforce_cooldown=False)
+
+        assert result["status"] == "error"
+        assert result["batches_completed"] == 0
+        assert result["failed_batch"] == 1
+        assert "S3 write failed" in result["message"]
+        assert storage.put_json_calls == []
+
+
+@pytest.mark.asyncio
+class TestPartialFailure:
+    """Failure after successful batches → partial, work is acknowledged."""
+
+    async def test_second_batch_failure_returns_partial_with_metrics(self):
+        storage = FakeStorage(_notes(4))  # 2 batches of 2
+        svc = _consolidator(batch_size=2)
+
+        with patch(
+            "live_mem.core.consolidator.get_storage", return_value=storage
+        ), patch.object(
+            svc,
+            "_call_llm",
+            new=AsyncMock(
+                side_effect=[
+                    _ok_llm_result(),
+                    {"status": "error", "message": INCIDENT_LLM_ERROR},
+                ]
+            ),
+        ), patch.object(
+            svc, "_write_results", new=AsyncMock(return_value=_ok_write_result(2))
+        ):
+            result = await svc.consolidate("sp", agent="CLR", enforce_cooldown=False)
+
+        assert result["status"] == "partial", (
+            "Batch 2 failed after batch 1 was applied: reporting plain ok "
+            "hides the failure, reporting plain error hides the applied work"
+        )
+        assert result["batches_completed"] == 1
+        assert result["notes_processed"] == 2
+        assert result["failed_batch"] == 2
+        assert "262144" in result["message"]
+        assert "1 batch(es) applied" in result["message"]
+        # Work WAS integrated → meta must reflect it.
+        assert len(storage.put_json_calls) == 1
+        meta_key, meta = storage.put_json_calls[0]
+        assert meta_key == "sp/_meta.json"
+        assert meta["total_notes_processed"] == 2
+
+
+@pytest.mark.asyncio
+class TestFullSuccessNonRegression:
+    """The nominal contract must not change."""
+
+    async def test_all_batches_ok_still_returns_ok(self):
+        storage = FakeStorage(_notes(4))
+        svc = _consolidator(batch_size=2)
+
+        with patch(
+            "live_mem.core.consolidator.get_storage", return_value=storage
+        ), patch.object(
+            svc, "_call_llm", new=AsyncMock(return_value=_ok_llm_result())
+        ), patch.object(
+            svc, "_write_results", new=AsyncMock(return_value=_ok_write_result(2))
+        ):
+            result = await svc.consolidate("sp", agent="CLR", enforce_cooldown=False)
+
+        assert result["status"] == "ok"
+        assert result["batches_completed"] == 2
+        assert result["notes_processed"] == 4
+        assert "failed_batch" not in result
+        assert "message" not in result
+        assert len(storage.put_json_calls) == 1
+
+    async def test_no_notes_still_returns_ok(self):
+        storage = FakeStorage([])
+        svc = _consolidator(batch_size=2)
+
+        with patch(
+            "live_mem.core.consolidator.get_storage", return_value=storage
+        ):
+            result = await svc.consolidate("sp", agent="CLR", enforce_cooldown=False)
+
+        assert result["status"] == "ok"
+        assert result["notes_processed"] == 0
+
+
+@pytest.mark.asyncio
+class TestQueueIntegration:
+    """End-to-end through the queue: the job must expose the failure."""
+
+    async def test_job_is_failed_with_exploitable_error_on_batch_failure(self):
+        reset_consolidation_queue_for_tests()
+
+        class IncidentConsolidator:
+            async def consolidate(
+                self, space_id, agent="", enforce_cooldown=True, progress_callback=None
+            ):
+                # Shape returned by consolidate() after the fix when the
+                # first batch fails (incident replay).
+                return {
+                    "status": "error",
+                    "space_id": space_id,
+                    "notes_processed": 0,
+                    "batches_total": 50,
+                    "batches_completed": 0,
+                    "failed_batch": 1,
+                    "message": f"Batch 1/50 failed: {INCIDENT_LLM_ERROR}",
+                }
+
+        queue = ConsolidationQueueService()
+        with patch(
+            "live_mem.core.consolidation_queue.get_consolidator",
+            return_value=IncidentConsolidator(),
+        ):
+            job = await queue.enqueue("terraform-provider", "CLR", "CLR")
+            for _ in range(50):
+                status = await queue.get_job(job["job_id"])
+                if status["status"] in ("succeeded", "failed"):
+                    break
+                await asyncio.sleep(0.01)
+
+        assert status["status"] == "failed", (
+            "A consolidation with 0/50 batches completed must never be "
+            "exposed as succeeded (this is the exact incident symptom)"
+        )
+        assert "262144" in status["error"]
+        assert status["result"]["notes_processed"] == 0
+        assert status["progress"]["phase"] == "failed"
+
+    async def test_partial_result_is_exposed_as_failed_job(self):
+        reset_consolidation_queue_for_tests()
+
+        class PartialConsolidator:
+            async def consolidate(
+                self, space_id, agent="", enforce_cooldown=True, progress_callback=None
+            ):
+                return {
+                    "status": "partial",
+                    "space_id": space_id,
+                    "notes_processed": 60,
+                    "batches_total": 50,
+                    "batches_completed": 30,
+                    "failed_batch": 31,
+                    "message": f"Batch 31/50 failed: {INCIDENT_LLM_ERROR}",
+                }
+
+        queue = ConsolidationQueueService()
+        with patch(
+            "live_mem.core.consolidation_queue.get_consolidator",
+            return_value=PartialConsolidator(),
+        ):
+            job = await queue.enqueue("terraform-provider", "CLR", "CLR")
+            for _ in range(50):
+                status = await queue.get_job(job["job_id"])
+                if status["status"] in ("succeeded", "failed"):
+                    break
+                await asyncio.sleep(0.01)
+
+        assert status["status"] == "failed"
+        assert "Batch 31/50" in status["error"]
+        # Partial progress must stay visible for the operator.
+        assert status["result"]["batches_completed"] == 30
+        assert status["result"]["notes_processed"] == 60