from decimal import Decimal
from datetime import datetime, timezone as dt_timezone
from pathlib import Path

from django.test import TestCase, tag, override_settings
from django.utils import timezone

from api.models import (
    BrowserConfig,
    BrowserUseAgent,
    BrowserUseAgentTask,
    PersistentAgent,
    PersistentAgentCompletion,
    PersistentAgentCommsEndpoint,
    PersistentAgentError,
    PersistentAgentStep,
    PersistentAgentSystemStep,
    PersistentAgentToolCall,
    TaskCredit,
    CommsChannel,
    UserBilling,
    UserPreference,
)
from django.contrib.auth import get_user_model

from unittest.mock import MagicMock, patch

import uuid
from util.analytics import AnalyticsEvent, AnalyticsSource
from util.constants.task_constants import TASKS_UNLIMITED
from api.agent.core.event_processing import _ensure_credit_for_tool
from api.agent.core.llm_config import (
    AgentLLMTier,
    apply_tier_credit_multiplier,
    clear_runtime_tier_override,
    get_agent_baseline_llm_tier,
    get_agent_llm_tier,
    get_credit_multiplier_for_tier,
    get_next_lower_configured_tier,
    get_runtime_tier_override,
    set_runtime_tier_override,
)
from api.agent.core.prompt_context import (
    add_budget_awareness_sections,
    compute_burn_rate,
    get_agent_daily_credit_state,
)
from constants.grant_types import GrantTypeChoices
from constants.plans import PlanNames
from api.agent.core import event_processing as ep
from api.agent.core import burn_control as bc
from tests.utils.llm_seed import get_intelligence_tier


class _DummySpan:
    def add_event(self, *_args, **_kwargs):
        return None

    def set_attribute(self, *_args, **_kwargs):
        return None


