diff --git a/conformance.toml b/conformance.toml
index 3fc989c..f38dd6c 100644
--- a/conformance.toml
+++ b/conformance.toml
@@ -826,9 +826,10 @@ note = "The graph-engine §6 event surface + llm-provider §7 error surface (PR
 
 # Spec v0.78.0 (proposal 0083).  Per-prompt token-budget observability
 # (prompt-management §3 / graph-engine §6 / observability §5.5.15).
-# Not-yet overall: the completion-path OTel rendering + fixtures 126-129 run
-# (PR A), but the Langfuse WARNING (130) and structured-output-failure (131)
-# paths stay deferred until their PRs; flips to implemented when they land.
+# Not-yet overall: the completion-path OTel rendering + fixtures 126-129 (PR A)
+# and the Langfuse WARNING + §7 log + fixture 130 (PR B) run; only the
+# structured-output-failure path (131) stays deferred until PR C, which is when
+# this flips to implemented.
 [proposals."0083"]
 status = "not-yet"
 
diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py
index 04430ac..8066c57 100644
--- a/src/openarmature/observability/langfuse/observer.py
+++ b/src/openarmature/observability/langfuse/observer.py
@@ -47,6 +47,7 @@
 )
 from openarmature.graph.observer import ObserverEvent
 from openarmature.observability.lineage import is_strict_prefix
