fix(loops): clamp stored reasoning efforts at fire time

products/tasks/backend/logic/services/loop_runs.py

@@ -27,7 +27,10 @@
 from products.tasks.backend.metrics import observe_loop_auto_paused, observe_loop_fire
 from products.tasks.backend.models import Channel, Loop, LoopFire, LoopTrigger, Task, TaskRun
 from products.tasks.backend.temporal.constants import LOOP_RUN_IDLE_TIMEOUT_SECONDS, LOOP_RUN_STALE_SECONDS
-from products.tasks.backend.temporal.process_task.utils import get_default_model_for_runtime_adapter
+from products.tasks.backend.temporal.process_task.utils import (
+    get_default_model_for_runtime_adapter,
+    get_supported_reasoning_efforts,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -529,15 +532,21 @@ def _create_loop_task_and_run(loop: Loop, trigger: LoopTrigger | None, trigger_c
         "repositories": loop.repositories,
         "context_target": context_target,
     }
+    # A loop with no pinned model deliberately stays unset on the row; the
+    # default is resolved per fire so it can improve over time. A stored effort
+    # the resolved model doesn't support (a default change can shrink the
+    # supported set under existing loops) falls back to auto rather than
+    # launching a run the runtime would reject.
+    effective_model = loop.model or get_default_model_for_runtime_adapter(loop.runtime_adapter)
+    supported_efforts = {e.value for e in get_supported_reasoning_efforts(loop.runtime_adapter, effective_model)}
+    reasoning_effort = loop.reasoning_effort if loop.reasoning_effort in supported_efforts else None
     extra_state: dict[str, Any] = {
         "loop_id": str(loop.id),
         "loop_trigger_id": str(trigger.id) if trigger is not None else None,
         "trigger_context": trigger_context,
         "runtime_adapter": loop.runtime_adapter,
-        # A loop with no pinned model deliberately stays unset on the row; the
-        # default is resolved per fire so it can improve over time.
-        "model": loop.model or get_default_model_for_runtime_adapter(loop.runtime_adapter),
-        "reasoning_effort": loop.reasoning_effort,
+        "model": effective_model,
+        "reasoning_effort": reasoning_effort,
         "config_snapshot": config_snapshot,
     }
     # Carries the loop's sandbox secrets/network policy into the run the same way a

products/tasks/backend/temporal/process_task/utils.py

@@ -195,7 +195,7 @@ def get_models_for_runtime_adapter(runtime_adapter: RuntimeAdapter | str | None)
 
 # Applied at fire time when a loop leaves its model unset ("" / None): a blank
 # model means "let PostHog pick", so defaults can improve without rewriting
-# stored loops.
+# stored loops. Mirrored by LOOP_DEFAULT_MODELS in posthog-code's loops UI.
 DEFAULT_MODEL_BY_RUNTIME_ADAPTER: dict[str, str] = {
     RuntimeAdapter.CLAUDE.value: "claude-sonnet-5",
     RuntimeAdapter.CODEX.value: "gpt-5",

products/tasks/backend/tests/test_loop_runs.py

@@ -454,6 +454,37 @@ def test_successful_fire_creates_an_internal_task_with_the_full_config_snapshot(
         self.assertEqual(task_run.state["config_snapshot"]["notifications"], loop.notifications)
         self.assertEqual(task_run.state["config_snapshot"]["repositories"], loop.repositories)
 
+    @parameterized.expand(
+        [
+            ("claude_default_resolves_to_sonnet_5", "claude", "", None, "claude-sonnet-5", None),
+            ("codex_default_resolves_to_gpt5", "codex", "", None, "gpt-5", None),
+            ("supported_effort_on_default_model_is_kept", "claude", "", "high", "claude-sonnet-5", "high"),
+            ("unsupported_effort_on_default_model_falls_back_to_auto", "codex", "", "xhigh", "gpt-5", None),
+            ("pinned_model_keeps_its_supported_effort", "claude", "claude-sonnet-5", "low", "claude-sonnet-5", "low"),
+            (
+                "pinned_model_clamps_unsupported_stored_effort",
+                "claude",
+                "@cf/zai-org/glm-5.2",
+                "low",
+                "@cf/zai-org/glm-5.2",
+                None,
+            ),
+        ]
+    )
+    def test_fire_resolves_model_and_reasoning_effort(
+        self, _name, runtime_adapter, model, reasoning_effort, expected_model, expected_effort
+    ):
+        loop = self.create_loop(runtime_adapter=runtime_adapter, model=model, reasoning_effort=reasoning_effort)
+        trigger = self.create_trigger(loop)
+
+        result = fire_loop(loop, trigger, f"fire-{_name}", "rendered context")
+
+        self.assertTrue(result.created)
+        assert result.task_run_id is not None
+        task_run = TaskRun.objects.get(id=result.task_run_id)
+        self.assertEqual(task_run.state["model"], expected_model)
+        self.assertEqual(task_run.state["reasoning_effort"], expected_effort)
+
     @parameterized.expand(
         [
             ("default_behaviors_are_report_only", {}, {}, False, "read_only"),

products/tasks/backend/tests/test_loops_api.py

@@ -109,6 +109,32 @@ def _create_loop(
 
 
 class LoopCRUDAPITest(LoopsAPITestCase):
+    @parameterized.expand(
+        [
+            ("blank_model_with_effort_the_default_supports", "claude", "", "high", status.HTTP_201_CREATED),
+            ("blank_model_with_effort_the_default_rejects", "codex", "", "xhigh", status.HTTP_400_BAD_REQUEST),
+            ("pinned_glm_with_supported_effort", "claude", "@cf/zai-org/glm-5.2", "max", status.HTTP_201_CREATED),
+            (
+                "pinned_glm_with_unsupported_effort",
+                "claude",
+                "@cf/zai-org/glm-5.2",
+                "medium",
+                status.HTTP_400_BAD_REQUEST,
+            ),
+            ("model_outside_the_adapter_catalog", "claude", "openai/gpt-5.6-sol", None, status.HTTP_400_BAD_REQUEST),
+        ]
+    )
+    def test_create_validates_model_and_reasoning_effort(
+        self, _name, runtime_adapter, model, reasoning_effort, expected_status
+    ):
+        payload = self._valid_loop_payload(
+            runtime_adapter=runtime_adapter, model=model, reasoning_effort=reasoning_effort
+        )
+
+        response = self.owner_client.post(self._loops_url(), payload, format="json")
+
+        self.assertEqual(response.status_code, expected_status, response.content)
+
     def test_create_list_retrieve_update_delete_loop(self):
         payload = self._valid_loop_payload(
             name="Weekly digest",