@tag("batch_event_processing_credits")
class PersistentAgentCreditGateTests(TestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls._short_desc_patcher = patch(
            "api.agent.tasks.short_description.generate_agent_short_description_task.delay",
            return_value=None,
        )
        cls._short_desc_patcher.start()
        cls._mini_desc_patcher = patch(
            "api.agent.tasks.mini_description.generate_agent_mini_description_task.delay",
            return_value=None,
        )
        cls._mini_desc_patcher.start()
        cls._tags_patcher = patch(
            "api.agent.tasks.agent_tags.generate_agent_tags_task.delay",
            return_value=None,
        )
        cls._tags_patcher.start()
        cls._avatar_visual_patcher = patch(
            "api.agent.tasks.agent_avatar.generate_agent_visual_description_task.delay",
            return_value=None,
        )
        cls._avatar_visual_patcher.start()
        cls._avatar_patcher = patch(
            "api.agent.tasks.agent_avatar.generate_agent_avatar_task.delay",
            return_value=None,
        )
        cls._avatar_patcher.start()

    @classmethod
    def tearDownClass(cls):
        cls._short_desc_patcher.stop()
        cls._mini_desc_patcher.stop()
        cls._tags_patcher.stop()
        cls._avatar_visual_patcher.stop()
        cls._avatar_patcher.stop()
        super().tearDownClass()

    @classmethod
    def setUpTestData(cls):
        User = get_user_model()
        cls.user = User.objects.create_user(
            username=f"user-{uuid.uuid4()}",
            email=f"user-{uuid.uuid4()}@example.com",
            password="pass1234",
        )

        cls.browser_agent = BrowserUseAgent.objects.create(
            user=cls.user,
            name="BA for PA",
        )

        cls.agent = PersistentAgent.objects.create(
            user=cls.user,
            name="Test Persistent Agent",
            charter="Do useful things",
            browser_use_agent=cls.browser_agent,
        )

    def _grant_credits(self, credits: int, used: int):
        now = timezone.now()
        TaskCredit.objects.create(
            user=self.user,
            credits=credits,
            credits_used=used,
            granted_date=now,
            expiration_date=now + timezone.timedelta(days=30),
            grant_type="Compensation",
        )

    def test_proprietary_mode_out_of_credits_exits_early(self):
        # Force the credit check to report 0 available
        with patch("config.settings.GOBII_PROPRIETARY_MODE", True), patch(
            "api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner",
            return_value=0,
        ):
            # Patch the heavy loop to ensure it would raise if called
            with patch("api.agent.core.event_processing._run_agent_loop") as loop_mock:
                from api.agent.core.event_processing import _process_agent_events_locked

                _process_agent_events_locked(self.agent.id, _DummySpan())

                # Ensure loop never runs due to early exit
                loop_mock.assert_not_called()

        # The early exit creates a SystemStep with PROCESS_EVENTS + credit_insufficient
        sys_steps = PersistentAgentSystemStep.objects.filter(
            step__agent=self.agent,
            code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
        )
        self.assertTrue(sys_steps.exists(), "Expected a system step to be created on early exit")

        notes = list(sys_steps.values_list("notes", flat=True))
        self.assertIn("credit_insufficient", notes)

        # Ensure that no "Process events" description (from normal path) was created
        self.assertFalse(
            self.agent.steps.filter(description="Process events").exists(),
            "Normal event-window step should not be created on early exit",
        )

    def test_signup_preview_first_reply_bypasses_startup_credit_gate(self):
        self.agent.signup_preview_state = (
            PersistentAgent.SignupPreviewState.AWAITING_FIRST_REPLY_PAUSE
        )
        self.agent.save(update_fields=["signup_preview_state", "updated_at"])

        with patch("config.settings.GOBII_PROPRIETARY_MODE", True), patch(
            "api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner",
            return_value=0,
        ) as available_mock, patch(
            "api.agent.core.event_processing._run_agent_loop",
            return_value={},
        ) as loop_mock:
            from api.agent.core.event_processing import _process_agent_events_locked

            _process_agent_events_locked(self.agent.id, _DummySpan())

        available_mock.assert_not_called()
        loop_mock.assert_called_once()
        self.assertFalse(
            PersistentAgentSystemStep.objects.filter(
                step__agent=self.agent,
                code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
                notes="credit_insufficient",
            ).exists()
        )

    def test_owner_execution_pause_exits_early(self):
        UserBilling.objects.update_or_create(
            user=self.user,
            defaults={
                "execution_paused": True,
                "execution_pause_reason": "billing_delinquency",
                "execution_paused_at": timezone.now(),
            },
        )

        with patch("api.agent.core.event_processing._run_agent_loop") as loop_mock:
            from api.agent.core.event_processing import _process_agent_events_locked

            _process_agent_events_locked(self.agent.id, _DummySpan())

            loop_mock.assert_not_called()

        sys_steps = PersistentAgentSystemStep.objects.filter(
            step__agent=self.agent,
            code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
            notes__icontains="owner_execution_paused",
        )
        self.assertTrue(sys_steps.exists(), "Expected a paused-owner system step to be created")
        self.assertFalse(
            self.agent.steps.filter(description="Process events").exists(),
            "Normal event-window step should not be created when owner execution is paused",
        )

    def test_owner_execution_pause_system_step_is_deduped(self):
        UserBilling.objects.update_or_create(
            user=self.user,
            defaults={
                "execution_paused": True,
                "execution_pause_reason": "billing_delinquency",
                "execution_paused_at": timezone.now(),
            },
        )

        with patch("api.agent.core.event_processing._run_agent_loop") as loop_mock:
            from api.agent.core.event_processing import _process_agent_events_locked

            _process_agent_events_locked(self.agent.id, _DummySpan())
            _process_agent_events_locked(self.agent.id, _DummySpan())

            loop_mock.assert_not_called()

        sys_steps = PersistentAgentSystemStep.objects.filter(
            step__agent=self.agent,
            code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
            notes="owner_execution_paused:billing_delinquency",
        )
        self.assertEqual(sys_steps.count(), 1)
        self.assertEqual(
            PersistentAgentStep.objects.filter(
                agent=self.agent,
                description="Skipped processing because account execution is paused.",
            ).count(),
            1,
        )

    def test_proprietary_mode_with_credits_proceeds(self):
        # Give at least one available credit
        self._grant_credits(credits=1, used=0)

        with patch("config.settings.GOBII_PROPRIETARY_MODE", True), \
             patch("api.agent.core.event_processing._run_agent_loop") as loop_mock:
            # Return empty dict for token usage
            loop_mock.return_value = {}
            from api.agent.core.event_processing import _process_agent_events_locked

            _process_agent_events_locked(self.agent.id, _DummySpan())

            # Should proceed into normal path
            loop_mock.assert_called()

        # Should have created the normal PROCESS_EVENTS step (description = "Process events")
        self.assertTrue(
            self.agent.steps.filter(description="Process events").exists(),
            "Expected normal event processing step to be created",
        )

        # And should NOT include the credit_insufficient system note
        self.assertFalse(
            PersistentAgentSystemStep.objects.filter(
                step__agent=self.agent,
                code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
                notes__icontains="credit_insufficient",
            ).exists()
        )

    def test_non_proprietary_mode_skips_gate(self):
        # Even with no available credits, in non-proprietary mode we proceed
        self._grant_credits(credits=100, used=100)

        with patch("config.settings.GOBII_PROPRIETARY_MODE", False), \
             patch("api.agent.core.event_processing._run_agent_loop") as loop_mock:
            # Return empty dict for token usage
            loop_mock.return_value = {}
            from api.agent.core.event_processing import _process_agent_events_locked

            _process_agent_events_locked(self.agent.id, _DummySpan())

            loop_mock.assert_called()

        # No credit_insufficient note expected
        self.assertFalse(
            PersistentAgentSystemStep.objects.filter(
                step__agent=self.agent,
                code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
                notes__icontains="credit_insufficient",
            ).exists()
        )

    def test_daily_limit_canned_notice_removed_from_agent_code(self):
        phrase = "I reached my daily task limit and am not able to continue today."
        search_roots = [
            Path("api/agent/comms"),
            Path("api/agent/core"),
        ]
        matches = [
            str(path)
            for root in search_roots
            for path in root.rglob("*.py")
            if phrase in path.read_text()
        ]

        self.assertEqual(matches, [])

    def test_process_agent_events_enters_message_only_mode_when_daily_limit_reached(self):
        """Processing should continue in message-only mode when the hard daily limit is reached."""
        from api.agent.core.event_processing import _process_agent_events_locked

        self.agent.daily_credit_limit = 1
        self.agent.save(update_fields=["daily_credit_limit"])
        with patch('tasks.services.TaskCreditService.check_and_consume_credit_for_owner', return_value={'success': True, 'credit': None}):
            PersistentAgentStep.objects.create(
                agent=self.agent,
                description="Previously consumed",
                credits_cost=Decimal("2"),
            )

        fake_state = {
            "date": timezone.localdate(),
            "limit": Decimal("1"),
            "soft_target": Decimal("1"),
            "used": Decimal("2"),
            "remaining": Decimal("0"),
            "soft_target_remaining": Decimal("0"),
            "hard_limit": Decimal("2"),
            "hard_limit_remaining": Decimal("0"),
            "next_reset": timezone.now(),
        }

        with override_settings(GOBII_PROPRIETARY_MODE=True), \
             patch("config.settings.GOBII_PROPRIETARY_MODE", True), \
             patch("api.agent.core.event_processing.get_agent_daily_credit_state", return_value=fake_state), \
             patch(
                 "api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner",
                 return_value=0,
             ), \
             patch("api.agent.core.event_processing._run_agent_loop") as loop_mock:
            _process_agent_events_locked(self.agent.id, _DummySpan())
            loop_mock.assert_called_once()
            self.assertEqual(
                loop_mock.call_args.kwargs["credit_snapshot"]["daily_state"]["hard_limit_remaining"],
                Decimal("0"),
            )

        self.assertTrue(
            PersistentAgentSystemStep.objects.filter(
                step__agent=self.agent,
                code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
                notes="daily_credit_limit_exhausted",
            ).exists()
        )

    def test_proprietary_mode_unlimited_allows_processing(self):
        # In proprietary mode, if availability is unlimited (-1), we should proceed
        with patch("config.settings.GOBII_PROPRIETARY_MODE", True), \
             patch("api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner", return_value=TASKS_UNLIMITED), \
             patch("api.agent.core.event_processing._run_agent_loop") as loop_mock:
            # Return empty dict for token usage
            loop_mock.return_value = {}
            from api.agent.core.event_processing import _process_agent_events_locked

            _process_agent_events_locked(self.agent.id, _DummySpan())

            loop_mock.assert_called()
        
        # Ensure no credit_insufficient note was written
        self.assertFalse(
            PersistentAgentSystemStep.objects.filter(
                step__agent=self.agent,
                code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
                notes__icontains="credit_insufficient",
            ).exists()
        )

    def test_process_events_step_without_usage_has_no_completion(self):
        self._grant_credits(credits=1, used=0)
        zero_usage = {
            "prompt_tokens": 0,
            "completion_tokens": 0,
            "total_tokens": 0,
            "cached_tokens": 0,
            "model": None,
            "provider": None,
        }

        with patch("config.settings.GOBII_PROPRIETARY_MODE", True), \
             patch("api.agent.core.event_processing._run_agent_loop", return_value=zero_usage):
            from api.agent.core.event_processing import _process_agent_events_locked

            _process_agent_events_locked(self.agent.id, _DummySpan())

        step = PersistentAgentStep.objects.get(agent=self.agent, description="Process events")
        self.assertIsNone(step.completion)


@tag("batch_event_processing_credits")
class PersistentAgentToolCreditTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        User = get_user_model()
        cls.user = User.objects.create_user(
            username=f"tool-user-{uuid.uuid4()}",
            email=f"tool-user-{uuid.uuid4()}@example.com",
            password="pass1234",
        )

        cls.browser_agent = BrowserUseAgent.objects.create(
            user=cls.user,
            name="Tool BA",
        )

        cls.agent = PersistentAgent.objects.create(
            user=cls.user,
            name="Tool Agent",
            charter="Handle tool credits",
            browser_use_agent=cls.browser_agent,
        )

    def tearDown(self):
        PersistentAgentStep.objects.filter(agent=self.agent).delete()
        PersistentAgentSystemStep.objects.filter(step__agent=self.agent).delete()

    def _create_charged_pending_tool_step(
        self,
        *,
        tool_name: str,
        amount: Decimal = Decimal("2.000"),
    ):
        now = timezone.now()
        credit = TaskCredit.objects.create(
            user=self.user,
            credits=Decimal("10.000"),
            credits_used=amount,
            granted_date=now - timezone.timedelta(days=1),
            expiration_date=now + timezone.timedelta(days=30),
            grant_type=GrantTypeChoices.COMPENSATION,
        )
        completion = PersistentAgentCompletion.objects.create(agent=self.agent)
        step = PersistentAgentStep.objects.create(
            agent=self.agent,
            completion=completion,
            description="",
            credits_cost=amount,
            task_credit=credit,
        )
        PersistentAgentToolCall.objects.create(
            step=step,
            tool_name=tool_name,
            tool_params={},
            result="",
            status="pending",
        )
        completion.refresh_from_db()
        return credit, completion, step

    def _finalize_single_tool_result(
        self,
        *,
        tool_name: str,
        step: PersistentAgentStep,
        credit: TaskCredit,
        amount: Decimal,
        result: dict,
    ):
        prepared = ep._PreparedToolExecution(
            idx=1,
            tool_name=tool_name,
            tool_params={},
            exec_params={},
            pending_step=step,
            credits_consumed=amount,
            consumed_credit=credit,
            call_id="call-refund-test",
            explicit_continue=None,
            inferred_continue=False,
            parallel_safe=False,
            parallel_ineligible_reason=f"unsafe_tool:{tool_name}",
        )
        outcome = ep._ToolExecutionOutcome(
            prepared=prepared,
            result=result,
            duration_ms=10,
            updated_tools=None,
            variable_map={},
        )
        return ep._finalize_tool_batch(
            self.agent,
            [outcome],
            attach_completion=lambda _step_kwargs: None,
            attach_prompt_archive=lambda _step: None,
        )

    @override_settings(TOOL_CREDIT_NO_CHARGE_ON_ERROR_TOOLS={"create_video"})
    def test_allowlisted_tool_error_refunds_credit_and_clears_step_charge(self):
        amount = Decimal("2.000")
        credit, completion, step = self._create_charged_pending_tool_step(
            tool_name="create_video",
            amount=amount,
        )

        finalized = self._finalize_single_tool_result(
            tool_name="create_video",
            step=step,
            credit=credit,
            amount=amount,
            result={"status": "error", "message": "provider failed"},
        )

        self.assertTrue(finalized.followup_required)
        credit.refresh_from_db()
        step.refresh_from_db()
        completion.refresh_from_db()
        tool_call = PersistentAgentToolCall.objects.get(step=step)
        self.assertEqual(credit.credits_used, Decimal("0.000"))
        self.assertIsNone(step.credits_cost)
        self.assertIsNone(step.task_credit_id)
        self.assertIsNone(completion.credits_cost)
        self.assertEqual(tool_call.status, "error")

    @override_settings(TOOL_CREDIT_NO_CHARGE_ON_ERROR_TOOLS={"create_video"})
    def test_non_allowlisted_tool_error_remains_charged(self):
        amount = Decimal("2.000")
        credit, completion, step = self._create_charged_pending_tool_step(
            tool_name="sqlite_query",
            amount=amount,
        )

        self._finalize_single_tool_result(
            tool_name="sqlite_query",
            step=step,
            credit=credit,
            amount=amount,
            result={"status": "error", "message": "query failed"},
        )

        credit.refresh_from_db()
        step.refresh_from_db()
        completion.refresh_from_db()
        tool_call = PersistentAgentToolCall.objects.get(step=step)
        self.assertEqual(credit.credits_used, amount)
        self.assertEqual(step.credits_cost, amount)
        self.assertEqual(step.task_credit_id, credit.id)
        self.assertEqual(completion.credits_cost, amount)
        self.assertEqual(tool_call.status, "error")

    @override_settings(TOOL_CREDIT_NO_CHARGE_ON_ERROR_TOOLS={"create_video"})
    def test_allowlisted_tool_success_remains_charged(self):
        amount = Decimal("2.000")
        credit, completion, step = self._create_charged_pending_tool_step(
            tool_name="create_video",
            amount=amount,
        )

        self._finalize_single_tool_result(
            tool_name="create_video",
            step=step,
            credit=credit,
            amount=amount,
            result={"status": "ok", "file": "$[/exports/video.mp4]"},
        )

        credit.refresh_from_db()
        step.refresh_from_db()
        completion.refresh_from_db()
        tool_call = PersistentAgentToolCall.objects.get(step=step)
        self.assertEqual(credit.credits_used, amount)
        self.assertEqual(step.credits_cost, amount)
        self.assertEqual(step.task_credit_id, credit.id)
        self.assertEqual(completion.credits_cost, amount)
        self.assertEqual(tool_call.status, "complete")

    @patch("api.agent.core.event_processing.settings.GOBII_PROPRIETARY_MODE", True)
    @patch("api.agent.core.event_processing.TaskCreditService.check_and_consume_credit_for_owner")
    @patch("api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner")
    @patch("api.agent.core.event_processing.get_tool_credit_cost", return_value=Decimal("0.8"))
    def test_mid_loop_insufficient_when_cost_exceeds_available(
        self,
        mock_cost,
        mock_available,
        mock_consume,
    ):
        mock_available.return_value = Decimal("0.4")
        span = MagicMock()

        result = _ensure_credit_for_tool(self.agent, "sqlite_query", span=span)

        self.assertIs(result, False)
        mock_consume.assert_not_called()

        step = PersistentAgentStep.objects.get(agent=self.agent)
        self.assertIn("insufficient credits", step.description)
        self.assertTrue(
            PersistentAgentSystemStep.objects.filter(
                step=step,
                notes="credit_insufficient_mid_loop",
            ).exists()
        )

        span.add_event.assert_any_call("Tool skipped - insufficient credits mid-loop")
        span.set_attribute.assert_any_call("credit_check.tool_cost", 0.8)

    @patch("api.agent.core.event_processing.settings.GOBII_PROPRIETARY_MODE", True)
    @patch("api.agent.core.event_processing.TaskCreditService.check_and_consume_credit_for_owner")
    @patch("api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner")
    def test_signup_preview_first_reply_bypasses_tool_credit_consumption(
        self,
        mock_available,
        mock_consume,
    ):
        self.agent.signup_preview_state = (
            PersistentAgent.SignupPreviewState.AWAITING_FIRST_REPLY_PAUSE
        )
        self.agent.save(update_fields=["signup_preview_state", "updated_at"])
        span = MagicMock()

        result = _ensure_credit_for_tool(self.agent, "sqlite_query", span=span)

        self.assertEqual(result, {"cost": None, "credit": None})
        mock_available.assert_not_called()
        mock_consume.assert_not_called()
        span.add_event.assert_any_call("Signup preview credit bypass active")

    @patch("api.agent.core.event_processing.settings.GOBII_PROPRIETARY_MODE", True)
    @patch("api.agent.core.event_processing.TaskCreditService.check_and_consume_credit_for_owner")
    @patch("api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner", return_value=Decimal("1.2"))
    @patch("api.agent.core.event_processing.get_tool_credit_cost", return_value=Decimal("0.8"))
    def test_mid_loop_consumption_exception_records_error(
        self,
        mock_cost,
        _mock_available,
        mock_consume,
    ):
        mock_consume.side_effect = Exception("db down")
        span = MagicMock()

        result = _ensure_credit_for_tool(self.agent, "sqlite_query", span=span)

        self.assertIs(result, False)
        step = PersistentAgentStep.objects.get(agent=self.agent)
        self.assertIn("insufficient credits", step.description)
        self.assertTrue(
            PersistentAgentSystemStep.objects.filter(
                step=step,
                notes="credit_consumption_failure_mid_loop",
            ).exists()
        )

        span.add_event.assert_any_call("Credit consumption raised exception", {"error": "db down"})
        span.add_event.assert_any_call("Tool skipped - insufficient credits during processing")
        span.set_attribute.assert_any_call("credit_check.error", "db down")
        error = PersistentAgentError.objects.get(agent=self.agent)
        self.assertEqual(error.category, PersistentAgentError.Category.CREDIT_FAILURE)
        self.assertEqual(error.exception_class, "Exception")
        self.assertEqual(error.context["operation"], "consume_credit")
        self.assertEqual(error.context["tool_name"], "sqlite_query")
        self.assertEqual(error.context["cost"], "0.800")

    @patch("api.agent.core.event_processing.settings.GOBII_PROPRIETARY_MODE", True)
    @patch(
        "api.agent.core.event_processing.TaskCreditService.check_and_consume_credit_for_owner",
        return_value={"success": True, "credit": None},
    )
    @patch(
        "api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner",
        return_value=Decimal("10"),
    )
    @patch("api.agent.core.event_processing.get_tool_credit_cost", return_value=Decimal("1"))
    def test_soft_target_exceedance_allows_until_hard_limit(
        self,
        _mock_cost,
        _mock_available,
        _mock_consume,
    ):
        self.agent.daily_credit_limit = 5
        self.agent.save(update_fields=["daily_credit_limit"])
        PersistentAgentStep.objects.create(
            agent=self.agent,
            description="Soft target exceeded",
            credits_cost=Decimal("6"),
        )

        span = MagicMock()
        result = _ensure_credit_for_tool(self.agent, "sqlite_query", span=span)

        self.assertIsInstance(result, dict)
        self.assertEqual(result.get("cost"), Decimal("1"))
        self.assertFalse(
            self.agent.steps.filter(description__icontains="Skipped tool").exists(),
            "Soft target exhaustion should not emit a skip step until the hard limit is reached.",
        )

    @patch("api.agent.core.event_processing.Analytics.track_event")
    @patch("api.agent.core.event_processing.settings.GOBII_PROPRIETARY_MODE", True)
    @patch(
        "api.agent.core.event_processing.TaskCreditService.check_and_consume_credit_for_owner",
        return_value={"success": True, "credit": None},
    )
    @patch(
        "api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner",
        return_value=Decimal("10"),
    )
    @patch("api.agent.core.event_processing.get_tool_credit_cost", return_value=Decimal("1"))
    def test_soft_target_exceedance_emits_analytics(
        self,
        _mock_cost,
        _mock_available,
        _mock_consume,
        mock_track_event,
    ):
        self.agent.daily_credit_limit = 5
        self.agent.save(update_fields=["daily_credit_limit"])
        PersistentAgentStep.objects.create(
            agent=self.agent,
            description="Soft target threshold reached",
            credits_cost=Decimal("5"),
        )

        result = _ensure_credit_for_tool(self.agent, "sqlite_query")

        self.assertIsInstance(result, dict)
        self.assertEqual(result.get("cost"), Decimal("1"))
        mock_track_event.assert_called_once()
        kwargs = mock_track_event.call_args.kwargs
        self.assertEqual(kwargs["event"], AnalyticsEvent.PERSISTENT_AGENT_SOFT_LIMIT_EXCEEDED)
        self.assertEqual(kwargs["source"], AnalyticsSource.AGENT)
        self.assertEqual(kwargs["properties"].get("agent_id"), str(self.agent.id))

    @patch("api.agent.core.event_processing.settings.GOBII_PROPRIETARY_MODE", True)
    @patch("api.agent.core.event_processing.TaskCreditService.check_and_consume_credit_for_owner")
    @patch("api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner", return_value=TASKS_UNLIMITED)
    @patch("api.agent.core.event_processing.get_tool_credit_cost", return_value=Decimal("0.8"))
    def test_unlimited_skips_fractional_gate(
        self,
        mock_cost,
        _mock_available,
        mock_consume,
    ):
        mock_consume.return_value = {"success": True, "credit": object()}
        span = MagicMock()

        result = _ensure_credit_for_tool(self.agent, "sqlite_query", span=span)

        self.assertIsInstance(result, dict)
        self.assertEqual(result.get("cost"), Decimal("0.8"))
        self.assertIsNotNone(result.get("credit"))
        mock_consume.assert_called_once()
        span.set_attribute.assert_any_call("credit_check.consumed_in_loop", True)

    @patch("api.agent.core.event_processing.settings.GOBII_PROPRIETARY_MODE", True)
    @patch("api.agent.core.event_processing.TaskCreditService.check_and_consume_credit_for_owner")
    @patch("api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner", return_value=Decimal("5"))
    @patch("api.agent.core.event_processing.get_tool_credit_cost", return_value=Decimal("0.5"))
    def test_mid_loop_daily_limit_blocks_tool(
        self,
        mock_cost,
        _mock_available,
        mock_consume,
    ):
        span = MagicMock()
        self.agent.daily_credit_limit = 1
        self.agent.save(update_fields=["daily_credit_limit"])
        with patch('tasks.services.TaskCreditService.check_and_consume_credit_for_owner', return_value={'success': True, 'credit': None}):
            PersistentAgentStep.objects.create(
                agent=self.agent,
                description="Partial usage",
                credits_cost=Decimal("2.0"),
            )

        result = _ensure_credit_for_tool(self.agent, "sqlite_query", span=span)

        self.assertIs(result, False)
        mock_consume.assert_not_called()
        step = PersistentAgentStep.objects.filter(agent=self.agent).order_by('-created_at').first()
        self.assertIsNotNone(step)
        self.assertIn("daily credit limit", step.description.lower())
        self.assertTrue(
            PersistentAgentSystemStep.objects.filter(
                step=step,
                notes="daily_credit_limit_mid_loop",
            ).exists()
        )
        span.add_event.assert_any_call("Tool skipped - daily credit limit reached")

    @patch("api.agent.core.event_processing.Analytics.track_event")
    @patch("api.agent.core.event_processing.settings.GOBII_PROPRIETARY_MODE", True)
    @patch("api.agent.core.event_processing.TaskCreditService.check_and_consume_credit_for_owner")
    @patch("api.agent.core.event_processing.TaskCreditService.calculate_available_tasks_for_owner", return_value=Decimal("5"))
    @patch("api.agent.core.event_processing.get_tool_credit_cost", return_value=Decimal("0.5"))
    def test_mid_loop_daily_limit_blocks_tool_emits_analytics(
        self,
        mock_cost,
        _mock_available,
        mock_consume,
        mock_track_event,
    ):
        span = MagicMock()
        self.agent.daily_credit_limit = 1
        self.agent.save(update_fields=["daily_credit_limit"])
        with patch('tasks.services.TaskCreditService.check_and_consume_credit_for_owner', return_value={'success': True, 'credit': None}):
            PersistentAgentStep.objects.create(
                agent=self.agent,
                description="Usage to hit limit",
                credits_cost=Decimal("2.0"),
            )

        result = _ensure_credit_for_tool(self.agent, "sqlite_query", span=span)

        self.assertIs(result, False)
        mock_consume.assert_not_called()
        events = [
            call.kwargs
            for call in mock_track_event.call_args_list
        ]
        hard_limit_call = next(
            (kwargs for kwargs in events if kwargs.get("event") == AnalyticsEvent.PERSISTENT_AGENT_HARD_LIMIT_EXCEEDED),
            None,
        )
        self.assertIsNotNone(hard_limit_call, "Expected hard limit analytics event to be emitted")
        self.assertEqual(hard_limit_call["source"], AnalyticsSource.AGENT)
        self.assertEqual(hard_limit_call["properties"].get("agent_id"), str(self.agent.id))
        self.assertEqual(hard_limit_call["properties"].get("hard_limit"), '2.00')
        credits_used_today = hard_limit_call["properties"].get("credits_used_today")
        self.assertIsNotNone(credits_used_today, "credits_used_today should be included in analytics payload")
        self.assertEqual(Decimal(credits_used_today), Decimal("2.0"))
        self.assertEqual(hard_limit_call["properties"].get("hard_limit_remaining"), '0.00')

    def test_compute_burn_rate_no_data_returns_zero(self):
        metrics = compute_burn_rate(self.agent, window_minutes=60)
        self.assertEqual(metrics["burn_rate_per_hour"], Decimal("0"))
        self.assertEqual(metrics["window_total"], Decimal("0"))

    def test_compute_burn_rate_counts_recent_usage(self):
        with patch('tasks.services.TaskCreditService.check_and_consume_credit_for_owner', return_value={'success': True, 'credit': None}):
            step = PersistentAgentStep.objects.create(
                agent=self.agent,
                description="Recent usage",
                credits_cost=Decimal("3.0"),
            )
        PersistentAgentStep.objects.filter(pk=step.pk).update(
            created_at=timezone.now() - timezone.timedelta(minutes=10)
        )
        metrics = compute_burn_rate(self.agent, window_minutes=60)
        self.assertEqual(metrics["burn_rate_per_hour"], Decimal("3"))
        self.assertEqual(metrics["window_total"], Decimal("3"))

    def _build_daily_state_with_inactivity(self, *, inactive_days: int, burn_rate_per_hour: Decimal):
        now = timezone.now()
        self.agent.last_interaction_at = now - timezone.timedelta(days=inactive_days)
        self.agent.created_at = now - timezone.timedelta(days=120)
        self.agent.save(update_fields=["last_interaction_at", "created_at"])

        with patch(
            "api.agent.core.prompt_context.apply_tier_credit_multiplier",
            return_value=Decimal("8"),
        ), patch(
            "api.agent.core.prompt_context.get_daily_credit_settings_for_owner",
            return_value=MagicMock(
                burn_rate_window_minutes=60,
                burn_rate_threshold_per_hour=Decimal("4"),
                offpeak_burn_rate_threshold_per_hour=Decimal("4"),
                burn_rate_threshold_24h=Decimal("0"),
            ),
        ), patch(
            "api.agent.core.prompt_context.compute_burn_rate",
            return_value={
                "burn_rate_per_hour": burn_rate_per_hour,
                "window_minutes": 60,
            },
        ):
            return get_agent_daily_credit_state(self.agent)

    def _build_daily_state_with_timezone(
        self,
        *,
        timezone_name: str | None,
        now_value,
        standard_threshold: Decimal = Decimal("6"),
        offpeak_threshold: Decimal = Decimal("2"),
    ):
        if timezone_name is not None:
            UserPreference.update_known_preferences(
                self.user,
                {UserPreference.KEY_USER_TIMEZONE: timezone_name},
            )
        else:
            UserPreference.update_known_preferences(
                self.user,
                {UserPreference.KEY_USER_TIMEZONE: ""},
            )

        with patch(
            "api.agent.core.prompt_context.apply_tier_credit_multiplier",
            side_effect=lambda _agent, threshold, **_kwargs: threshold,
        ), patch(
            "api.agent.core.prompt_context.get_daily_credit_settings_for_owner",
            return_value=MagicMock(
                burn_rate_window_minutes=60,
                burn_rate_threshold_per_hour=standard_threshold,
                offpeak_burn_rate_threshold_per_hour=offpeak_threshold,
                burn_rate_threshold_24h=Decimal("0"),
            ),
        ), patch(
            "api.agent.core.prompt_context.compute_burn_rate",
            return_value={
                "burn_rate_per_hour": Decimal("1"),
                "window_minutes": 60,
            },
        ), patch(
            "api.agent.core.prompt_context.dj_timezone.now",
            return_value=now_value,
        ):
            return get_agent_daily_credit_state(self.agent)

    def test_daily_credit_state_burn_threshold_no_inactivity_keeps_base(self):
        state = self._build_daily_state_with_inactivity(
            inactive_days=0,
            burn_rate_per_hour=Decimal("1"),
        )

        self.assertEqual(state["burn_rate_inactive_weeks"], 0)
        self.assertEqual(state["burn_rate_base_threshold_per_hour"], Decimal("8"))
        self.assertEqual(state["burn_rate_threshold_per_hour"], Decimal("8.000"))

    def test_daily_credit_state_burn_threshold_one_week_halves(self):
        state = self._build_daily_state_with_inactivity(
            inactive_days=7,
            burn_rate_per_hour=Decimal("5"),
        )

        self.assertEqual(state["burn_rate_inactive_weeks"], 1)
        self.assertEqual(state["burn_rate_base_threshold_per_hour"], Decimal("8"))
        self.assertEqual(state["burn_rate_threshold_per_hour"], Decimal("4.000"))

    def test_daily_credit_state_multi_week_decay_enables_pause(self):
        state = self._build_daily_state_with_inactivity(
            inactive_days=21,
            burn_rate_per_hour=Decimal("6"),
        )

        self.assertEqual(state["burn_rate_inactive_weeks"], 3)
        self.assertEqual(state["burn_rate_base_threshold_per_hour"], Decimal("8"))
        self.assertEqual(state["burn_rate_threshold_per_hour"], Decimal("4.000"))
        self.assertLess(state["burn_rate_threshold_per_hour"], state["burn_rate_per_hour"])
        self.assertLess(state["burn_rate_per_hour"], state["burn_rate_base_threshold_per_hour"])

        with patch(
            "api.agent.core.burn_control.has_recent_user_message",
            return_value=False,
        ), patch("api.agent.core.burn_control.pause_for_burn_rate") as pause_mock:
            should_pause = bc.should_pause_for_burn_rate(
                self.agent,
                budget_ctx=None,
                daily_state=state,
                redis_client=MagicMock(get=MagicMock(return_value=None)),
            )

        self.assertTrue(should_pause)
        pause_mock.assert_called_once()

    def test_daily_credit_state_uses_offpeak_threshold_at_night(self):
        now_value = datetime(2026, 3, 10, 3, 0, tzinfo=dt_timezone.utc)
        state = self._build_daily_state_with_timezone(
            timezone_name="America/New_York",
            now_value=now_value,
        )

        self.assertTrue(state["burn_rate_offpeak_active"])
        self.assertEqual(state["burn_rate_timezone"], "America/New_York")
        self.assertEqual(state["burn_rate_base_threshold_per_hour"], Decimal("2"))

    def test_daily_credit_state_uses_standard_threshold_outside_offpeak(self):
        now_value = datetime(2026, 3, 10, 16, 0, tzinfo=dt_timezone.utc)
        state = self._build_daily_state_with_timezone(
            timezone_name="America/New_York",
            now_value=now_value,
        )

        self.assertFalse(state["burn_rate_offpeak_active"])
        self.assertEqual(state["burn_rate_timezone"], "America/New_York")
        self.assertEqual(state["burn_rate_base_threshold_per_hour"], Decimal("6"))

    def test_daily_credit_state_uses_utc_when_timezone_unset(self):
        now_value = datetime(2026, 3, 10, 23, 0, tzinfo=dt_timezone.utc)
        state = self._build_daily_state_with_timezone(
            timezone_name=None,
            now_value=now_value,
        )

        self.assertTrue(state["burn_rate_offpeak_active"])
        self.assertEqual(state["burn_rate_timezone"], "UTC")
        self.assertEqual(state["burn_rate_base_threshold_per_hour"], Decimal("2"))

    def test_daily_credit_state_offpeak_boundary_hours(self):
        # America/New_York (EDT, UTC-4 on March 10, 2026)
        # 21:59 -> 01:59 UTC, 22:00 -> 02:00 UTC, 05:59 -> 09:59 UTC, 06:00 -> 10:00 UTC
        state_2159 = self._build_daily_state_with_timezone(
            timezone_name="America/New_York",
            now_value=datetime(2026, 3, 10, 1, 59, tzinfo=dt_timezone.utc),
        )
        state_2200 = self._build_daily_state_with_timezone(
            timezone_name="America/New_York",
            now_value=datetime(2026, 3, 10, 2, 0, tzinfo=dt_timezone.utc),
        )
        state_0559 = self._build_daily_state_with_timezone(
            timezone_name="America/New_York",
            now_value=datetime(2026, 3, 10, 9, 59, tzinfo=dt_timezone.utc),
        )
        state_0600 = self._build_daily_state_with_timezone(
            timezone_name="America/New_York",
            now_value=datetime(2026, 3, 10, 10, 0, tzinfo=dt_timezone.utc),
        )

        self.assertFalse(state_2159["burn_rate_offpeak_active"])
        self.assertTrue(state_2200["burn_rate_offpeak_active"])
        self.assertTrue(state_0559["burn_rate_offpeak_active"])
        self.assertFalse(state_0600["burn_rate_offpeak_active"])

    @patch(
        "api.agent.core.prompt_context.get_tool_cost_overview",
        return_value=(Decimal("1"), {"send_email": Decimal("1.2"), "run_sql": Decimal("2.5")}),
    )
    def test_budget_sections_include_soft_target_and_burn_warning(self, _mock_costs):
        critical_group = MagicMock()
        budget_group = MagicMock()
        critical_group.group.return_value = budget_group
        next_reset = timezone.now() + timezone.timedelta(hours=4)
        state = {
            "limit": Decimal("10"),
            "soft_target": Decimal("5"),
            "used": Decimal("4"),
            "remaining": Decimal("6"),
            "soft_target_remaining": Decimal("1"),
            "next_reset": next_reset,
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_window_minutes": 60,
            "burn_rate_threshold_per_hour": Decimal("3"),
        }
        result = add_budget_awareness_sections(
            critical_group,
            current_iteration=2,
            max_iterations=4,
            daily_credit_state=state,
        )
        self.assertTrue(result)
        names = [call.args[0] for call in budget_group.section_text.call_args_list]
        self.assertIn("soft_target_progress", names)
        self.assertNotIn("burn_rate_warning", names)
        self.assertIn("tool_cost_awareness", names)
        soft_call = next(call for call in budget_group.section_text.call_args_list if call.args[0] == "soft_target_progress")
        self.assertIn("Soft target progress", soft_call.args[1])
        tool_call = next(call for call in budget_group.section_text.call_args_list if call.args[0] == "tool_cost_awareness")
        self.assertIn("send_email=1.2", tool_call.args[1])

    @patch(
        "api.agent.core.prompt_context.get_tool_cost_overview",
        return_value=(Decimal("1"), {}),
    )
    def test_budget_sections_do_not_emit_burn_rate_analytics_event(self, _mock_costs):
        critical_group = MagicMock()
        budget_group = MagicMock()
        critical_group.group.return_value = budget_group
        state = {
            "limit": Decimal("10"),
            "soft_target": Decimal("5"),
            "used": Decimal("4"),
            "remaining": Decimal("6"),
            "soft_target_remaining": Decimal("1"),
            "next_reset": timezone.now(),
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_window_minutes": 60,
            "burn_rate_threshold_per_hour": Decimal("3"),
        }
        with patch("api.agent.core.event_processing.Analytics.track_event") as track_mock:
            add_budget_awareness_sections(
                critical_group,
                current_iteration=1,
                max_iterations=2,
                daily_credit_state=state,
                agent=self.agent,
            )
        track_mock.assert_not_called()

    def test_budget_sections_handle_unlimited_soft_target(self):
        critical_group = MagicMock()
        budget_group = MagicMock()
        critical_group.group.return_value = budget_group
        state = {
            "limit": None,
            "soft_target": None,
            "used": Decimal("3"),
            "remaining": None,
            "soft_target_remaining": None,
            "next_reset": timezone.now(),
        }
        result = add_budget_awareness_sections(
            critical_group,
            current_iteration=1,
            max_iterations=0,
            daily_credit_state=state,
        )
        self.assertTrue(result)
        names = [call.args[0] for call in budget_group.section_text.call_args_list]
        self.assertNotIn("soft_target_progress", names)

    def test_budget_sections_include_browser_task_usage(self):
        critical_group = MagicMock()
        budget_group = MagicMock()
        critical_group.group.return_value = budget_group

        config, _ = BrowserConfig.objects.get_or_create(plan_name=PlanNames.FREE)
        config.max_browser_tasks = 2
        config.save()
        TaskCredit.objects.create(
            user=self.user,
            credits=Decimal("2"),
            credits_used=Decimal("0"),
            granted_date=timezone.now(),
            expiration_date=timezone.now() + timezone.timedelta(days=30),
            grant_type=GrantTypeChoices.COMPENSATION,
        )

        BrowserUseAgentTask.objects.create(
            agent=self.browser_agent,
            user=self.user,
            status=BrowserUseAgentTask.StatusChoices.COMPLETED,
            prompt="One",
        )
        BrowserUseAgentTask.objects.create(
            agent=self.browser_agent,
            user=self.user,
            status=BrowserUseAgentTask.StatusChoices.FAILED,
            prompt="Two",
        )

        result = add_budget_awareness_sections(
            critical_group,
            current_iteration=1,
            max_iterations=5,
            daily_credit_state=None,
            agent=self.agent,
        )
        self.assertTrue(result)
        names = [call.args[0] for call in budget_group.section_text.call_args_list]
        self.assertIn("browser_task_usage", names)
        usage_call = next(call for call in budget_group.section_text.call_args_list if call.args[0] == "browser_task_usage")
        self.assertIn("2/2", usage_call.args[1])
        self.assertIn("browser_task_usage_warning", names)

    def test_burn_rate_pause_schedules_follow_up_and_exits_loop(self):
        fake_store: dict[str, str] = {}

        class FakeRedis:
            def get(self, key):
                return fake_store.get(key)

            def set(self, key, value, ex=None):
                fake_store[key] = value
                return True

            def delete(self, key):
                return 1 if fake_store.pop(key, None) is not None else 0

        fake_redis = FakeRedis()
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
        }

        with patch("api.agent.core.event_processing.get_redis_client", return_value=fake_redis), \
             patch("api.agent.core.event_processing.get_agent_daily_credit_state", return_value=burn_state), \
             patch("api.agent.core.event_processing.build_prompt_context") as build_prompt_mock, \
             patch(
                 "api.agent.core.event_processing.process_agent_events_task",
                 create=True,
             ) as follow_up_task, \
             patch("api.agent.core.burn_control.Analytics.track_event") as track_event_mock:
            follow_up_task.apply_async = MagicMock()

            usage = ep._run_agent_loop(
                self.agent,
                is_first_run=True,
                credit_snapshot=None,
                run_sequence_number=1,
            )

        self.assertEqual(usage["total_tokens"], 0)
        build_prompt_mock.assert_not_called()

        follow_up_task.apply_async.assert_called_once()
        _, kwargs = follow_up_task.apply_async.call_args
        self.assertEqual(kwargs["countdown"], ep.BURN_RATE_COOLDOWN_SECONDS)
        token = kwargs["kwargs"]["burn_follow_up_token"]
        self.assertEqual(fake_store.get(bc.burn_follow_up_key(self.agent.id)), token)
        self.assertTrue(fake_store.get(bc.burn_cooldown_key(self.agent.id)))

        self.assertTrue(
            PersistentAgentSystemStep.objects.filter(
                step__agent=self.agent,
                code=PersistentAgentSystemStep.Code.BURN_RATE_COOLDOWN,
            ).exists()
        )
        track_event_mock.assert_called_once()
        track_kwargs = track_event_mock.call_args.kwargs
        self.assertEqual(track_kwargs["user_id"], self.user.id)
        self.assertEqual(track_kwargs["event"], AnalyticsEvent.PERSISTENT_AGENT_BURN_RATE_LIMIT_REACHED)

    def test_burn_rate_pause_schedules_follow_up_when_next_cron_is_within_cooldown(self):
        fake_store: dict[str, str] = {}

        class FakeRedis:
            def get(self, key):
                return fake_store.get(key)

            def set(self, key, value, ex=None):
                fake_store[key] = value
                return True

            def delete(self, key):
                return 1 if fake_store.pop(key, None) is not None else 0

        fake_redis = FakeRedis()
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
        }

        self.agent.schedule = "@hourly"
        self.agent.save(update_fields=["schedule"])

        with patch("api.agent.core.event_processing.get_redis_client", return_value=fake_redis), \
             patch("api.agent.core.event_processing.get_agent_daily_credit_state", return_value=burn_state), \
             patch("api.agent.core.event_processing.build_prompt_context") as build_prompt_mock, \
             patch(
                 "api.agent.core.event_processing.process_agent_events_task",
                 create=True,
             ) as follow_up_task:
            follow_up_task.apply_async = MagicMock()

            usage = ep._run_agent_loop(
                self.agent,
                is_first_run=True,
                credit_snapshot=None,
                run_sequence_number=1,
            )

        self.assertEqual(usage["total_tokens"], 0)
        build_prompt_mock.assert_not_called()
        follow_up_task.apply_async.assert_called_once()
        _, kwargs = follow_up_task.apply_async.call_args
        self.assertEqual(kwargs["countdown"], ep.BURN_RATE_COOLDOWN_SECONDS)
        token = kwargs["kwargs"]["burn_follow_up_token"]
        self.assertEqual(fake_store.get(bc.burn_follow_up_key(self.agent.id)), token)

    def test_burn_rate_pause_skips_follow_up_when_next_cron_is_after_cooldown_but_soon(self):
        fake_store: dict[str, str] = {}

        class FakeRedis:
            def get(self, key):
                return fake_store.get(key)

            def set(self, key, value, ex=None):
                fake_store[key] = value
                return True

            def delete(self, key):
                return 1 if fake_store.pop(key, None) is not None else 0

        fake_redis = FakeRedis()
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
        }
        next_run = timezone.now() + timezone.timedelta(minutes=90)

        with patch("api.agent.core.event_processing.get_redis_client", return_value=fake_redis), \
             patch("api.agent.core.event_processing.get_agent_daily_credit_state", return_value=burn_state), \
             patch("api.agent.core.event_processing.build_prompt_context") as build_prompt_mock, \
             patch("api.agent.core.burn_control._next_scheduled_run", return_value=next_run), \
             patch(
                 "api.agent.core.event_processing.process_agent_events_task",
                 create=True,
             ) as follow_up_task:
            follow_up_task.apply_async = MagicMock()

            usage = ep._run_agent_loop(
                self.agent,
                is_first_run=True,
                credit_snapshot=None,
                run_sequence_number=1,
            )

        self.assertEqual(usage["total_tokens"], 0)
        build_prompt_mock.assert_not_called()
        follow_up_task.apply_async.assert_not_called()
        self.assertIsNone(fake_store.get(bc.burn_follow_up_key(self.agent.id)))

    def test_burn_rate_rechecks_fresh_daily_state_each_iteration(self):
        fake_store: dict[str, str] = {}

        class FakeRedis:
            def get(self, key):
                return fake_store.get(key)

            def set(self, key, value, ex=None):
                fake_store[key] = value
                return True

            def delete(self, key):
                return 1 if fake_store.pop(key, None) is not None else 0

        fake_redis = FakeRedis()
        burn_states = [
            {
                "burn_rate_per_hour": Decimal("1"),
                "burn_rate_threshold_per_hour": Decimal("3"),
                "burn_rate_window_minutes": 60,
            },
            {
                "burn_rate_per_hour": Decimal("5"),
                "burn_rate_threshold_per_hour": Decimal("3"),
                "burn_rate_window_minutes": 60,
            },
        ]

        def next_burn_state(*_args, **_kwargs):
            if burn_states:
                return burn_states.pop(0)
            return {
                "burn_rate_per_hour": Decimal("5"),
                "burn_rate_threshold_per_hour": Decimal("3"),
                "burn_rate_window_minutes": 60,
            }

        response = MagicMock()
        # Use "let me" continuation signal so loop continues past no-tool streak limit
        response.choices = [MagicMock(message=MagicMock(content="Let me think about this", tool_calls=[], function_call=None))]
        token_usage = {
            "prompt_tokens": 0,
            "completion_tokens": 0,
            "total_tokens": 0,
            "cached_tokens": 0,
        }

        with patch("api.agent.core.event_processing.get_redis_client", return_value=fake_redis), \
             patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 2), \
             patch("api.agent.core.event_processing.get_agent_daily_credit_state", side_effect=next_burn_state) as burn_state_fn, \
             patch("api.agent.core.event_processing.build_prompt_context", return_value=([], 0, None)), \
             patch("api.agent.core.event_processing.get_llm_config_with_failover", return_value=[{}]), \
             patch("api.agent.core.event_processing._completion_with_failover", return_value=(response, token_usage)), \
             patch("tasks.services.TaskCreditService.check_and_consume_credit_for_owner", return_value={"success": True, "credit": None}), \
             patch(
                 "api.agent.core.event_processing.process_agent_events_task",
                 create=True,
             ) as follow_up_task:
            follow_up_task.apply_async = MagicMock()

            usage = ep._run_agent_loop(
                self.agent,
                is_first_run=True,
                credit_snapshot=None,
                run_sequence_number=1,
            )

        self.assertEqual(usage["total_tokens"], 0)
        self.assertEqual(burn_state_fn.call_count, 2)
        follow_up_task.apply_async.assert_called_once()
        self.assertTrue(fake_store.get(bc.burn_cooldown_key(self.agent.id)))
        self.assertTrue(
            PersistentAgentSystemStep.objects.filter(
                step__agent=self.agent,
                code=PersistentAgentSystemStep.Code.BURN_RATE_COOLDOWN,
            ).exists()
        )

    def test_runtime_tier_step_down_activates_once_with_recent_user_message(self):
        self.agent.preferred_llm_tier = get_intelligence_tier("ultra_max")
        self.agent.save(update_fields=["preferred_llm_tier"])
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
        }

        try:
            with override_settings(GOBII_PROPRIETARY_MODE=True), \
                 patch("api.agent.core.llm_config.get_owner_plan", return_value={"id": "pro"}), \
                 patch("api.agent.core.burn_control.has_recent_user_message", return_value=True), \
                 patch("api.agent.core.burn_control.Analytics.track_event") as track_event_mock:
                stepped = bc.maybe_step_down_runtime_tier_for_burn_rate(
                    self.agent,
                    daily_state=burn_state,
                    span=_DummySpan(),
                )
                stepped_again = bc.maybe_step_down_runtime_tier_for_burn_rate(
                    self.agent,
                    daily_state=burn_state,
                    span=_DummySpan(),
                )

                self.assertTrue(stepped)
                self.assertFalse(stepped_again)
                self.assertEqual(get_agent_baseline_llm_tier(self.agent), AgentLLMTier.ULTRA_MAX)
                self.assertEqual(get_runtime_tier_override(self.agent), AgentLLMTier.ULTRA)
                self.assertEqual(get_agent_llm_tier(self.agent), AgentLLMTier.ULTRA)
                self.assertFalse(
                    PersistentAgentSystemStep.objects.filter(step__agent=self.agent).exists()
                )
                self.assertFalse(
                    PersistentAgentStep.objects.filter(
                        agent=self.agent,
                        description__contains="baseline_tier=ultra_max;runtime_tier=ultra",
                    ).exists()
                )
                track_event_mock.assert_called_once()
                self.assertEqual(
                    track_event_mock.call_args.kwargs["event"],
                    AnalyticsEvent.PERSISTENT_AGENT_BURN_RATE_RUNTIME_TIER_STEPPED_DOWN,
                )
        finally:
            clear_runtime_tier_override(self.agent)

    def test_burn_rate_pause_preserves_hourly_only_when_24h_threshold_disabled(self):
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
            "burn_rate_24h_total": Decimal("1"),
            "burn_rate_threshold_24h": Decimal("0"),
        }

        with patch("api.agent.core.burn_control.has_recent_user_message", return_value=False), \
             patch("api.agent.core.burn_control.pause_for_burn_rate") as pause_mock:
            action = bc.handle_burn_rate_limit(
                self.agent,
                budget_ctx=None,
                daily_state=burn_state,
                redis_client=MagicMock(get=MagicMock(return_value=None)),
            )

        self.assertEqual(action, bc.BurnRateAction.PAUSED)
        pause_mock.assert_called_once()

    def test_burn_rate_gate_skips_action_when_24h_total_below_threshold(self):
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
            "burn_rate_24h_total": Decimal("50"),
            "burn_rate_threshold_24h": Decimal("100"),
        }

        with patch("api.agent.core.burn_control.has_recent_user_message") as recent_mock, \
             patch("api.agent.core.burn_control.pause_for_burn_rate") as pause_mock:
            action = bc.handle_burn_rate_limit(
                self.agent,
                budget_ctx=None,
                daily_state=burn_state,
                redis_client=MagicMock(get=MagicMock(return_value=None)),
            )

        self.assertEqual(action, bc.BurnRateAction.NONE)
        recent_mock.assert_not_called()
        pause_mock.assert_not_called()

    def test_burn_rate_pause_requires_24h_total_above_configured_threshold(self):
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
            "burn_rate_24h_total": Decimal("101"),
            "burn_rate_threshold_24h": Decimal("100"),
        }

        with patch("api.agent.core.burn_control.has_recent_user_message", return_value=False), \
             patch("api.agent.core.burn_control.pause_for_burn_rate") as pause_mock:
            action = bc.handle_burn_rate_limit(
                self.agent,
                budget_ctx=None,
                daily_state=burn_state,
                redis_client=MagicMock(get=MagicMock(return_value=None)),
            )

        self.assertEqual(action, bc.BurnRateAction.PAUSED)
        pause_mock.assert_called_once()
        self.assertEqual(pause_mock.call_args.kwargs["burn_24h_total"], Decimal("101"))
        self.assertEqual(pause_mock.call_args.kwargs["burn_24h_threshold"], Decimal("100"))

    def test_runtime_tier_step_down_requires_24h_total_above_configured_threshold(self):
        self.agent.preferred_llm_tier = get_intelligence_tier("ultra_max")
        self.agent.save(update_fields=["preferred_llm_tier"])
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
            "burn_rate_24h_total": Decimal("101"),
            "burn_rate_threshold_24h": Decimal("100"),
        }

        try:
            with override_settings(GOBII_PROPRIETARY_MODE=True), \
                 patch("api.agent.core.llm_config.get_owner_plan", return_value={"id": "pro"}), \
                 patch("api.agent.core.burn_control.has_recent_user_message", return_value=True):
                action = bc.handle_burn_rate_limit(
                    self.agent,
                    budget_ctx=None,
                    daily_state=burn_state,
                )

            self.assertEqual(action, bc.BurnRateAction.STEPPED_DOWN)
            self.assertEqual(get_runtime_tier_override(self.agent), AgentLLMTier.ULTRA)
        finally:
            clear_runtime_tier_override(self.agent)

    def test_runtime_tier_override_does_not_change_daily_burn_threshold(self):
        self.agent.preferred_llm_tier = get_intelligence_tier("ultra_max")
        self.agent.save(update_fields=["preferred_llm_tier"])

        try:
            with override_settings(GOBII_PROPRIETARY_MODE=True), \
                 patch("api.agent.core.llm_config.get_owner_plan", return_value={"id": "pro"}), \
                 patch(
                     "api.agent.core.prompt_context.get_daily_credit_settings_for_owner",
                     return_value=MagicMock(
                         burn_rate_window_minutes=60,
                         burn_rate_threshold_per_hour=Decimal("4"),
                         offpeak_burn_rate_threshold_per_hour=Decimal("4"),
                         burn_rate_threshold_24h=Decimal("40"),
                     ),
                 ), \
                 patch(
                     "api.agent.core.prompt_context.compute_burn_rate",
                     return_value={
                         "burn_rate_per_hour": Decimal("1"),
                         "window_minutes": 60,
                         "window_total": Decimal("1"),
                     },
                 ):
                set_runtime_tier = get_next_lower_configured_tier(get_agent_baseline_llm_tier(self.agent))
                self.assertEqual(set_runtime_tier, AgentLLMTier.ULTRA)
                set_runtime_tier_override(self.agent, set_runtime_tier)

                state = get_agent_daily_credit_state(self.agent)
                baseline_threshold = apply_tier_credit_multiplier(
                    self.agent,
                    Decimal("4"),
                    use_runtime_override=False,
                )
                runtime_threshold = apply_tier_credit_multiplier(self.agent, Decimal("4"))

                self.assertEqual(state["burn_rate_threshold_per_hour"], baseline_threshold)
                self.assertNotEqual(runtime_threshold, baseline_threshold)
                self.assertEqual(
                    state["burn_rate_threshold_24h"],
                    apply_tier_credit_multiplier(
                        self.agent,
                        Decimal("40"),
                        use_runtime_override=False,
                    ),
                )
        finally:
            clear_runtime_tier_override(self.agent)

    def test_run_loop_uses_downgraded_runtime_tier_after_burn_step_down(self):
        self.agent.preferred_llm_tier = get_intelligence_tier("ultra_max")
        self.agent.save(update_fields=["preferred_llm_tier"])
        burn_state = {
            "burn_rate_per_hour": Decimal("5"),
            "burn_rate_threshold_per_hour": Decimal("3"),
            "burn_rate_window_minutes": 60,
        }
        response = MagicMock()
        response.choices = [
            MagicMock(message=MagicMock(content="Done for now", tool_calls=[], function_call=None))
        ]
        token_usage = {
            "prompt_tokens": 0,
            "completion_tokens": 0,
            "total_tokens": 0,
            "cached_tokens": 0,
        }
        observed = {}

        def capture_failover(*_args, **_kwargs):
            observed["routing_tier"] = get_agent_llm_tier(self.agent)
            observed["runtime_multiplier"] = get_credit_multiplier_for_tier(observed["routing_tier"])
            return [{}]

        with override_settings(GOBII_PROPRIETARY_MODE=True), \
             patch("api.agent.core.llm_config.get_owner_plan", return_value={"id": "pro"}), \
             patch("api.agent.core.burn_control.has_recent_user_message", return_value=True), \
             patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1), \
             patch("api.agent.core.event_processing.get_agent_daily_credit_state", return_value=burn_state), \
             patch("api.agent.core.event_processing.build_prompt_context", return_value=([], 0, None)), \
             patch("api.agent.core.event_processing.get_llm_config_with_failover", side_effect=capture_failover), \
             patch("tasks.services.TaskCreditService.check_and_consume_credit_for_owner", return_value={"success": True, "credit": None}), \
             patch("api.agent.core.event_processing._completion_with_failover", return_value=(response, token_usage)):
            usage = ep._run_agent_loop(
                self.agent,
                is_first_run=False,
                credit_snapshot=None,
                run_sequence_number=1,
            )

        self.assertEqual(usage["total_tokens"], 0)
        self.assertEqual(observed["routing_tier"], AgentLLMTier.ULTRA)
        self.assertEqual(get_runtime_tier_override(self.agent), None)
        self.assertFalse(
            PersistentAgentSystemStep.objects.filter(step__agent=self.agent).exists()
        )
        self.assertFalse(
            PersistentAgentStep.objects.filter(
                agent=self.agent,
                description__contains="baseline_tier=ultra_max;runtime_tier=ultra",
            ).exists()
        )
        self.assertFalse(
            PersistentAgentSystemStep.objects.filter(
                step__agent=self.agent,
                code=PersistentAgentSystemStep.Code.BURN_RATE_COOLDOWN,
            ).exists()
        )