+from openarmature.observability.llm_event import _token_budget_evaluations
 
 from .client import (
     LangfuseClient,
@@ -1753,6 +1754,20 @@ def _handle_typed_llm_completion(self, event: LlmCompletionEvent) -> None:
         end_kwargs: dict[str, Any] = {}
         if usage is not None:
             end_kwargs["usage"] = usage
+        # §8.4.3 (proposal 0083): a token-budget exceedance sets the Generation's
+        # advisory WARNING level + a statusMessage naming each breached bound. The
+        # call succeeded (this is the success handler), so WARNING never displaces
+        # a hard ERROR — the failure path is untouched. No breach -> end as before.
+        breached = [
+            ev
+            for ev in _token_budget_evaluations(event.token_budget, event.usage)
+            if ev["actual"] > ev["max"]
+        ]
+        if breached:
+            end_kwargs["level"] = "WARNING"
+            end_kwargs["status_message"] = "token budget exceeded: " + ", ".join(
+                f"{ev['kind']} {ev['actual']} > {ev['max']}" for ev in breached
+            )
         handle.end(end_time=end_time, **end_kwargs)
 
     def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None:
@@ -2214,6 +2229,19 @@ def _typed_event_metadata(
         active_group = event.active_prompt_group
         if active_group is not None:
             metadata["prompt_group_name"] = active_group.group_name
+        # §8.4.3 (proposal 0083): the declared token-budget bounds map to
+        # metadata.token_budget.*, each key present only when that bound is
+        # non-None. Omitted entirely when no budget was declared. Shared with the
+        # failure handler (harmless there; a failed Generation carries it too).
+        token_budget = event.token_budget
+        if token_budget is not None:
+            budget: dict[str, Any] = {}
+            if token_budget.input_max_tokens is not None:
+                budget["input_max_tokens"] = token_budget.input_max_tokens
+            if token_budget.total_max_tokens is not None:
+                budget["total_max_tokens"] = token_budget.total_max_tokens
+            if budget:
+                metadata["token_budget"] = budget
         if event.caller_invocation_metadata is not None:
             _apply_caller_metadata(metadata, event.caller_invocation_metadata)
         # Response-side metadata. A completion always carries it; a
diff --git a/src/openarmature/observability/llm_event.py b/src/openarmature/observability/llm_event.py
index ef64139..f567eab 100644
--- a/src/openarmature/observability/llm_event.py
+++ b/src/openarmature/observability/llm_event.py
@@ -162,4 +162,40 @@ def serialize_tool_calls(tool_calls: Sequence[ToolCall]) -> list[dict[str, Any]]
     return [{"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in tool_calls]
 
 
-__all__ = ["LLM_NAMESPACE", "LlmEventPayload", "serialize_tool_calls"]
+def _token_budget_evaluations(token_budget: Any, usage: Any) -> list[dict[str, Any]]:
+    # §5.5.15 / §11.2 (proposal 0083): the per-bound token-budget evaluation
+    # shared by the OTel span attrs, the metric recorder, and the Langfuse
+    # WARNING / §7 log surfaces. Returns one entry per DECLARED bound that also
+    # has a usable actual to compare against, so a bound whose count the provider
+    # did not report (None) is omitted -- mirroring the token.usage gate rather
+    # than coercing a missing count to 0. A declared bound of 0 IS evaluated: the
+    # exceeded test is a strict ``actual > max``, so a 0 bound is exceeded by any
+    # positive usage per §5.5.15; only the utilization ratio is undefined for a 0
+    # denominator, and the metric recorder skips that one sample. Empty when there
+    # is no budget, no usage, or no bound with a reported count; the caller then
+    # leaves the exceeded signal absent (not false).
+    if token_budget is None or usage is None:
+        return []
+    evaluations: list[dict[str, Any]] = []
+    input_max = getattr(token_budget, "input_max_tokens", None)
+    if input_max is not None and usage.prompt_tokens is not None:
+        evaluations.append({"kind": "input", "actual": usage.prompt_tokens, "max": input_max})
+    total_max = getattr(token_budget, "total_max_tokens", None)
+    if total_max is not None:
+        if usage.total_tokens is not None:
+            total_actual = usage.total_tokens
+        elif usage.prompt_tokens is not None and usage.completion_tokens is not None:
+            total_actual = usage.prompt_tokens + usage.completion_tokens
+        else:
+            total_actual = None
+        if total_actual is not None:
+            evaluations.append({"kind": "total", "actual": total_actual, "max": total_max})
+    return evaluations
+
+
+__all__ = [
+    "LLM_NAMESPACE",
+    "LlmEventPayload",
+    "_token_budget_evaluations",
+    "serialize_tool_calls",
+]
diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py
index 35a56e7..1817134 100644
--- a/src/openarmature/observability/otel/observer.py
+++ b/src/openarmature/observability/otel/observer.py
@@ -75,6 +75,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import time
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, field
@@ -117,7 +118,17 @@
 )
 from openarmature.graph.observer import ObserverEvent
 from openarmature.observability.lineage import is_strict_prefix
-from openarmature.observability.llm_event import serialize_tool_calls
+from openarmature.observability.llm_event import _token_budget_evaluations, serialize_tool_calls
+
+# §7 (proposal 0083): the vendor-neutral token-budget WARNING log surface.
+# The logger name is our choice (the spec doesn't pin one). The OTLP log
+# bridge attaches correlation_id (via its LogRecord factory) but NOT OTel
+# trace_id / span_id -- the observer never activates its spans and this record
+# is emitted outside any span context -- so §7 correlation is via
+# correlation_id. The record names the prompt + the breached bound(s), one per
+# over-budget attempt, independent of the §11 metrics opt-in.
+logger = logging.getLogger("openarmature.observability")
+
 
 # Span-stack key shape:
 # ``(namespace, attempt_index, fan_out_index, branch_name)`` — these
@@ -214,37 +225,6 @@
 ]
 
 
-def _token_budget_evaluations(token_budget: Any, usage: Any) -> list[dict[str, Any]]:
-    # §5.5.15 / §11.2 (proposal 0083): the per-bound token-budget evaluation
-    # shared by the span attrs and the metric recorder. Returns one entry per
-    # DECLARED bound that also has a usable actual to compare against, so a bound
-    # whose count the provider did not report (None) is omitted -- mirroring the
-    # token.usage gate rather than coercing a missing count to 0. A declared
-    # bound of 0 IS evaluated: the exceeded test is a strict ``actual > max``, so
-    # a 0 bound is exceeded by any positive usage per §5.5.15; only the
-    # utilization ratio is undefined for a 0 denominator, and the metric recorder
-    # skips that one sample. Empty when there is no budget, no usage, or no bound
-    # with a reported count; the caller then leaves the exceeded signal absent
-    # (not false).
-    if token_budget is None or usage is None:
-        return []
-    evaluations: list[dict[str, Any]] = []
-    input_max = getattr(token_budget, "input_max_tokens", None)
-    if input_max is not None and usage.prompt_tokens is not None:
-        evaluations.append({"kind": "input", "actual": usage.prompt_tokens, "max": input_max})
-    total_max = getattr(token_budget, "total_max_tokens", None)
-    if total_max is not None:
-        if usage.total_tokens is not None:
-            total_actual = usage.total_tokens
-        elif usage.prompt_tokens is not None and usage.completion_tokens is not None:
-            total_actual = usage.prompt_tokens + usage.completion_tokens
-        else:
-            total_actual = None
-        if total_actual is not None:
-            evaluations.append({"kind": "total", "actual": total_actual, "max": total_max})
-    return evaluations
-
-
 def _read_spec_version() -> str:
     """Read the spec version pinned at package level. Lazy import
     avoids a circular at module-load time (the package's ``__init__``
@@ -836,6 +816,10 @@ async def __call__(
         # LlmRetryAttemptEvent — one span per attempt under call-level
         # retry (attempt_index 0..N-1), one for a no-retry call.
         if isinstance(event, LlmRetryAttemptEvent):
+            # §7 token-budget WARNING log is its own surface: it fires
+            # independently of both the span (disable_llm_spans) and metric
+            # (enable_metrics) gates.
+            self._maybe_log_token_budget_exceedance(event)
             # §11 metrics record per attempt, independent of span emission
             # (§11.1) — so this runs regardless of disable_llm_spans.
             if self.enable_metrics:
@@ -1528,6 +1512,31 @@ def _record_llm_metrics(self, event: LlmRetryAttemptEvent) -> None:
                 if ev["actual"] > ev["max"]:
                     self._token_budget_exceeded_counter.add(1, dims)
 
+    def _maybe_log_token_budget_exceedance(self, event: LlmRetryAttemptEvent) -> None:
+        # §7 (proposal 0083): one vendor-neutral WARNING log record per
+        # over-budget attempt (not per bound). Its own surface -- independent of
+        # disable_llm_spans and enable_metrics -- so it is dispatched here rather
+        # than from the span/metric handlers. Names the prompt + each breached
+        # bound (kind, actual, budget). The OTLP log bridge attaches
+        # correlation_id via its LogRecord factory; it does NOT populate OTel
+        # trace_id / span_id, because the observer never makes its spans current
+        # (start_span with explicit parents) and this record is emitted from the
+        # detached delivery worker, so no span is active at log time. The record
+        # correlates via correlation_id.
+        breached = [
+            ev
+            for ev in _token_budget_evaluations(event.token_budget, event.usage)
+            if ev["actual"] > ev["max"]
+        ]
+        if not breached:
+            return
+        active_prompt = event.active_prompt
+        prompt_ident: Any = active_prompt.name if active_prompt is not None else None
+        if active_prompt is not None and active_prompt.version is not None:
+            prompt_ident = f"{active_prompt.name} {active_prompt.version}"
+        breaches = ", ".join(f"{ev['kind']} {ev['actual']} > {ev['max']}" for ev in breached)
+        logger.warning("token budget exceeded for prompt %r: %s", prompt_ident, breaches)
+
     def _handle_typed_llm_retry_attempt(self, event: LlmRetryAttemptEvent) -> None:
         """Open + close one ``openarmature.llm.complete`` span from a
         per-attempt LlmRetryAttemptEvent.
diff --git a/tests/conformance/test_fixture_parsing.py b/tests/conformance/test_fixture_parsing.py
index 0ae8bce..67bc2af 100644
--- a/tests/conformance/test_fixture_parsing.py
+++ b/tests/conformance/test_fixture_parsing.py
@@ -612,9 +612,8 @@ def _id(case: tuple[str, Path]) -> str:
     ),
     # Proposal 0083 (per-prompt token-budget observability, v0.78.0) -- the
     # completion-path fixtures (126-129) parse + run in test_observability; the
-    # Langfuse WARNING (130) + structured-output-failure (131) paths stay
-    # deferred until their PRs.
-    "observability/130-langfuse-token-budget-warning-level": "Proposal 0083 token-budget; not implemented",
+    # Langfuse WARNING (130) parses + runs via test_observability_langfuse; the
+    # structured-output-failure (131) path stays deferred until its PR.
     "observability/131-token-budget-on-structured-output-failure": (
         "Proposal 0083 token-budget; not implemented"
     ),
diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py
index 35e91af..f3c9a71 100644
--- a/tests/conformance/test_observability_langfuse.py
+++ b/tests/conformance/test_observability_langfuse.py
@@ -41,6 +41,7 @@
     PromptManager,
     SamplingConfig,
     TextPrompt,
+    TokenBudget,
 )
 from openarmature.prompts.context import with_active_prompt
 
