feat(code): warn above configurable session cost threshold

libs/code/deepagents_code/app.py

@@ -3937,6 +3937,33 @@ def __init__(
         estimates here.
         """
 
+        from deepagents_code.config_manifest import (
+            SESSION_COST_WARNING_THRESHOLD_USD_DEFAULT,
+            get_option,
+            load_config_toml,
+            resolve_scalar,
+        )
+
+        cost_warning_option = get_option("warnings.session_cost_threshold_usd")
+        cost_warning_threshold: object = SESSION_COST_WARNING_THRESHOLD_USD_DEFAULT
+        if cost_warning_option is not None:
+            cost_warning_threshold, _ = resolve_scalar(
+                cost_warning_option, toml_data=load_config_toml()
+            )
+        if not isinstance(cost_warning_threshold, float) or not math.isfinite(
+            cost_warning_threshold
+        ):
+            logger.warning(
+                "Invalid session cost warning threshold %r; using the default",
+                cost_warning_threshold,
+            )
+            cost_warning_threshold = SESSION_COST_WARNING_THRESHOLD_USD_DEFAULT
+        self._session_cost_warning_threshold_usd = cost_warning_threshold
+        """Configured soft limit for the active thread's estimated cost."""
+
+        self._session_cost_warning_shown = False
+        """Whether the active thread has already crossed its cost soft limit."""
+
         self._provisional_cost_usd: float = 0.0
         """Streamed spend the graph has not reported a total for yet.
 
@@ -7619,6 +7646,21 @@ def _set_session_cost(
         self._session_cost_usd = _coerce_session_cost_usd(cost_usd)
         self._provisional_cost_usd = 0.0
         self._refresh_session_cost_display()
+        threshold = self._session_cost_warning_threshold_usd
+        if (
+            not self._session_cost_warning_shown
+            and 0 < threshold < self._session_cost_usd
+        ):
+            self._session_cost_warning_shown = True
+            self.notify(
+                f"Estimated session cost is {format_cost(self._session_cost_usd)}, "
+                f"above the configured {format_cost(threshold)} threshold. Consider "
+                "/offload to reduce context usage or /clear to start fresh.",
+                title="Session cost warning",
+                severity="warning",
+                timeout=12,
+                markup=False,
+            )
 
     @property
     def _displayed_cost_usd(self) -> float:
@@ -7648,6 +7690,7 @@ def _reset_thread_usage(
             has_restored_model_usage or self._thread_restored_cost_usd > 0
         )
         self._thread_has_completed_turn = False
+        self._session_cost_warning_shown = False
         self._set_session_cost(self._thread_restored_cost_usd)
 
     def _mark_thread_turn_completed(self) -> None:

libs/code/deepagents_code/config_manifest.py

@@ -114,6 +114,9 @@
 Zero or negative disables the suggestion.
 """
 
+SESSION_COST_WARNING_THRESHOLD_USD_DEFAULT = 50.0
+"""Default warning threshold in USD; zero or negative disables the warning."""
+
 LANGSMITH_PROJECT_DEFAULT = "deepagents-code"
 """Project agent traces fall back to when no project env var is set.
 
@@ -1540,6 +1543,16 @@ def _credential_options() -> tuple[ConfigOption, ...]:
         toml_keys=("threads", "columns"),
     ),
     # --- Warnings ------------------------------------------------------
+    ConfigOption(
+        key="warnings.session_cost_threshold_usd",
+        group="Warnings",
+        summary=(
+            "Warn once when estimated thread cost exceeds this USD amount (0 disables)."
+        ),
+        kind=OptionKind.FLOAT,
+        default=SESSION_COST_WARNING_THRESHOLD_USD_DEFAULT,
+        toml_keys=("warnings", "session_cost_threshold_usd"),
+    ),
     ConfigOption(
         key="warnings.suppress",
         group="Warnings",

libs/code/tests/unit_tests/test_resume_state.py

@@ -340,6 +340,33 @@ def test_server_total_replaces_the_displayed_value(self) -> None:
         assert app._session_cost_usd == pytest.approx(1.25)
         assert app._displayed_cost_usd == pytest.approx(1.25)
 
+    def test_cost_threshold_warns_once_per_thread(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(
+            "deepagents_code.config_manifest.load_config_toml",
+            lambda: {"warnings": {"session_cost_threshold_usd": 1.0}},
+        )
+        app = DeepAgentsApp()
+        notifications: list[str] = []
+        monkeypatch.setattr(
+            app,
+            "notify",
+            lambda message, **_: notifications.append(message),
+        )
+
+        app._set_session_cost(1.0)
+        app._set_session_cost(1.01)
+        app._set_session_cost(2.0)
+
+        assert len(notifications) == 1
+        assert "$1.01" in notifications[0]
+        assert "/offload" in notifications[0]
+        assert "/clear" in notifications[0]
+
+        app._reset_thread_usage(1.5)
+        assert len(notifications) == 2
+
     def test_streamed_estimate_shows_ahead_of_the_server_total(self) -> None:
         """Spend the graph has not reported yet still moves the display."""
         app = DeepAgentsApp()