@@ -122,6 +123,12 @@
         # on the ERROR failed Generation. Drives the real failure path via
         # calls_llm.response_schema + expected_error.
         "123-langfuse-failed-generation-renders-output-usage-finish-reason",
+        # 130 (proposal 0083): a SUCCESSFUL completion whose prompt_tokens exceed
+        # the declared input_max_tokens renders the advisory WARNING level +
+        # statusMessage naming the breached bound, and maps the declared budget
+        # to metadata.token_budget.*. Drives the input-exceeded path via
+        # renders_prompt + the prompt's token_budget block.
+        "130-langfuse-token-budget-warning-level",
     }
 )
 
@@ -425,6 +432,18 @@ def __init__(self, prompts: dict[str, dict[str, Any]], *, with_langfuse_referenc
             observability_entities: dict[str, Any] | None = None
             if with_langfuse_reference and "langfuse_prompt_reference" in spec:
                 observability_entities = {"langfuse_prompt": spec["langfuse_prompt_reference"]}
+            # Proposal 0083: the prompt's token_budget block propagates onto the
+            # TextPrompt so PromptManager.render carries it to the PromptResult
+            # (and thence the typed event). An empty / all-null budget collapses
+            # to None, mirroring the real backends.
+            token_budget_spec = cast("dict[str, Any] | None", spec.get("token_budget"))
+            token_budget = TokenBudget(**token_budget_spec) if token_budget_spec is not None else None
+            if (
+                token_budget is not None
+                and token_budget.input_max_tokens is None
+                and token_budget.total_max_tokens is None
+            ):
+                token_budget = None
             self._prompts[prompt_name] = TextPrompt(
                 name=spec["name"],
                 version=spec["version"],
@@ -433,6 +452,7 @@ def __init__(self, prompts: dict[str, dict[str, Any]], *, with_langfuse_referenc
                 template_hash=spec["template_hash"],
                 fetched_at=now,
                 observability_entities=observability_entities,
+                token_budget=token_budget,
             )
 
     async def fetch(
diff --git a/tests/unit/test_observability_langfuse.py b/tests/unit/test_observability_langfuse.py
index 90c3af5..bc7d054 100644
--- a/tests/unit/test_observability_langfuse.py
+++ b/tests/unit/test_observability_langfuse.py
@@ -1384,6 +1384,135 @@ async def test_typed_llm_event_emits_generation_with_expected_fields() -> None:
     assert obs.ended is True
 
 
+async def test_typed_llm_completion_over_budget_sets_warning_level() -> None:
+    # §8.4.3 (proposal 0083): a SUCCESSFUL completion whose prompt_tokens exceed
+    # the declared input_max_tokens sets the Generation's advisory WARNING level +
+    # a statusMessage naming the breached bound, and maps the declared budget to
+    # metadata.token_budget. The call succeeded, so the level is WARNING (not
+    # ERROR).
+    from openarmature.llm.response import Usage
+    from openarmature.observability.correlation import (
+        _reset_invocation_id,
+        _set_invocation_id,
+    )
+    from openarmature.prompts import TokenBudget
+    from tests._helpers.typed_event import make_typed_event
+
+    client = InMemoryLangfuseClient()
+    observer = LangfuseObserver(client=client)
+    token = _set_invocation_id("inv-tb-warn")
+    try:
+        await observer(
+            make_typed_event(
+                invocation_id="inv-tb-warn",
+                usage=Usage(prompt_tokens=20, completion_tokens=1, total_tokens=21),
+                token_budget=TokenBudget(input_max_tokens=10),
+            )
+        )
+    finally:
+        _reset_invocation_id(token)
+
+    gen = next(o for o in client.traces["inv-tb-warn"].observations if o.type == "generation")
+    assert gen.level == "WARNING"
+    assert gen.status_message == "token budget exceeded: input 20 > 10"
+    assert gen.metadata.get("token_budget") == {"input_max_tokens": 10}
+
+
+async def test_typed_llm_completion_under_budget_no_warning_level() -> None:
+    # §8.4.3 (proposal 0083): a completion under budget keeps the default level
+    # (no advisory WARNING) while still mapping the declared budget to
+    # metadata.token_budget.
+    from openarmature.llm.response import Usage
+    from openarmature.observability.correlation import (
+        _reset_invocation_id,
+        _set_invocation_id,
+    )
+    from openarmature.prompts import TokenBudget
+    from tests._helpers.typed_event import make_typed_event
+
+    client = InMemoryLangfuseClient()
+    observer = LangfuseObserver(client=client)
+    token = _set_invocation_id("inv-tb-ok")
+    try:
+        await observer(
+            make_typed_event(
+                invocation_id="inv-tb-ok",
+                usage=Usage(prompt_tokens=5, completion_tokens=1, total_tokens=6),
+                token_budget=TokenBudget(input_max_tokens=40),
+            )
+        )
+    finally:
+        _reset_invocation_id(token)
+
+    gen = next(o for o in client.traces["inv-tb-ok"].observations if o.type == "generation")
+    assert gen.level == "DEFAULT"
+    assert gen.status_message is None
+    assert gen.metadata.get("token_budget") == {"input_max_tokens": 40}
+
+
+async def test_typed_llm_completion_both_bounds_breach_statusmessage() -> None:
+    # §8.4.3 (proposal 0083): a completion breaching BOTH bounds names both in the
+    # statusMessage, input then total.
+    from openarmature.llm.response import Usage
+    from openarmature.observability.correlation import (
+        _reset_invocation_id,
+        _set_invocation_id,
+    )
+    from openarmature.prompts import TokenBudget
+    from tests._helpers.typed_event import make_typed_event
+
+    client = InMemoryLangfuseClient()
+    observer = LangfuseObserver(client=client)
+    token = _set_invocation_id("inv-tb-both")
+    try:
+        await observer(
+            make_typed_event(
+                invocation_id="inv-tb-both",
+                usage=Usage(prompt_tokens=20, completion_tokens=10, total_tokens=30),
+                token_budget=TokenBudget(input_max_tokens=10, total_max_tokens=15),
+            )
+        )
+    finally:
+        _reset_invocation_id(token)
+
+    gen = next(o for o in client.traces["inv-tb-both"].observations if o.type == "generation")
+    assert gen.level == "WARNING"
+    assert gen.status_message == "token budget exceeded: input 20 > 10, total 30 > 15"
+    assert gen.metadata.get("token_budget") == {"input_max_tokens": 10, "total_max_tokens": 15}
+
+
+async def test_typed_llm_failed_generation_carries_token_budget_metadata() -> None:
+    # §5.5.15 (proposal 0083): the shared metadata renders token_budget on a
+    # FAILED Generation too (it carries the active prompt's budget). ERROR level
+    # stands -- the advisory WARNING never displaces a hard failure (§8.4.3).
+    from openarmature.observability.correlation import (
+        _reset_invocation_id,
+        _set_invocation_id,
+    )
+    from openarmature.prompts import TokenBudget
+    from tests._helpers.typed_event import make_failed_event
+
+    client = InMemoryLangfuseClient()
+    observer = LangfuseObserver(client=client)
+    token = _set_invocation_id("inv-tb-fail")
+    try:
+        await observer(
+            make_failed_event(
+                invocation_id="inv-tb-fail",
+                error_category="provider_unavailable",
+                error_type="ProviderUnavailable",
+                error_message="503",
+                token_budget=TokenBudget(input_max_tokens=10),
+            )
+        )
+    finally:
+        _reset_invocation_id(token)
+
+    gen = next(o for o in client.traces["inv-tb-fail"].observations if o.type == "generation")
+    assert gen.level == "ERROR"
+    assert gen.metadata.get("token_budget") == {"input_max_tokens": 10}
+
+
 async def test_structured_output_failure_generation_renders_response_surface() -> None:
     # Proposal 0082: a structured_output_invalid failure renders the response-side
     # surface (output payload-gated, usage, metadata.finish_reason) on the ERROR
diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py
index 0a66581..017dd32 100644
--- a/tests/unit/test_observability_otel.py
+++ b/tests/unit/test_observability_otel.py
@@ -1477,6 +1477,146 @@ async def test_token_budget_zero_total_bound_exceeds_but_skips_utilization() ->
     assert util == []
 
 
+async def test_token_budget_exceeded_emits_one_warning_log(caplog: pytest.LogCaptureFixture) -> None:
+    # §7 (proposal 0083): an over-budget attempt emits exactly ONE WARNING record
+    # on the openarmature.observability logger naming the breached bound; an
+    # under-budget attempt emits none. The record is one-per-attempt, not
+    # one-per-bound, and fires independent of the span-attr / metric surfaces.
+    from openarmature.llm.response import Usage
+    from openarmature.prompts import TokenBudget
+    from tests._helpers.typed_event import make_retry_attempt_event
+
+    caplog.set_level(logging.WARNING, logger="openarmature.observability")
+
+    over = make_retry_attempt_event(
+        usage=Usage(prompt_tokens=20, completion_tokens=1, total_tokens=21),
+        token_budget=TokenBudget(input_max_tokens=10),
+    )
+    await _drive_metrics_events([over])
+    warns = [r for r in caplog.records if r.name == "openarmature.observability"]
+    assert len(warns) == 1
+    assert warns[0].levelno == logging.WARNING
+    assert "input 20 > 10" in warns[0].getMessage()
+
+    caplog.clear()
+    under = make_retry_attempt_event(
+        usage=Usage(prompt_tokens=5, completion_tokens=1, total_tokens=6),
+        token_budget=TokenBudget(input_max_tokens=40),
+    )
+    await _drive_metrics_events([under])
+    assert [r for r in caplog.records if r.name == "openarmature.observability"] == []
+
+    # The §7 log is its own surface: it fires even with BOTH the span
+    # (disable_llm_spans) and metric (enable_metrics off) surfaces suppressed.
+    caplog.clear()
+    _, llm_spans = await _drive_metrics_events([over], enable_metrics=False, disable_llm_spans=True)
+    assert llm_spans == []
+    warns = [r for r in caplog.records if r.name == "openarmature.observability"]
+    assert len(warns) == 1
+    assert "input 20 > 10" in warns[0].getMessage()
+
+
+async def test_token_budget_warning_log_names_prompt_identity(caplog: pytest.LogCaptureFixture) -> None:
+    # §7 (proposal 0083): the WARNING log names the active prompt's identity
+    # (name + version) alongside the breached bound.
+    from datetime import UTC, datetime
+
+    from openarmature.llm.messages import UserMessage
+    from openarmature.llm.response import Usage
+    from openarmature.prompts import PromptResult, TokenBudget
+    from tests._helpers.typed_event import make_retry_attempt_event
+
+    caplog.set_level(logging.WARNING, logger="openarmature.observability")
+    now = datetime.now(UTC)
+    result = PromptResult(
+        name="classify",
+        version="v7",
+        label="production",
+        template_hash="sha256:tpl",
+        rendered_hash="sha256:r",
+        messages=[UserMessage(content="x")],
+        variables={},
+        fetched_at=now,
+        rendered_at=now,
+    )
+    event = make_retry_attempt_event(
+        active_prompt=result,
+        usage=Usage(prompt_tokens=20, completion_tokens=1, total_tokens=21),
+        token_budget=TokenBudget(input_max_tokens=10),
+    )
+    await _drive_metrics_events([event])
+    warns = [r for r in caplog.records if r.name == "openarmature.observability"]
+    assert len(warns) == 1
+    msg = warns[0].getMessage()
+    assert "classify v7" in msg
+    assert "input 20 > 10" in msg
+
+
+async def test_token_budget_warning_log_on_structured_output_failure(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # §7 (proposal 0083): the WARNING log fires on an over-budget
+    # structured_output_invalid failure attempt (it carries usage, proposal
+    # 0082), parity with the completion path; a no-usage failure emits none.
+    from openarmature.llm.response import Usage
+    from openarmature.prompts import TokenBudget
+    from tests._helpers.typed_event import make_retry_attempt_event
+
+    caplog.set_level(logging.WARNING, logger="openarmature.observability")
+    failed = make_retry_attempt_event(
+        error_category="structured_output_invalid",
+        finish_reason="length",
+        usage=Usage(prompt_tokens=20, completion_tokens=16, total_tokens=36),
+        token_budget=TokenBudget(input_max_tokens=10),
+    )
+    await _drive_metrics_events([failed])
+    warns = [r for r in caplog.records if r.name == "openarmature.observability"]
+    assert len(warns) == 1
+    assert "input 20 > 10" in warns[0].getMessage()
+
+    # A no-usage failure has nothing to evaluate -> no log.
+    caplog.clear()
+    no_usage = make_retry_attempt_event(
+        error_category="provider_unavailable",
+        finish_reason=None,
+        usage=None,
+        token_budget=TokenBudget(input_max_tokens=10),
+    )
+    await _drive_metrics_events([no_usage])
+    assert [r for r in caplog.records if r.name == "openarmature.observability"] == []
+
+
+async def test_token_budget_warning_log_multibound_and_per_attempt(caplog: pytest.LogCaptureFixture) -> None:
+    # §7 (proposal 0083): a both-bounds breach renders "input .. > .., total .. > .."
+    # in ONE record (input then total order); N over-budget attempts emit N records
+    # (one per exceedance, not per bound).
+    from openarmature.llm.response import Usage
+    from openarmature.prompts import TokenBudget
+    from tests._helpers.typed_event import make_retry_attempt_event
+
+    caplog.set_level(logging.WARNING, logger="openarmature.observability")
+    both = make_retry_attempt_event(
+        usage=Usage(prompt_tokens=20, completion_tokens=10, total_tokens=30),
+        token_budget=TokenBudget(input_max_tokens=10, total_max_tokens=15),
+    )
+    await _drive_metrics_events([both])
+    warns = [r for r in caplog.records if r.name == "openarmature.observability"]
+    assert len(warns) == 1
+    assert "input 20 > 10, total 30 > 15" in warns[0].getMessage()
+
+    caplog.clear()
+    attempts = [
+        make_retry_attempt_event(
+            llm_attempt_index=i,
+            usage=Usage(prompt_tokens=20, completion_tokens=1, total_tokens=21),
+            token_budget=TokenBudget(input_max_tokens=10),
+        )
+        for i in range(2)
+    ]
+    await _drive_metrics_events(attempts)
+    assert len([r for r in caplog.records if r.name == "openarmature.observability"]) == 2
+
+
 async def test_llm_span_zero_duration_when_latency_missing() -> None:
     # When the typed event omits latency_ms (None), the handler falls
     # back to a zero-duration span at end_time rather than guessing