import json
import re
import sqlite3
import shutil
import tempfile
from datetime import timedelta
from decimal import Decimal
from types import SimpleNamespace
from django.contrib import admin as django_admin
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.core.files.storage import FileSystemStorage
from django.db import DatabaseError
from django.test import RequestFactory, TestCase, tag, override_settings
from django.utils import timezone
from unittest.mock import patch, MagicMock, ANY
from waffle.testutils import override_switch

import zstandard as zstd
from allauth.account.models import EmailAddress

from api.agent.core.event_processing import (
    OrchestratorPromptStale,
    _completion_with_failover,
    _execute_tool_call_runtime,
    _execute_prepared_tool_call,
    _execute_prepared_tool_batch,
    _finalize_tool_batch,
    _latest_inbound_message_needs_reply,
    _should_continue_for_pending_progress_reply,
    _should_continue_for_unanswered_inbound_after_tools,
    _gate_send_chat_tool_for_delivery,
    _tool_definition_names_for_completion,
    _persist_tool_call_step,
    build_prompt_context,
    _get_completed_process_run_count,
    _FinalizedToolBatch,
    _PreparedToolBatch,
    _PreparedToolExecution,
    _ToolExecutionOutcome,
    _run_agent_loop,
    process_agent_events,
)
from api.agent.core.llm_utils import EmptyLiteLLMResponseError
from api.agent.core.processing_flags import (
    PendingDrainSettings,
    _human_inbound_generation_key,
    bump_human_inbound_generation,
    clear_processing_stop_requested,
    get_consumed_human_inbound_generation,
    get_human_inbound_generation,
    is_human_inbound_generation_consumed,
    is_processing_stop_requested,
    mark_human_inbound_generation_consumed,
    set_processing_stop_requested,
    set_processing_queued_flag,
)
from tests.utils.llm_seed import get_intelligence_tier
from api.agent.comms.adapters import ParsedMessage
from api.agent.comms.human_input_requests import submit_human_input_responses_batch
from api.agent.comms.message_service import ingest_inbound_message, ingest_inbound_webhook_message
from api.agent.peer_comm import PeerMessagingService
from config.redis_client import get_redis_client
from api.agent.core.internal_reasoning import (
    INTERNAL_REASONING_PREFIX,
    build_internal_reasoning_description,
)
from api.agent.core.prompt_context import (
    get_agent_tools,
    get_prompt_token_budget,
    message_history_limit,
    skill_prompt_limit,
    tool_call_history_limit,
)
from api.admin import PersistentAgentPromptArchiveAdmin, PromptConfigAdmin, ToolConfigAdmin
from api.agent.tools.schedule_updater import execute_update_schedule as _execute_update_schedule
from api.agent.tools.http_request import execute_http_request as _execute_http_request
from api.agent.tools.tool_manager import enable_tools
from api.agent.tools.sqlite_state import reset_sqlite_db_path, set_sqlite_db_path
from api.agent.tasks.process_events import (
    process_agent_cron_trigger_task,
    process_agent_events_task,
    _remove_orphaned_celery_beat_task,
)
from api.models import (
    CommsChannel,
    BrowserUseAgent,
    BrowserUseAgentTask,
    BrowserUseAgentTaskStep,
    AgentPeerLink,
    MCPServerConfig,
    Organization,
    OrganizationMembership,
    ProxyServer,
    build_web_agent_address,
    build_web_user_address,
    PersistentAgent,
    PersistentAgentCommsEndpoint,
    PersistentAgentConversation,
    PersistentAgentConversationParticipant,
    PersistentAgentMessage,
    PersistentAgentSkill,
    PersistentAgentStep,
    PersistentAgentCronTrigger,
    PersistentAgentSecret,
    PersistentAgentPromptArchive,
    PersistentAgentCompletion,
    PersistentAgentError,
    PersistentAgentHumanInputRequest,
    PersistentAgentInboundWebhook,
    PersistentAgentSystemStep,
    PersistentAgentSystemMessage,
    PersistentAgentToolCall,
    PersistentAgentSystemSkillState,
    PromptConfig,
    TaskCredit,
    UserBilling,
    ToolConfig,
)
from constants.grant_types import GrantTypeChoices
from constants.plans import PlanNamesChoices
from api.agent.core.llm_config import AgentLLMTier
from api.agent.core.burn_control import BurnRateAction
from api.services.prompt_settings import get_prompt_settings, invalidate_prompt_settings_cache
from api.services.tool_settings import (
    DEFAULT_MIN_CRON_SCHEDULE_MINUTES,
    get_tool_settings_for_plan,
    invalidate_tool_settings_cache,
)
from api.services.web_sessions import start_web_session
from api.services.signup_preview import is_signup_preview_processing_paused
from util.personal_signup_preview import GENERIC_STARTER_CHARTER
from tests.utils.token_usage import make_completion_response

User = get_user_model()


@tag("batch_event_processing")
class PromptContextBuilderTests(TestCase):
    """Unit tests for `build_prompt_context`."""

    def setUp(self):
        self.addCleanup(invalidate_prompt_settings_cache)
        self.user = User.objects.create_user(
            username="prompt_tester@example.com",
            email="prompt_tester@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="PromptBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="PromptAgent",
            charter="Test prompt context",
            browser_use_agent=self.browser_agent,
        )
        self.endpoint = PersistentAgentCommsEndpoint.objects.create(
            owner_agent=self.agent,
            channel="email",
            address="agent@example.com",
            is_primary=True,
        )
        self.external_endpoint = PersistentAgentCommsEndpoint.objects.create(
            channel="email",
            address="user@example.com",
        )
        now = timezone.now()
        existing_credit = TaskCredit.objects.filter(
            user=self.user,
            plan=PlanNamesChoices.FREE,
            grant_type=GrantTypeChoices.PLAN,
            additional_task=False,
            voided=False,
        ).order_by("-granted_date", "-pk").first()
        credit_defaults = {
            "credits": Decimal("50"),
            "credits_used": Decimal("0"),
            "granted_date": now,
            "expiration_date": now + timedelta(days=30),
            "plan": PlanNamesChoices.FREE,
            "grant_type": GrantTypeChoices.PLAN,
            "additional_task": False,
            "voided": False,
        }
        if existing_credit is None:
            TaskCredit.objects.create(user=self.user, **credit_defaults)
        else:
            TaskCredit.objects.filter(pk=existing_credit.pk).update(**credit_defaults)
        self._storage_dir = tempfile.mkdtemp()
        self._storage = FileSystemStorage(location=self._storage_dir)
        self._storage_patch = patch('api.agent.core.prompt_context.default_storage', self._storage)
        self._admin_storage_patch = patch('api.admin.default_storage', self._storage)
        self._print_patch = patch('api.agent.core.prompt_context.print')
        self._storage_patch.start()
        self._admin_storage_patch.start()
        self._print_patch.start()
        self.addCleanup(self._storage_patch.stop)
        self.addCleanup(self._admin_storage_patch.stop)
        self.addCleanup(self._print_patch.stop)
        self.addCleanup(lambda: shutil.rmtree(self._storage_dir, ignore_errors=True))

    def _configure_unified_history_limits(
        self,
        *,
        tool_limit: int,
        unified_limit: int,
        hysteresis: int,
        reasoning_limit: int = 3,
    ):
        config, _ = PromptConfig.objects.get_or_create(singleton_id=1)
        for tier_prefix in ("standard", "premium", "max", "ultra", "ultra_max"):
            setattr(config, f"{tier_prefix}_tool_call_history_limit", tool_limit)
            setattr(config, f"{tier_prefix}_unified_history_limit", unified_limit)
            setattr(config, f"{tier_prefix}_unified_history_hysteresis", hysteresis)
        config.internal_reasoning_history_limit = reasoning_limit
        config.save()
        invalidate_prompt_settings_cache()
        return config

    def _build_user_prompt_content(self) -> str:
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        return user_message["content"]

    def _create_history_step(
        self,
        *,
        created_at,
        description: str,
        tool_name: str | None = None,
        result: dict | None = None,
    ) -> PersistentAgentStep:
        step = PersistentAgentStep.objects.create(
            agent=self.agent,
            description=description,
        )
        if tool_name is not None:
            PersistentAgentToolCall.objects.create(
                step=step,
                tool_name=tool_name,
                tool_params={"query": tool_name},
                result=json.dumps(result or {"status": tool_name}),
            )
        PersistentAgentStep.objects.filter(pk=step.pk).update(created_at=created_at)
        step.refresh_from_db()
        return step

    def _create_completed_browser_task(
        self,
        *,
        prompt: str,
        updated_at,
        result_value=None,
        status=BrowserUseAgentTask.StatusChoices.COMPLETED,
        error_message: str | None = None,
    ) -> BrowserUseAgentTask:
        task = BrowserUseAgentTask.objects.create(
            agent=self.browser_agent,
            user=self.user,
            prompt=prompt,
            status=status,
            error_message=error_message,
        )
        if result_value is None and status == BrowserUseAgentTask.StatusChoices.COMPLETED:
            result_value = {"task": prompt}
        if result_value is not None:
            BrowserUseAgentTaskStep.objects.create(
                task=task,
                step_number=1,
                description=f"Result for {prompt}",
                is_result=True,
                result_value=result_value,
            )
        BrowserUseAgentTask.objects.filter(pk=task.pk).update(updated_at=updated_at)
        task.refresh_from_db()
        return task

    def _create_inbound_email_message(self, *, body: str, timestamp) -> PersistentAgentMessage:
        message = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body=body,
            raw_payload={},
            seq=f"UH{int(timestamp.timestamp() * 1_000_000):024d}"[:26],
        )
        PersistentAgentMessage.objects.filter(pk=message.pk).update(timestamp=timestamp)
        message.refresh_from_db()
        return message

    def test_prompt_uses_configured_skill_prompt_limit(self):
        config, _ = PromptConfig.objects.get_or_create(singleton_id=1)
        config.standard_skill_prompt_limit = 2
        config.save()
        invalidate_prompt_settings_cache()

        now = timezone.now()
        for idx in range(4):
            skill = PersistentAgentSkill.objects.create(
                agent=self.agent,
                name=f"prompt-skill-{idx}",
                description=f"description-{idx}",
                version=1,
                tools=["sqlite_batch"],
                instructions=f"instructions for skill {idx}",
            )
            PersistentAgentSkill.objects.filter(id=skill.id).update(
                updated_at=now + timedelta(minutes=idx),
                last_used_at=now + timedelta(hours=idx),
            )

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), \
             patch("api.agent.core.prompt_context.ensure_comms_compacted"), \
             patch("api.agent.core.prompt_context.get_agent_llm_tier", return_value=AgentLLMTier.STANDARD):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn("<agent_skills>", content)
        self.assertIn("Skill: prompt-skill-3 (v1)", content)
        self.assertIn("Skill: prompt-skill-2 (v1)", content)
        self.assertNotIn("Skill: prompt-skill-1 (v1)", content)
        self.assertNotIn("Skill: prompt-skill-0 (v1)", content)
        self.assertNotIn("System Skill: Runtime Planning\nKey:", content)
        self.assertIn("Omitted skills due to prompt limit:", content)
        self.assertIn("- prompt-skill-1", content)
        self.assertIn("- prompt-skill-0", content)
        self.assertIn("- System Skill: Runtime Planning (runtime_planning)", content)

    def test_prompt_includes_default_runtime_planning_system_skill_when_limit_allows(self):
        config, _ = PromptConfig.objects.get_or_create(singleton_id=1)
        config.standard_skill_prompt_limit = 1
        config.save()
        invalidate_prompt_settings_cache()

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), \
             patch("api.agent.core.prompt_context.ensure_comms_compacted"), \
             patch("api.agent.core.prompt_context.get_agent_llm_tier", return_value=AgentLLMTier.STANDARD):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn("<agent_skills>", content)
        self.assertIn("System Skill: Runtime Planning", content)
        self.assertIn("Use `update_plan` to track substantial multi-step work", content)

    def test_update_plan_tool_execution_refreshes_runtime_planning_system_skill(self):
        prepared = _PreparedToolExecution(
            idx=0,
            tool_name="update_plan",
            tool_params={"plan": [{"step": "Check source", "status": "doing"}]},
            exec_params={"plan": [{"step": "Check source", "status": "doing"}]},
            pending_step=None,
            credits_consumed=None,
            consumed_credit=None,
            call_id="call-plan",
            explicit_continue=None,
            inferred_continue=True,
            parallel_safe=False,
            parallel_ineligible_reason="unsafe_tool:update_plan",
        )

        outcome = _execute_prepared_tool_call(
            self.agent,
            prepared,
            budget_ctx=None,
            eval_run_id=None,
        )

        self.assertEqual(outcome.result["status"], "ok")
        state = PersistentAgentSystemSkillState.objects.get(
            agent=self.agent,
            skill_key="runtime_planning",
        )
        self.assertIsNotNone(state.last_used_at)
        self.assertEqual(state.usage_count, 1)

    def test_prompt_omits_skill_section_when_skill_prompt_limit_is_zero(self):
        config, _ = PromptConfig.objects.get_or_create(singleton_id=1)
        config.standard_skill_prompt_limit = 0
        config.save()
        invalidate_prompt_settings_cache()

        PersistentAgentSkill.objects.create(
            agent=self.agent,
            name="hidden-skill",
            description="Should not appear in prompt context.",
            version=1,
            tools=["sqlite_batch"],
            instructions="This should stay out of the prompt.",
        )

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), \
             patch("api.agent.core.prompt_context.ensure_comms_compacted"), \
             patch("api.agent.core.prompt_context.get_agent_llm_tier", return_value=AgentLLMTier.STANDARD):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertNotIn("<agent_skills>", content)
        self.assertNotIn("Skill: hidden-skill (v1)", content)
        self.assertNotIn("System Skill: Runtime Planning", content)
        self.assertNotIn("Use `update_plan` to track steps", content)

    def _build_org_agent_web_interaction(self, *, org_slug: str, member_email: str, is_org_member: bool):
        org = Organization.objects.create(
            name=f"{org_slug} name",
            slug=org_slug,
            plan="free",
            created_by=self.user,
        )
        billing = org.billing
        billing.purchased_seats = 2 if is_org_member else 1
        billing.save(update_fields=["purchased_seats"])
        OrganizationMembership.objects.create(
            org=org,
            user=self.user,
            role=OrganizationMembership.OrgRole.OWNER,
            status=OrganizationMembership.OrgStatus.ACTIVE,
        )
        andrew = User.objects.create_user(
            username=f"{org_slug}-andrew",
            email=member_email,
            password="secret",
            first_name="Andrew",
            last_name="Christianson",
        )
        if is_org_member:
            OrganizationMembership.objects.create(
                org=org,
                user=andrew,
                role=OrganizationMembership.OrgRole.MEMBER,
                status=OrganizationMembership.OrgStatus.ACTIVE,
            )

        org_browser_agent = BrowserUseAgent.objects.create(
            user=self.user,
            name=f"{org_slug} browser agent",
        )
        org_agent = PersistentAgent.objects.create(
            user=self.user,
            organization=org,
            name=f"{org_slug} prompt agent",
            charter="Org prompt test",
            browser_use_agent=org_browser_agent,
        )
        agent_web_endpoint = PersistentAgentCommsEndpoint.objects.create(
            owner_agent=org_agent,
            channel="web",
            address=build_web_agent_address(org_agent.id),
        )
        andrew_address = build_web_user_address(andrew.id, org_agent.id)
        andrew_web_endpoint = PersistentAgentCommsEndpoint.objects.create(
            channel="web",
            address=andrew_address,
        )
        conversation = PersistentAgentConversation.objects.create(
            owner_agent=org_agent,
            channel="web",
            address=andrew_address,
        )
        PersistentAgentConversationParticipant.objects.create(
            conversation=conversation,
            endpoint=agent_web_endpoint,
            role=PersistentAgentConversationParticipant.ParticipantRole.AGENT,
        )
        PersistentAgentConversationParticipant.objects.create(
            conversation=conversation,
            endpoint=andrew_web_endpoint,
            role=PersistentAgentConversationParticipant.ParticipantRole.HUMAN_USER,
        )
        PersistentAgentMessage.objects.create(
            owner_agent=org_agent,
            from_endpoint=andrew_web_endpoint,
            to_endpoint=agent_web_endpoint,
            conversation=conversation,
            is_outbound=False,
            body="Hello from Andrew on web",
            seq=f"{org_slug.upper().replace('-', '')[:9]}{int(timezone.now().timestamp() * 1_000_000):017d}"[:26],
        )
        return org_agent, andrew_address

    def test_message_metadata_in_prompt(self):
        """Test that message metadata (from, channel) is included in the prompt."""
        # Create a mock event window with one message
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="Hello agent!",
            seq=f"TEST{int(timezone.now().timestamp() * 1_000_000):022d}"[:26],
        )
        # Build the prompt context
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        # Find the user message in the context
        user_message = next((m for m in context if m['role'] == 'user'), None)

        self.assertIsNotNone(user_message)
        
        # Check that the content includes the structured format with message metadata
        content = user_message['content']
        
        # Verify the event block exists and contains message metadata
        self.assertIn('<event_', content)  # Event sections start with <event_
        self.assertIn('_message_inbound_email>', content)  # Contains message event type
        self.assertIn(f'On {self.external_endpoint.channel}, you received a message from {self.external_endpoint.address}:', content)
        self.assertIn('<body>', content)  # Updated to match current format
        self.assertIn('Hello agent!', content)
        self.assertIn('</body>', content)  # Updated to match current format

        # Verify other expected blocks are present
        self.assertIn('<charter>Test prompt context</charter>', content)
        self.assertIn('<schedule>No schedule configured</schedule>', content)
        self.assertIn('<current_datetime>', content)
        self.assertIn('</current_datetime>', content)
        self.assertIn('<pacing_guidance>', content)
        self.assertIn('<time_since_last_interaction>', content)
        self.assertIn('<burn_rate_status>', content)

    def test_unified_history_message_headers_omit_recent_age_suffix(self):
        message = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="Age suffix check",
            seq=f"AGESFX{int(timezone.now().timestamp() * 1_000_000):019d}"[:26],
        )
        target_time = timezone.now() - timedelta(minutes=5)
        PersistentAgentMessage.objects.filter(pk=message.pk).update(timestamp=target_time)
        message.refresh_from_db()

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn(
            f"[{message.timestamp.isoformat()}] On email, you received a message from {self.external_endpoint.address}:",
            content,
        )
        self.assertNotRegex(content, re.compile(r"\]\s+\d+[smh] ago,"))

    def test_prompt_discourages_messages_freshness_polling(self):
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="Fresh inbound message for this run",
            seq=f"FRESH{int(timezone.now().timestamp() * 1_000_000):021d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        system_message = next((m for m in context if m['role'] == 'system'), None)
        user_message = next((m for m in context if m['role'] == 'user'), None)

        self.assertIsNotNone(system_message)
        self.assertIsNotNone(user_message)
        system_content = system_message["content"]
        user_content = user_message["content"]
        combined = f"{system_content}\n{user_content}"

        self.assertRegex(
            combined,
            re.compile(r"do not (?:query|poll)\s+__messages.*anything new", re.IGNORECASE),
        )
        self.assertRegex(
            combined,
            re.compile(
                r"use\s+__messages\s+only\s+for\s+structured\s+analysis,\s+filtering/aggregation,\s+or\s+historical\s+lookup",
                re.IGNORECASE,
            ),
        )
        self.assertRegex(
            combined,
            re.compile(r"new inbound messages.*already.*(unified history|shown below)", re.IGNORECASE),
        )
        self.assertNotIn(
            "inbound_unreadish → SELECT * FROM __messages WHERE is_outbound=0 ORDER BY timestamp DESC",
            system_content,
        )

    def test_planning_preview_prompt_excludes_signup_preview_handoff(self):
        self.agent.charter = GENERIC_STARTER_CHARTER
        self.agent.preferred_contact_endpoint = self.external_endpoint
        self.agent.signup_preview_state = PersistentAgent.SignupPreviewState.AWAITING_FIRST_REPLY_PAUSE
        self.agent.planning_state = PersistentAgent.PlanningState.PLANNING
        self.agent.save(
            update_fields=[
                "charter",
                "preferred_contact_endpoint",
                "signup_preview_state",
                "planning_state",
                "updated_at",
            ]
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent, is_first_run=True)

        system_message = next((m for m in context if m["role"] == "system"), None)

        self.assertIsNotNone(system_message)
        self.assertIn("## Planning Mode", system_message["content"])
        self.assertIn("If there is no concrete task to do yet, your first action should be one concise welcome message", system_message["content"])
        self.assertIn("After the welcome, continue Planning Mode", system_message["content"])
        self.assertIn("Stay in planning only until planning is completed or skipped", system_message["content"])
        self.assertIn("Read-only research is allowed and often useful during planning", system_message["content"])
        self.assertIn("substantive execution or deliverable work before planning ends", system_message["content"])
        self.assertIn(
            "Do not ask planning questions about communication channels, delivery methods, integrations, accounts, or implementation approach unless the user explicitly asks to configure or choose them",
            system_message["content"],
        )
        self.assertNotIn("delivery cadence", system_message["content"])
        self.assertNotIn("## Signup Preview Handoff", system_message["content"])
        self.assertNotIn("## Signup Preview First-Run Override", system_message["content"])
        self.assertNotIn("limited preview", system_message["content"])
        self.assertIn(f"<charter>{GENERIC_STARTER_CHARTER}</charter>", next(
            m for m in context if m["role"] == "user"
        )["content"])

    def test_signup_preview_handoff_prompt_runs_after_planning_without_first_run(self):
        self.agent.charter = GENERIC_STARTER_CHARTER
        self.agent.preferred_contact_endpoint = self.external_endpoint
        for planning_state in (
            PersistentAgent.PlanningState.COMPLETED,
            PersistentAgent.PlanningState.SKIPPED,
        ):
            with self.subTest(planning_state=planning_state):
                self.agent.signup_preview_state = PersistentAgent.SignupPreviewState.AWAITING_FIRST_REPLY_PAUSE
                self.agent.planning_state = PersistentAgent.PlanningState.PLANNING
                self.agent.save(
                    update_fields=[
                        "charter",
                        "preferred_contact_endpoint",
                        "signup_preview_state",
                        "planning_state",
                        "updated_at",
                    ]
                )
                PersistentAgentMessage.objects.create(
                    owner_agent=self.agent,
                    from_endpoint=self.endpoint,
                    to_endpoint=self.external_endpoint,
                    is_outbound=True,
                    body=f"Planning welcome before {planning_state}",
                )
                self.agent.planning_state = planning_state
                self.agent.save(update_fields=["planning_state", "updated_at"])

                with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
                     patch('api.agent.core.prompt_context.ensure_comms_compacted'):
                    context, _, _ = build_prompt_context(self.agent, is_first_run=False)

                system_message = next((m for m in context if m["role"] == "system"), None)

                self.assertIsNotNone(system_message)
                system_content = system_message["content"]
                self.assertIn("## Signup Preview Handoff", system_content)
                self.assertIn("Contact channel: email", system_content)
                self.assertIn("the plan is ready", system_content)
                self.assertIn("after they finish signup", system_content)
                self.assertNotIn("## Planning Mode", system_content)
                self.assertNotIn("## REQUIRED: First-Run Welcome", system_content)
                self.assertIn(f"<charter>{GENERIC_STARTER_CHARTER}</charter>", next(
                    m for m in context if m["role"] == "user"
                )["content"])

    def test_unified_history_uses_collaborator_name_for_web_sender(self):
        self.user.first_name = "Will"
        self.user.save(update_fields=["first_name"])
        collaborator = User.objects.create_user(
            username="andrew-collab",
            email="andrew@example.com",
            password="secret",
            first_name="Andrew",
        )
        agent_web_endpoint = PersistentAgentCommsEndpoint.objects.create(
            owner_agent=self.agent,
            channel="web",
            address=build_web_agent_address(self.agent.id),
        )
        collaborator_address = build_web_user_address(collaborator.id, self.agent.id)
        collaborator_web_endpoint = PersistentAgentCommsEndpoint.objects.create(
            channel="web",
            address=collaborator_address,
        )
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=collaborator_web_endpoint,
            to_endpoint=agent_web_endpoint,
            is_outbound=False,
            body="Hello from Andrew",
            seq=f"WEBNAME{int(timezone.now().timestamp() * 1_000_000):019d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn(
            f"On web, you received a message from {collaborator_address} - Andrew:",
            content,
        )
        self.assertNotIn("On web, you received a message from Will:", content)

    def test_unified_history_web_sender_has_no_email_fallback(self):
        collaborator = User.objects.create_user(
            username="andrew_handle",
            email="andrew@example.com",
            password="secret",
            first_name="",
            last_name="",
        )
        agent_web_endpoint = PersistentAgentCommsEndpoint.objects.create(
            owner_agent=self.agent,
            channel="web",
            address=build_web_agent_address(self.agent.id),
        )
        collaborator_address = build_web_user_address(collaborator.id, self.agent.id)
        collaborator_web_endpoint = PersistentAgentCommsEndpoint.objects.create(
            channel="web",
            address=collaborator_address,
        )
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=collaborator_web_endpoint,
            to_endpoint=agent_web_endpoint,
            is_outbound=False,
            body="Web sender identity fallback check",
            seq=f"WEBNOEMAIL{int(timezone.now().timestamp() * 1_000_000):016d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn(
            f"On web, you received a message from {collaborator_address} - andrew_handle:",
            content,
        )
        self.assertNotIn("On web, you received a message from andrew@example.com:", content)

    def test_user_identity_mentions_shared_chat_sender_guidance(self):
        self.user.first_name = "Will"
        self.user.save(update_fields=["first_name"])

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn("<user_identity>", content)
        self.assertIn(
            "In shared chats, address the most recent inbound sender from unified history/recent contacts;",
            content,
        )

    def test_user_endpoints_include_org_member_email_for_interacted_web_user(self):
        org_agent, andrew_address = self._build_org_agent_web_interaction(
            org_slug="prompt-org",
            member_email="andrew@example.com",
            is_org_member=True,
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(org_agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn("<user_endpoints>", content)
        self.assertIn(f"- web: {andrew_address} - Andrew Christianson", content)
        self.assertIn("- email: andrew@example.com - Andrew Christianson", content)

    def test_user_endpoints_do_not_add_email_for_non_org_member_web_user(self):
        org_agent, andrew_address = self._build_org_agent_web_interaction(
            org_slug="prompt-org-missing-member",
            member_email="andrew-nonmember@example.com",
            is_org_member=False,
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(org_agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn("<user_endpoints>", content)
        self.assertIn(f"- web: {andrew_address} - Andrew Christianson", content)
        self.assertNotIn("- email: andrew-nonmember@example.com - Andrew Christianson", content)

    def test_build_prompt_context_populates_messages_sqlite_table(self):
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="Hello from sqlite snapshot",
            raw_payload={"subject": "Snapshot Subject", "hide_in_chat": True},
            seq=f"SQLMSG{int(timezone.now().timestamp() * 1_000_000):020d}"[:26],
        )

        sqlite_tmp = tempfile.TemporaryDirectory()
        db_path = f"{sqlite_tmp.name}/state.db"
        token = set_sqlite_db_path(db_path)
        try:
            with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
                 patch('api.agent.core.prompt_context.ensure_comms_compacted'):
                build_prompt_context(self.agent)

            conn = sqlite3.connect(db_path)
            try:
                cur = conn.cursor()
                cur.execute(
                    """
                    SELECT channel, is_outbound, subject, body, is_hidden_in_chat, attachment_paths_json, rejected_attachments_json
                    FROM "__messages"
                    ORDER BY timestamp DESC
                    LIMIT 1;
                    """
                )
                row = cur.fetchone()
                self.assertIsNotNone(row)
                assert row is not None
                self.assertEqual(row[0], "email")
                self.assertEqual(row[1], 0)
                self.assertEqual(row[2], "Snapshot Subject")
                self.assertIn("Hello from sqlite snapshot", row[3])
                self.assertEqual(row[4], 1)
                self.assertEqual(json.loads(row[5]), [])
                self.assertEqual(json.loads(row[6]), [])
            finally:
                conn.close()
        finally:
            reset_sqlite_db_path(token)
            sqlite_tmp.cleanup()

    def test_sqlite_messages_snapshot_includes_rejected_attachment_metadata(self):
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="Please review the email body.",
            raw_payload={
                "subject": "Oversize file attempt",
                "rejected_attachments": [
                    {
                        "filename": "deck.pdf",
                        "size_bytes": 71303168,
                        "limit_bytes": 10485760,
                        "reason_code": "too_large",
                        "channel": "email",
                    }
                ],
            },
            seq=f"SQLRJ{int(timezone.now().timestamp() * 1_000_000):021d}"[:26],
        )

        sqlite_tmp = tempfile.TemporaryDirectory()
        db_path = f"{sqlite_tmp.name}/state.db"
        token = set_sqlite_db_path(db_path)
        try:
            with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
                 patch('api.agent.core.prompt_context.ensure_comms_compacted'):
                build_prompt_context(self.agent)

            conn = sqlite3.connect(db_path)
            try:
                cur = conn.cursor()
                cur.execute(
                    """
                    SELECT rejected_attachments_json
                    FROM "__messages"
                    ORDER BY timestamp DESC
                    LIMIT 1;
                    """
                )
                row = cur.fetchone()
                self.assertIsNotNone(row)
                assert row is not None
                self.assertEqual(
                    json.loads(row[0]),
                    [
                        {
                            "filename": "deck.pdf",
                            "size_bytes": 71303168,
                            "limit_bytes": 10485760,
                            "reason_code": "too_large",
                            "channel": "email",
                        }
                    ],
                )
            finally:
                conn.close()
        finally:
            reset_sqlite_db_path(token)
            sqlite_tmp.cleanup()

    def test_prompt_marks_outbound_attachment_count_and_paths(self):
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.endpoint,
            to_endpoint=self.external_endpoint,
            is_outbound=True,
            body="Please review the exported CSV.",
            raw_payload={
                "subject": "Batch delivery",
                "filespace_nodes": [{"path": "/exports/report.csv"}],
            },
            seq=f"ATTACH{int(timezone.now().timestamp() * 1_000_000):020d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn(
            f"On email, you sent a message to {self.external_endpoint.address} [attachments: 1]:",
            content,
        )
        self.assertIn("<attachments>", content)
        self.assertIn("$[/exports/report.csv]", content)

    def test_prompt_marks_outbound_zero_attachment_count(self):
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.endpoint,
            to_endpoint=self.external_endpoint,
            is_outbound=True,
            body="Please see the attached CSV below.",
            raw_payload={"subject": "Attachment included"},
            seq=f"CLAIM{int(timezone.now().timestamp() * 1_000_000):021d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn(
            (
                f"On email, you sent a message to {self.external_endpoint.address} "
                "[attachments: 0]:"
            ),
            content,
        )

    def test_prompt_describes_sqlite_snapshots_without_banning_read_file(self):
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn(
            "Use read_file for contents of known filespace paths; use sqlite_batch on __tool_results or __files only for prior tool outputs or file metadata.",
            content,
        )
        self.assertNotIn("Query __tool_results and __files with sqlite_batch (not read_file).", content)

    def test_recent_contacts_include_email_message_id(self):
        conversation = PersistentAgentConversation.objects.create(
            owner_agent=self.agent,
            channel=CommsChannel.EMAIL,
            address=self.external_endpoint.address,
        )
        message = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.endpoint,
            conversation=conversation,
            is_outbound=True,
            body="Recent thread body",
            raw_payload={
                "subject": "Recent thread subject",
                "message_id": "<recent-thread@example.com>",
            },
            seq=f"RCMID{int(timezone.now().timestamp() * 1_000_000):020d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn("<recent_contacts>", content)
        self.assertIn(f"reply_to_message_id: {message.id}", content)
        self.assertNotIn("<recent-thread@example.com>", content)
        self.assertIn("recent subj: Recent thread subject", content)

    def test_recent_contacts_include_legacy_outbound_reply_to_message_id(self):
        conversation = PersistentAgentConversation.objects.create(
            owner_agent=self.agent,
            channel=CommsChannel.EMAIL,
            address=self.external_endpoint.address,
        )
        message = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.endpoint,
            conversation=conversation,
            is_outbound=True,
            body="Legacy outbound body",
            raw_payload={"subject": "Legacy outbound subject"},
            seq=f"RCLEGACY{int(timezone.now().timestamp() * 1_000_000):018d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn("recent subj: Legacy outbound subject", content)
        self.assertIn(f"reply_to_message_id: {message.id}", content)

    def test_unified_history_includes_email_reply_to_message_id_component(self):
        conversation = PersistentAgentConversation.objects.create(
            owner_agent=self.agent,
            channel=CommsChannel.EMAIL,
            address=self.external_endpoint.address,
        )
        message = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.endpoint,
            conversation=conversation,
            is_outbound=True,
            body="History body",
            raw_payload={
                "subject": "History subject",
                "message_id": "<history-thread@example.com>",
            },
            seq=f"HISTMID{int(timezone.now().timestamp() * 1_000_000):019d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn(
            f"On email, you sent a message to {self.external_endpoint.address} [attachments: 0]:",
            content,
        )
        self.assertIn(f"<reply_to_message_id>{message.id}</reply_to_message_id>", content)
        self.assertNotIn("<history-thread@example.com>", content)

    def test_unified_history_inbound_email_includes_reply_to_message_id_component(self):
        conversation = PersistentAgentConversation.objects.create(
            owner_agent=self.agent,
            channel=CommsChannel.EMAIL,
            address=self.external_endpoint.address,
        )
        message = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            conversation=conversation,
            is_outbound=False,
            body="Inbound history body",
            raw_payload={
                "subject": "Inbound history subject",
                "message_id": "<inbound-history@example.com>",
            },
            seq=f"HINBOUND{int(timezone.now().timestamp() * 1_000_000):018d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn('_message_inbound_email>', content)
        self.assertIn(
            f"On email, you received a message from {self.external_endpoint.address}:",
            content,
        )
        self.assertIn(f"<reply_to_message_id>{message.id}</reply_to_message_id>", content)
        self.assertNotIn("<inbound-history@example.com>", content)

    def test_unified_history_includes_legacy_outbound_reply_to_message_id(self):
        conversation = PersistentAgentConversation.objects.create(
            owner_agent=self.agent,
            channel=CommsChannel.EMAIL,
            address=self.external_endpoint.address,
        )
        message = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.endpoint,
            conversation=conversation,
            is_outbound=True,
            body="Legacy history body",
            raw_payload={"subject": "Legacy history subject"},
            seq=f"HLEGACY{int(timezone.now().timestamp() * 1_000_000):019d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn("Legacy history subject", content)
        self.assertIn(f"<reply_to_message_id>{message.id}</reply_to_message_id>", content)

    def test_system_prompt_includes_attachment_preflight_guidance(self):
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        system_message = next((m for m in context if m["role"] == "system"), None)
        self.assertIsNotNone(system_message)
        content = system_message["content"]

        self.assertIn("# Attachment pre-flight", content)
        self.assertIn("RIGHT: send_email(..., attachments=[result.attach])", content)
        self.assertIn("For resend/reply/duplicate risk: verify prior sends via __messages.attachment_count", content)
        self.assertIn("__messages.rejected_attachments_json", content)

    def test_prompt_does_not_add_attachment_suffix_to_inbound_messages(self):
        PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="Can you resend the file?",
            raw_payload={"subject": "Inbound follow-up"},
            seq=f"INBOUND{int(timezone.now().timestamp() * 1_000_000):019d}"[:26],
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]

        self.assertIn(
            f"On email, you received a message from {self.external_endpoint.address}:",
            content,
        )
        self.assertNotIn(
            f"On email, you received a message from {self.external_endpoint.address} [attachments:",
            content,
        )

    def test_messages_sqlite_snapshot_includes_full_bodies_up_to_budget(self):
        base = timezone.now()
        oldest = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="AAAAAA",
            seq=f"SQLA{int(timezone.now().timestamp() * 1_000_000):022d}"[:26],
        )
        middle = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="BBBBBB",
            seq=f"SQLB{int(timezone.now().timestamp() * 1_000_000):022d}"[:26],
        )
        newest = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="CCCCCC",
            seq=f"SQLC{int(timezone.now().timestamp() * 1_000_000):022d}"[:26],
        )

        PersistentAgentMessage.objects.filter(pk=oldest.pk).update(timestamp=base - timedelta(minutes=3))
        PersistentAgentMessage.objects.filter(pk=middle.pk).update(timestamp=base - timedelta(minutes=2))
        PersistentAgentMessage.objects.filter(pk=newest.pk).update(timestamp=base - timedelta(minutes=1))

        sqlite_tmp = tempfile.TemporaryDirectory()
        db_path = f"{sqlite_tmp.name}/state.db"
        token = set_sqlite_db_path(db_path)
        try:
            with patch("api.agent.core.prompt_context.SQLITE_MESSAGES_SNAPSHOT_MAX_BYTES", 12), \
                 patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
                 patch('api.agent.core.prompt_context.ensure_comms_compacted'):
                build_prompt_context(self.agent)

            conn = sqlite3.connect(db_path)
            try:
                cur = conn.cursor()
                cur.execute(
                    """
                    SELECT body, body_is_truncated, body_truncated_bytes
                    FROM "__messages"
                    ORDER BY timestamp DESC;
                    """
                )
                rows = cur.fetchall()
                self.assertEqual(len(rows), 2)
                self.assertEqual(rows[0][0], "CCCCCC")
                self.assertEqual(rows[1][0], "BBBBBB")
                self.assertEqual(rows[0][1], 0)
                self.assertEqual(rows[1][1], 0)
                self.assertEqual(rows[0][2], 0)
                self.assertEqual(rows[1][2], 0)
            finally:
                conn.close()
        finally:
            reset_sqlite_db_path(token)
            sqlite_tmp.cleanup()

    def test_messages_sqlite_snapshot_respects_record_cap(self):
        base = timezone.now()
        oldest = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="A",
            seq=f"SQLR1{int(timezone.now().timestamp() * 1_000_000):021d}"[:26],
        )
        middle = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="B",
            seq=f"SQLR2{int(timezone.now().timestamp() * 1_000_000):021d}"[:26],
        )
        newest = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body="C",
            seq=f"SQLR3{int(timezone.now().timestamp() * 1_000_000):021d}"[:26],
        )

        PersistentAgentMessage.objects.filter(pk=oldest.pk).update(timestamp=base - timedelta(minutes=3))
        PersistentAgentMessage.objects.filter(pk=middle.pk).update(timestamp=base - timedelta(minutes=2))
        PersistentAgentMessage.objects.filter(pk=newest.pk).update(timestamp=base - timedelta(minutes=1))

        sqlite_tmp = tempfile.TemporaryDirectory()
        db_path = f"{sqlite_tmp.name}/state.db"
        token = set_sqlite_db_path(db_path)
        try:
            with patch("api.agent.core.prompt_context.SQLITE_MESSAGES_SNAPSHOT_MAX_RECORDS", 2), \
                 patch("api.agent.core.prompt_context.SQLITE_MESSAGES_SNAPSHOT_MAX_BYTES", 1000), \
                 patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
                 patch('api.agent.core.prompt_context.ensure_comms_compacted'):
                build_prompt_context(self.agent)

            conn = sqlite3.connect(db_path)
            try:
                cur = conn.cursor()
                cur.execute(
                    """
                    SELECT body
                    FROM "__messages"
                    ORDER BY timestamp DESC;
                    """
                )
                rows = cur.fetchall()
                self.assertEqual(len(rows), 2)
                self.assertEqual(rows[0][0], "C")
                self.assertEqual(rows[1][0], "B")
            finally:
                conn.close()
        finally:
            reset_sqlite_db_path(token)
            sqlite_tmp.cleanup()

    def test_prompt_omits_implied_send_without_active_web_session(self):
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        system_message = next((m for m in context if m['role'] == 'system'), None)
        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(system_message)
        self.assertIsNotNone(user_message)
        combined = f"{system_message['content']}\n{user_message['content']}"
        self.assertNotIn("Implied Send", combined)
        self.assertNotIn("<implied_send_status>", combined)
        self.assertNotIn("implied_send_status", combined)

    def test_session_tool_gating_excludes_send_chat_message_without_active_web_session_when_verified_fallback_exists(self):
        EmailAddress.objects.create(
            user=self.user,
            email=self.user.email,
            verified=True,
            primary=True,
        )
        tools = get_agent_tools(self.agent)
        gated_tools = _gate_send_chat_tool_for_delivery(tools, self.agent)
        tool_names = [
            entry.get("function", {}).get("name")
            for entry in gated_tools
            if isinstance(entry, dict)
        ]
        self.assertNotIn("send_chat_message", tool_names)

    def test_session_tool_gating_includes_send_chat_message_without_active_web_session_when_no_verified_fallback(self):
        tools = get_agent_tools(self.agent)
        gated_tools = _gate_send_chat_tool_for_delivery(tools, self.agent)
        tool_names = [
            entry.get("function", {}).get("name")
            for entry in gated_tools
            if isinstance(entry, dict)
        ]
        self.assertIn("send_chat_message", tool_names)

    def test_session_tool_gating_includes_send_chat_message_with_active_web_session(self):
        start_web_session(self.agent, self.user)
        tools = get_agent_tools(self.agent)
        gated_tools = _gate_send_chat_tool_for_delivery(tools, self.agent)
        tool_names = [
            entry.get("function", {}).get("name")
            for entry in gated_tools
            if isinstance(entry, dict)
        ]
        self.assertIn("send_chat_message", tool_names)

    def test_prompt_adds_retry_hint_when_send_chat_message_unavailable(self):
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        system_message = next((m for m in context if m['role'] == 'system'), None)
        self.assertIsNotNone(system_message)
        content = system_message["content"]
        self.assertIn(
            "and send_email/send_sms/send_agent_message for other channels",
            content,
        )
        self.assertIn(
            "If send_chat_message is unavailable, use the most recently active non-web channel from unified history/recent contacts",
            content,
        )

    def test_prompt_includes_implied_send_with_active_web_session(self):
        start_web_session(self.agent, self.user)
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        system_message = next((m for m in context if m['role'] == 'system'), None)
        self.assertIsNotNone(system_message)
        content = system_message['content']
        self.assertIn("Implied Send", content)

    def test_prompt_omits_implied_send_when_primary_model_disables_it(self):
        start_web_session(self.agent, self.user)
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch(
                 'api.agent.core.prompt_context.get_llm_config_with_failover',
                 return_value=[('endpoint', 'openai/gpt-4o-mini', {'allow_implied_send': False})],
             ):
            context, _, _ = build_prompt_context(self.agent)

        system_message = next((m for m in context if m['role'] == 'system'), None)
        self.assertIsNotNone(system_message)
        content = system_message['content']
        self.assertNotIn("Implied Send", content)
        self.assertNotIn("send_<channel>", content)
        self.assertIn(
            "Text-only replies are not delivered when implied send is unavailable",
            content,
        )
        self.assertNotIn(
            "Text-only replies are not delivered without an active web chat session",
            content,
        )

    def test_prompt_omits_implied_send_when_any_failover_model_disables_it(self):
        start_web_session(self.agent, self.user)
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch(
                 'api.agent.core.prompt_context.get_llm_config_with_failover',
                 return_value=[
                     ('endpoint-primary', 'openai/gpt-4o-mini', {'allow_implied_send': True}),
                     ('endpoint-fallback', 'openai/gpt-4.1-mini', {'allow_implied_send': False}),
                 ],
             ):
            context, _, _, metadata = build_prompt_context(self.agent, include_metadata=True)

        system_message = next((m for m in context if m['role'] == 'system'), None)
        self.assertIsNotNone(system_message)
        content = system_message['content']
        self.assertFalse(metadata["prompt_allows_implied_send"])
        self.assertEqual(len(metadata["prompt_failover_configs"]), 2)
        self.assertNotIn("Implied Send", content)
        self.assertIn(
            "Text-only replies are not delivered when implied send is unavailable",
            content,
        )

    def test_tool_call_history_includes_cost_component(self):
        """Tool-call unified history should include a dedicated <cost> component."""
        with patch(
            "api.models.TaskCreditService.check_and_consume_credit_for_owner",
            return_value={"success": True, "credit": None},
        ):
            step = PersistentAgentStep.objects.create(
                agent=self.agent,
                description="Tool call: search_tools",
                credits_cost=Decimal("1.234"),
            )
        PersistentAgentToolCall.objects.create(
            step=step,
            tool_name="search_tools",
            tool_params={"query": "hello"},
            result=json.dumps({"status": "ok"}),
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message['content']
        self.assertIn("_tool_call>", content)
        self.assertIn("<cost>1.234 credits</cost>", content)

    def test_completed_browser_tasks_use_unified_history_window_not_tool_history_limit(self):
        self._configure_unified_history_limits(
            tool_limit=2,
            unified_limit=10,
            hysteresis=2,
        )

        base_time = timezone.now()

        self._create_history_step(
            created_at=base_time - timedelta(minutes=4),
            description="Tool call: older_lookup",
            tool_name="older_lookup",
        )
        self._create_completed_browser_task(
            prompt="Older completed task",
            updated_at=base_time - timedelta(minutes=3)
        )
        self._create_history_step(
            created_at=base_time - timedelta(minutes=2),
            description="Tool call: newer_lookup",
            tool_name="newer_lookup",
        )
        self._create_completed_browser_task(
            prompt="Newest completed task",
            updated_at=base_time - timedelta(minutes=1)
        )

        content = self._build_user_prompt_content()

        self.assertIn("Tool older_lookup called.", content)
        self.assertIn("Tool newer_lookup called.", content)
        self.assertIn("Older completed task", content)
        self.assertIn("Newest completed task", content)

    def test_completed_browser_tasks_remain_visible_below_unified_history_threshold(self):
        self._configure_unified_history_limits(
            tool_limit=30,
            unified_limit=70,
            hysteresis=20,
            reasoning_limit=3,
        )

        base_time = timezone.now()
        task_prompts = [
            "Open https://example.com and report the page title only.",
            "Regression browser task 2",
            "Regression browser task 3",
            "Regression browser task 4",
        ]
        for idx, prompt in enumerate(task_prompts):
            self._create_completed_browser_task(
                prompt=prompt,
                updated_at=base_time - timedelta(minutes=10 - idx),
            )

        for idx in range(26):
            self._create_history_step(
                created_at=base_time - timedelta(seconds=120 - idx),
                description=f"visible step {idx}",
            )

        content_before = self._build_user_prompt_content()
        self.assertIn(task_prompts[0], content_before)

        for idx in range(2):
            self._create_history_step(
                created_at=base_time + timedelta(seconds=idx + 1),
                description=f"newest visible step {idx}",
            )

        content_after = self._build_user_prompt_content()
        self.assertIn(task_prompts[0], content_after)
        self.assertIn(task_prompts[3], content_after)

    def test_unified_history_hysteresis_drops_oldest_events_from_merged_stream(self):
        self._configure_unified_history_limits(
            tool_limit=30,
            unified_limit=5,
            hysteresis=2,
            reasoning_limit=0,
        )

        base_time = timezone.now() - timedelta(minutes=20)
        self._create_inbound_email_message(
            body="merged message 1",
            timestamp=base_time + timedelta(minutes=1),
        )
        self._create_history_step(
            created_at=base_time + timedelta(minutes=2),
            description="merged step 1",
        )
        self._create_completed_browser_task(
            prompt="merged task 1",
            updated_at=base_time + timedelta(minutes=3),
        )
        self._create_inbound_email_message(
            body="merged message 2",
            timestamp=base_time + timedelta(minutes=4),
        )
        self._create_history_step(
            created_at=base_time + timedelta(minutes=5),
            description="merged step 2",
        )
        self._create_completed_browser_task(
            prompt="merged task 2",
            updated_at=base_time + timedelta(minutes=6),
        )
        self._create_inbound_email_message(
            body="merged message 3",
            timestamp=base_time + timedelta(minutes=7),
        )
        self._create_history_step(
            created_at=base_time + timedelta(minutes=8),
            description="merged step 3",
        )
        self._create_completed_browser_task(
            prompt="merged task 3",
            updated_at=base_time + timedelta(minutes=9),
        )

        content = self._build_user_prompt_content()

        self.assertNotIn("merged message 1", content)
        self.assertNotIn("merged step 1", content)
        self.assertNotIn("merged task 1", content)
        self.assertNotIn("merged message 2", content)
        self.assertIn("merged step 2", content)
        self.assertIn("merged task 2", content)
        self.assertIn("merged message 3", content)
        self.assertIn("merged step 3", content)
        self.assertIn("merged task 3", content)

    def test_completed_browser_tasks_keep_recency_order_in_merged_history(self):
        self._configure_unified_history_limits(
            tool_limit=1,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=0,
        )

        base_time = timezone.now()
        older_prompt = "ordered browser task older"
        newer_prompt = "ordered browser task newer"
        newest_prompt = "ordered browser task newest"
        self._create_completed_browser_task(
            prompt=older_prompt,
            updated_at=base_time - timedelta(minutes=5),
        )
        self._create_history_step(
            created_at=base_time - timedelta(minutes=4),
            description="ordered step between older and newer",
        )
        self._create_completed_browser_task(
            prompt=newer_prompt,
            updated_at=base_time - timedelta(minutes=3),
        )
        self._create_history_step(
            created_at=base_time - timedelta(minutes=2),
            description="ordered step between newer and newest",
        )
        self._create_completed_browser_task(
            prompt=newest_prompt,
            updated_at=base_time - timedelta(minutes=1),
        )

        content = self._build_user_prompt_content()

        self.assertIn(older_prompt, content)
        self.assertIn(newer_prompt, content)
        self.assertIn(newest_prompt, content)
        self.assertLess(content.index(older_prompt), content.index(newer_prompt))
        self.assertLess(content.index(newer_prompt), content.index(newest_prompt))

    def test_completed_browser_task_results_are_stored_in_sqlite_without_inline_duplication(self):
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=0,
        )

        browser_result = (
            "Successfully extracted all 2 registry items.\n\n"
            "<result>\n"
            "[\n"
            "  {\"brand\": \"Philips Avent\", \"product_name\": \"Natural Glass Bottle\", \"category\": \"Feeding\"},\n"
            "  {\"brand\": \"Momcozy\", \"product_name\": \"Milk Storage Bags\", \"category\": \"Feeding\"}\n"
            "]\n"
            "</result>"
        )
        task = self._create_completed_browser_task(
            prompt="Extract all registry items as JSON.",
            updated_at=timezone.now(),
            result_value=browser_result,
        )

        sqlite_tmp = tempfile.TemporaryDirectory()
        db_path = f"{sqlite_tmp.name}/state.db"
        token = set_sqlite_db_path(db_path)

        try:
            content = self._build_user_prompt_content()
            expected_result_id = str(task.id)

            self.assertIn(str(task.id), content)
            self.assertIn(expected_result_id, content)
            self.assertNotIn("Natural Glass Bottle", content)
            self.assertNotIn("Successfully extracted all 2 registry items.", content)

            conn = sqlite3.connect(db_path)
            try:
                row = conn.execute(
                    """
                    SELECT result_json
                    FROM __tool_results
                    WHERE result_id = ? AND tool_name = ?
                    """,
                    (expected_result_id, "spawn_web_task_result"),
                ).fetchone()
            finally:
                conn.close()
            self.assertIsNotNone(row)

            payload = json.loads(row[0])
            self.assertEqual(str(task.id), payload["task_id"])
            self.assertEqual("completed", payload["status"])
            self.assertEqual("Extract all registry items as JSON.", payload["prompt"])
            self.assertEqual("Philips Avent", payload["result"][0]["brand"])
            self.assertEqual("Natural Glass Bottle", payload["result"][0]["product_name"])
            self.assertIn("Successfully extracted all 2 registry items.", payload["raw_text"])
        finally:
            reset_sqlite_db_path(token)
            sqlite_tmp.cleanup()

    def test_failed_browser_tasks_are_snapshotted_into_sqlite(self):
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=0,
        )

        task = self._create_completed_browser_task(
            prompt="Log in and download the CSV.",
            updated_at=timezone.now(),
            status=BrowserUseAgentTask.StatusChoices.FAILED,
            error_message="Navigation timeout while loading dashboard.",
        )

        sqlite_tmp = tempfile.TemporaryDirectory()
        db_path = f"{sqlite_tmp.name}/state.db"
        token = set_sqlite_db_path(db_path)

        try:
            content = self._build_user_prompt_content()
            expected_result_id = str(task.id)

            self.assertIn(expected_result_id, content)
            self.assertNotIn("Navigation timeout while loading dashboard.", content)

            conn = sqlite3.connect(db_path)
            try:
                row = conn.execute(
                    """
                    SELECT result_json
                    FROM __tool_results
                    WHERE result_id = ? AND tool_name = ?
                    """,
                    (expected_result_id, "spawn_web_task_result"),
                ).fetchone()
            finally:
                conn.close()
            self.assertIsNotNone(row)

            payload = json.loads(row[0])
            self.assertEqual(str(task.id), payload["task_id"])
            self.assertEqual("failed", payload["status"])
            self.assertEqual("Log in and download the CSV.", payload["prompt"])
            self.assertEqual(
                "Navigation timeout while loading dashboard.",
                payload["error_message"],
            )
        finally:
            reset_sqlite_db_path(token)
            sqlite_tmp.cleanup()

    def test_failed_browser_tasks_default_error_message_is_snapshotted(self):
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=0,
        )

        task = self._create_completed_browser_task(
            prompt="Submit the form.",
            updated_at=timezone.now(),
            status=BrowserUseAgentTask.StatusChoices.FAILED,
        )

        sqlite_tmp = tempfile.TemporaryDirectory()
        db_path = f"{sqlite_tmp.name}/state.db"
        token = set_sqlite_db_path(db_path)

        try:
            self._build_user_prompt_content()
            expected_result_id = str(task.id)

            conn = sqlite3.connect(db_path)
            try:
                row = conn.execute(
                    """
                    SELECT result_json
                    FROM __tool_results
                    WHERE result_id = ? AND tool_name = ?
                    """,
                    (expected_result_id, "spawn_web_task_result"),
                ).fetchone()
            finally:
                conn.close()
            self.assertIsNotNone(row)

            payload = json.loads(row[0])
            self.assertEqual("Task failed.", payload["error_message"])
        finally:
            reset_sqlite_db_path(token)
            sqlite_tmp.cleanup()

    def test_cancelled_browser_tasks_use_error_message_field(self):
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=0,
        )

        task = self._create_completed_browser_task(
            prompt="Download the report.",
            updated_at=timezone.now(),
            status=BrowserUseAgentTask.StatusChoices.CANCELLED,
        )

        sqlite_tmp = tempfile.TemporaryDirectory()
        db_path = f"{sqlite_tmp.name}/state.db"
        token = set_sqlite_db_path(db_path)

        try:
            self._build_user_prompt_content()
            expected_result_id = str(task.id)

            conn = sqlite3.connect(db_path)
            try:
                row = conn.execute(
                    """
                    SELECT result_json
                    FROM __tool_results
                    WHERE result_id = ? AND tool_name = ?
                    """,
                    (expected_result_id, "spawn_web_task_result"),
                ).fetchone()
            finally:
                conn.close()
            self.assertIsNotNone(row)

            payload = json.loads(row[0])
            self.assertEqual("cancelled", payload["status"])
            self.assertEqual("Task has been cancelled.", payload["error_message"])
            self.assertNotIn("message", payload)
        finally:
            reset_sqlite_db_path(token)
            sqlite_tmp.cleanup()

    def test_prompt_context_caps_older_internal_reasoning_but_keeps_current_streak(self):
        """Older reasoning stays capped once a newer visible step breaks the streak."""
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=1,
        )

        base_time = timezone.now()
        descriptions = [
            f"{INTERNAL_REASONING_PREFIX} reasoning one",
            f"{INTERNAL_REASONING_PREFIX} reasoning two",
            "Follow up with the user",
            f"{INTERNAL_REASONING_PREFIX} reasoning three",
            f"{INTERNAL_REASONING_PREFIX} reasoning four",
        ]

        for idx, description in enumerate(descriptions):
            step = PersistentAgentStep.objects.create(
                agent=self.agent,
                description=description,
            )
            PersistentAgentStep.objects.filter(pk=step.pk).update(
                created_at=base_time + timedelta(seconds=idx)
            )

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), patch(
            "api.agent.core.prompt_context.ensure_comms_compacted"
        ):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertNotIn("reasoning one", content)
        self.assertIn("reasoning two", content)
        self.assertIn("reasoning three", content)
        self.assertIn("reasoning four", content)
        self.assertIn("Follow up with the user", content)

    def test_prompt_context_preserves_current_reasoning_only_streak(self):
        """The newest contiguous reasoning-only streak bypasses the historical reasoning cap."""
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=1,
        )

        base_time = timezone.now()
        descriptions = [
            "Older visible step",
            f"{INTERNAL_REASONING_PREFIX} current reasoning one",
            f"{INTERNAL_REASONING_PREFIX} current reasoning two",
            f"{INTERNAL_REASONING_PREFIX} current reasoning three",
            f"{INTERNAL_REASONING_PREFIX} current reasoning four",
            f"{INTERNAL_REASONING_PREFIX} current reasoning five",
            f"{INTERNAL_REASONING_PREFIX} current reasoning six",
        ]

        for idx, description in enumerate(descriptions):
            step = PersistentAgentStep.objects.create(
                agent=self.agent,
                description=description,
            )
            PersistentAgentStep.objects.filter(pk=step.pk).update(
                created_at=base_time + timedelta(seconds=idx)
            )

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), patch(
            "api.agent.core.prompt_context.ensure_comms_compacted"
        ):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn("Older visible step", content)
        self.assertIn("current reasoning one", content)
        self.assertIn("current reasoning two", content)
        self.assertIn("current reasoning three", content)
        self.assertIn("current reasoning four", content)
        self.assertIn("current reasoning five", content)
        self.assertIn("current reasoning six", content)

    def test_prompt_context_preserves_current_reasoning_only_streak_when_history_limit_zero(self):
        """A zero historical reasoning cap should still keep the newest reasoning-only streak."""
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=0,
        )

        base_time = timezone.now()
        descriptions = [
            f"{INTERNAL_REASONING_PREFIX} older reasoning",
            "Visible step between reasoning batches",
            f"{INTERNAL_REASONING_PREFIX} current reasoning one",
            f"{INTERNAL_REASONING_PREFIX} current reasoning two",
        ]

        for idx, description in enumerate(descriptions):
            step = PersistentAgentStep.objects.create(
                agent=self.agent,
                description=description,
            )
            PersistentAgentStep.objects.filter(pk=step.pk).update(
                created_at=base_time + timedelta(seconds=idx)
            )

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), patch(
            "api.agent.core.prompt_context.ensure_comms_compacted"
        ):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertNotIn("older reasoning", content)
        self.assertIn("Visible step between reasoning batches", content)
        self.assertIn("current reasoning one", content)
        self.assertIn("current reasoning two", content)

    def test_prompt_context_labels_reasoning_only_steps_as_no_action(self):
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=10,
            hysteresis=2,
            reasoning_limit=3,
        )

        PersistentAgentStep.objects.create(
            agent=self.agent,
            description=build_internal_reasoning_description(
                "I should send the answer now.",
                reasoning_only=True,
            ),
        )

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), patch(
            "api.agent.core.prompt_context.ensure_comms_compacted"
        ):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn("reasoning-only, no user-visible action or tool call", content)
        self.assertIn("I should send the answer now.", content)

    def test_prompt_context_overfetches_past_filtered_reasoning_steps(self):
        """Filtered older reasoning should not crowd out older visible steps."""
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=5,
            hysteresis=2,
            reasoning_limit=0,
        )

        base_time = timezone.now()
        crowding_reasoning_count = 12
        newest_visible_step = PersistentAgentStep.objects.create(
            agent=self.agent,
            description="Newest visible step",
        )
        PersistentAgentStep.objects.filter(pk=newest_visible_step.pk).update(
            created_at=base_time + timedelta(seconds=crowding_reasoning_count + 1)
        )
        for idx in range(crowding_reasoning_count):
            step = PersistentAgentStep.objects.create(
                agent=self.agent,
                description=f"{INTERNAL_REASONING_PREFIX} hidden reasoning {idx}",
            )
            PersistentAgentStep.objects.filter(pk=step.pk).update(
                created_at=base_time + timedelta(seconds=idx + 1)
            )

        visible_step = PersistentAgentStep.objects.create(
            agent=self.agent,
            description="Older visible step",
        )
        PersistentAgentStep.objects.filter(pk=visible_step.pk).update(
            created_at=base_time
        )

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), patch(
            "api.agent.core.prompt_context.ensure_comms_compacted"
        ):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn("Newest visible step", content)
        self.assertIn("Older visible step", content)
        self.assertNotIn("hidden reasoning", content)

    def test_prompt_context_preserves_latest_reasoning_only_completion_when_reasoning_limit_zero(self):
        self._configure_unified_history_limits(
            tool_limit=5,
            unified_limit=5,
            hysteresis=2,
            reasoning_limit=0,
        )

        base_time = timezone.now()
        preserved_completion = PersistentAgentCompletion.objects.create(agent=self.agent)
        preserved_step = PersistentAgentStep.objects.create(
            agent=self.agent,
            completion=preserved_completion,
            description=build_internal_reasoning_description(
                "preserved reasoning",
                reasoning_only=True,
            ),
        )
        PersistentAgentStep.objects.filter(pk=preserved_step.pk).update(created_at=base_time)

        tool_completion = PersistentAgentCompletion.objects.create(agent=self.agent)
        filtered_reasoning_step = PersistentAgentStep.objects.create(
            agent=self.agent,
            completion=tool_completion,
            description=build_internal_reasoning_description("filtered reasoning"),
        )
        PersistentAgentStep.objects.filter(pk=filtered_reasoning_step.pk).update(
            created_at=base_time + timedelta(seconds=1)
        )
        tool_step = PersistentAgentStep.objects.create(
            agent=self.agent,
            completion=tool_completion,
            description="Did a tool-backed follow-up",
        )
        PersistentAgentToolCall.objects.create(
            step=tool_step,
            tool_name="search_tools",
            tool_params={"query": "competitors"},
            result=json.dumps({"status": "ok"}),
        )
        PersistentAgentStep.objects.filter(pk=tool_step.pk).update(
            created_at=base_time + timedelta(seconds=2)
        )

        for idx in range(5):
            visible_step = PersistentAgentStep.objects.create(
                agent=self.agent,
                description=f"Visible step {idx}",
            )
            PersistentAgentStep.objects.filter(pk=visible_step.pk).update(
                created_at=base_time + timedelta(seconds=idx + 3)
            )

        with patch("api.agent.core.prompt_context.ensure_steps_compacted"), patch(
            "api.agent.core.prompt_context.ensure_comms_compacted"
        ):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        content = user_message["content"]
        self.assertIn("preserved reasoning", content)
        self.assertNotIn("filtered reasoning", content)

    def test_mcp_servers_listed_in_prompt(self):
        """Accessible MCP servers should be enumerated in the prompt context."""
        MCPServerConfig.objects.create(
            scope=MCPServerConfig.Scope.PLATFORM,
            name="test-sheets",
            display_name="Test Sheets",
            url="https://mcp.example.com",
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        user_message = next((m for m in context if m['role'] == 'user'), None)
        self.assertIsNotNone(user_message)
        content = user_message['content']
        self.assertIn("These are the MCP servers you have access to.", content)
        self.assertIn("Test Sheets", content)
        self.assertIn("search_tools", content)

    def test_admin_system_message_is_injected_once(self):
        """Admin-authored system directives should appear in the system prompt and be marked delivered."""
        directive = PersistentAgentSystemMessage.objects.create(
            agent=self.agent,
            body="Drop everything and update the quarterly results deck today.",
            created_by=self.user,
        )

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, _ = build_prompt_context(self.agent)

        system_message = next((m for m in context if m['role'] == 'system'), None)
        self.assertIsNotNone(system_message)
        content = system_message['content']
        self.assertIn("A note from the Gobii team:", content)
        self.assertIn("Drop everything and update the quarterly results deck today.", content)

        sys_steps = PersistentAgentSystemStep.objects.filter(
            code=PersistentAgentSystemStep.Code.SYSTEM_DIRECTIVE,
            step__agent=self.agent,
        )
        self.assertEqual(sys_steps.count(), 1)
        self.assertIn("Drop everything and update the quarterly results deck today.", sys_steps.first().step.description)

        directive.refresh_from_db()
        self.assertIsNotNone(directive.delivered_at)

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            second_context, _, _ = build_prompt_context(self.agent)

        second_system = next((m for m in second_context if m['role'] == 'system'), None)
        self.assertIsNotNone(second_system)
        self.assertNotIn("Drop everything and update the quarterly results deck today.", second_system['content'])
        self.assertEqual(
            PersistentAgentSystemStep.objects.filter(
                code=PersistentAgentSystemStep.Code.SYSTEM_DIRECTIVE,
                step__agent=self.agent,
            ).count(),
            1,
        )

    def test_prompt_archive_saved_to_storage(self):
        """Prompt archives should be written to object storage as compressed JSON."""
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'):
            context, _, prompt_archive_id = build_prompt_context(self.agent)

        archive_dir = f"persistent_agents/{self.agent.id}/prompt_archives"
        _, files = self._storage.listdir(archive_dir)
        self.assertEqual(len(files), 1, "Expected a single prompt archive file")
        archive_path = f"{archive_dir}/{files[0]}"

        with self._storage.open(archive_path, "rb") as fh:
            compressed_bytes = fh.read()

        decompressed = zstd.ZstdDecompressor().decompress(compressed_bytes)
        payload = json.loads(decompressed.decode("utf-8"))

        self.assertEqual(payload["agent_id"], str(self.agent.id))
        self.assertEqual(payload["token_budget"], get_prompt_token_budget(self.agent))
        self.assertIn("system_prompt", payload)
        self.assertIn("user_prompt", payload)
        user_message = next((m for m in context if m["role"] == "user"), None)
        self.assertIsNotNone(user_message)
        self.assertEqual(payload["user_prompt"], user_message["content"])
        self.assertEqual(PersistentAgentPromptArchive.objects.count(), 1)
        archive_row = PersistentAgentPromptArchive.objects.get(agent=self.agent)
        self.assertEqual(archive_row.storage_key, archive_path)
        self.assertEqual(archive_row.raw_bytes, len(decompressed))
        self.assertEqual(archive_row.compressed_bytes, len(compressed_bytes))
        self.assertEqual(archive_row.tokens_before, payload["tokens_before"])
        self.assertEqual(archive_row.tokens_after, payload["tokens_after"])
        self.assertEqual(archive_row.tokens_saved, payload["tokens_saved"])

        admin_user = User.objects.create_superuser(
            username="prompt_archive_admin",
            email="prompt_archive_admin@example.com",
            password="secret",
        )
        request = RequestFactory().get("/")
        request.user = admin_user
        admin_view = PersistentAgentPromptArchiveAdmin(PersistentAgentPromptArchive, django_admin.site)
        response = admin_view.download_view(request, archive_row.pk)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response["Content-Type"], "application/json")
        self.assertIn(".json", response["Content-Disposition"])
        downloaded_bytes = b"".join(response.streaming_content)
        self.assertEqual(downloaded_bytes, decompressed)

    def test_prompt_archive_links_to_step(self):
        """Running the agent loop should attach the prompt archive to the first generated step."""
        response_message = MagicMock()
        response_message.tool_calls = None
        response_message.function_call = None
        response_message.content = "Reasoning output"
        response_choice = MagicMock(message=response_message)
        response = MagicMock()
        response.choices = [response_choice]
        response.model_extra = {
            "usage": MagicMock(
                prompt_tokens=12,
                completion_tokens=6,
                total_tokens=18,
                prompt_tokens_details=MagicMock(cached_tokens=0),
            )
        }

        token_usage = {
            "prompt_tokens": 12,
            "completion_tokens": 6,
            "total_tokens": 18,
            "model": "mock-model",
            "provider": "mock-provider",
            "cached_tokens": 0,
        }

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch('api.agent.core.prompt_context.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]):
            with patch('api.agent.core.event_processing._completion_with_failover', return_value=(response, token_usage)):
                from api.agent.core import event_processing as ep
                with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 1):
                    _run_agent_loop(self.agent, is_first_run=False)

        archive = PersistentAgentPromptArchive.objects.get(agent=self.agent)
        self.assertIsNotNone(archive.step, "Prompt archive should be linked to a step")
        linked_archive = PersistentAgentPromptArchive.objects.get(step=archive.step)
        self.assertEqual(linked_archive.id, archive.id)
        self.assertIsNotNone(archive.step.completion)
        self.assertEqual(archive.step.completion.prompt_tokens, token_usage["prompt_tokens"])

    def test_agent_loop_passes_preferred_provider(self):
        """Agent loop should forward the preferred provider returned by the helper."""
        response_message = MagicMock()
        response_message.tool_calls = None
        response_message.function_call = None
        response_message.content = "Reasoning output"
        response_choice = MagicMock(message=response_message)
        response = MagicMock()
        response.choices = [response_choice]
        response.model_extra = {
            "usage": MagicMock(
                prompt_tokens=12,
                completion_tokens=6,
                total_tokens=18,
                prompt_tokens_details=MagicMock(cached_tokens=0),
            )
        }
        token_usage = {
            "prompt_tokens": 12,
            "completion_tokens": 6,
            "total_tokens": 18,
            "model": "mock-model",
            "provider": "mock-provider",
            "cached_tokens": 0,
        }
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch('api.agent.core.prompt_context.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
             patch('api.agent.core.event_processing._get_recent_preferred_config', return_value=("mock", "mock-model")) as mock_helper, \
             patch('api.agent.core.event_processing._completion_with_failover', return_value=(response, token_usage)) as mock_completion:
            from api.agent.core import event_processing as ep
            with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 1):
                _run_agent_loop(self.agent, is_first_run=False, run_sequence_number=3)

        mock_helper.assert_called_once_with(agent=self.agent, run_sequence_number=3)
        call_kwargs = mock_completion.call_args.kwargs
        self.assertEqual(call_kwargs["preferred_config"], ("mock", "mock-model"))

    def test_agent_loop_skips_preference_on_second_run(self):
        """Agent loop should not request preferred configs on the agent's second run."""
        response_message = MagicMock()
        response_message.tool_calls = None
        response_message.function_call = None
        response_message.content = "Reasoning output"
        response_choice = MagicMock(message=response_message)
        response = MagicMock()
        response.choices = [response_choice]
        response.model_extra = {}
        token_usage = {
            "model": "mock-model",
            "provider": "mock-provider",
        }
        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch('api.agent.core.prompt_context.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
             patch('api.agent.core.event_processing._get_recent_preferred_config', return_value=None) as mock_helper, \
             patch('api.agent.core.event_processing._completion_with_failover', return_value=(response, token_usage)) as mock_completion:
            from api.agent.core import event_processing as ep
            with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 1):
                _run_agent_loop(self.agent, is_first_run=False, run_sequence_number=2)

        mock_helper.assert_called_once_with(agent=self.agent, run_sequence_number=2)
        call_kwargs = mock_completion.call_args.kwargs
        self.assertIsNone(call_kwargs["preferred_config"])

    def test_completion_record_keeps_model_when_usage_missing(self):
        """PersistentAgentCompletion should store provider/model even if usage isn't provided."""
        response_message = MagicMock()
        response_message.tool_calls = None
        response_message.function_call = None
        response_message.content = "Reasoning output"
        response_choice = MagicMock(message=response_message)
        response = MagicMock()
        response.choices = [response_choice]
        response.model_extra = {}

        token_usage = {
            "model": "mock-model",
            "provider": "mock-provider",
        }

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch('api.agent.core.prompt_context.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
             patch('api.agent.core.event_processing._completion_with_failover', return_value=(response, token_usage)):
            from api.agent.core import event_processing as ep
            with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 1):
                _run_agent_loop(self.agent, is_first_run=False)

        completion = PersistentAgentCompletion.objects.get(agent=self.agent)
        self.assertEqual(completion.llm_model, "mock-model")
        self.assertEqual(completion.llm_provider, "mock-provider")

    def test_completion_record_stores_billing_snapshot(self):
        """Orchestrator completions should persist the owner billing snapshot."""
        response_message = MagicMock()
        response_message.tool_calls = None
        response_message.function_call = None
        response_message.content = "Reasoning output"
        response_choice = MagicMock(message=response_message)
        response = MagicMock()
        response.choices = [response_choice]
        response.model_extra = {}

        token_usage = {
            "model": "mock-model",
            "provider": "mock-provider",
        }

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch('api.agent.core.prompt_context.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
             patch('api.agent.core.event_processing._completion_with_failover', return_value=(response, token_usage)), \
             patch(
                 'api.agent.core.event_processing.get_billing_snapshot_for_owner',
                 return_value={"billing_plan": "scale", "billing_is_trial": False},
             ):
            from api.agent.core import event_processing as ep
            with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 1):
                _run_agent_loop(self.agent, is_first_run=False)

        completion = PersistentAgentCompletion.objects.get(agent=self.agent)
        self.assertEqual(completion.billing_plan, "scale")
        self.assertFalse(completion.billing_is_trial)

    def test_completion_record_reuses_billing_snapshot_across_iterations(self):
        """Billing snapshot should be resolved once per run even if the loop iterates."""
        enable_tools(self.agent, ["sqlite_batch"])

        tool_call = MagicMock()
        tool_call.function = MagicMock()
        tool_call.function.name = "sqlite_batch"
        tool_call.function.arguments = '{"sql": "UPDATE t SET id = 1", "will_continue_work": false}'

        message_first = MagicMock()
        message_first.tool_calls = [tool_call]
        message_first.function_call = None
        message_first.content = None
        response_first = MagicMock()
        response_first.choices = [MagicMock(message=message_first)]
        response_first.model_extra = {}

        message_second = MagicMock()
        message_second.tool_calls = None
        message_second.function_call = None
        message_second.content = "done"
        response_second = MagicMock()
        response_second.choices = [MagicMock(message=message_second)]
        response_second.model_extra = {}

        token_usage = {
            "prompt_tokens": 5,
            "completion_tokens": 5,
            "total_tokens": 10,
            "model": "mock-model",
            "provider": "mock-provider",
            "cached_tokens": 0,
        }

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch('api.agent.core.event_processing.build_prompt_context', return_value=([{"role": "system", "content": "sys"}], 1000, None)), \
             patch('api.agent.core.event_processing.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
             patch('api.agent.core.event_processing._completion_with_failover', side_effect=[(response_first, token_usage), (response_second, token_usage)]) as mock_completion, \
             patch('api.agent.core.event_processing.execute_enabled_tool', return_value={"status": "warning", "message": "0 rows affected"}), \
             patch('api.agent.core.event_processing._ensure_credit_for_tool', return_value={"cost": None, "credit": None}), \
             patch(
                 'api.agent.core.event_processing.get_billing_snapshot_for_owner',
                 return_value={"billing_plan": "scale", "billing_is_trial": False},
             ) as mock_billing_snapshot:
            from api.agent.core import event_processing as ep
            with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 2):
                _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(mock_completion.call_count, 2)
        mock_billing_snapshot.assert_called_once_with(self.user)

    def test_agent_loop_excludes_enable_database_from_tools(self):
        """LLM tool payload should not expose enable_database."""
        enable_tools(self.agent, ["sqlite_batch"])

        response_message = MagicMock()
        response_message.tool_calls = None
        response_message.function_call = None
        response_message.content = "Reasoning output"
        response_choice = MagicMock(message=response_message)
        response = MagicMock()
        response.choices = [response_choice]
        response.model_extra = {}
        token_usage = {"model": "mock-model", "provider": "mock-provider"}

        with patch('api.agent.core.event_processing.build_prompt_context', return_value=([{"role": "system", "content": "sys"}], 1000, None)), \
             patch('api.agent.core.event_processing.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
             patch('api.agent.core.event_processing._completion_with_failover', return_value=(response, token_usage)) as mock_completion:
            from api.agent.core import event_processing as ep
            with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 1):
                _run_agent_loop(self.agent, is_first_run=False)

        passed_tools = mock_completion.call_args.kwargs["tools"]
        tool_names = [
            entry.get("function", {}).get("name")
            for entry in passed_tools
            if isinstance(entry, dict)
        ]
        completion = PersistentAgentCompletion.objects.get(agent=self.agent)
        self.assertEqual(completion.llm_tool_names, tool_names)
        self.assertNotIn("enable_database", tool_names)
        self.assertIn("sqlite_batch", tool_names)

    def test_warning_status_requires_followup(self):
        """Tool warning status should trigger another loop iteration."""
        enable_tools(self.agent, ["sqlite_batch"])

        tool_call = MagicMock()
        tool_call.function = MagicMock()
        tool_call.function.name = "sqlite_batch"
        tool_call.function.arguments = '{"sql": "UPDATE t SET id = 1", "will_continue_work": false}'

        message_first = MagicMock()
        message_first.tool_calls = [tool_call]
        message_first.function_call = None
        message_first.content = None

        choice_first = MagicMock(message=message_first)
        response_first = MagicMock()
        response_first.choices = [choice_first]
        response_first.model_extra = {
            "usage": MagicMock(
                prompt_tokens=5,
                completion_tokens=5,
                total_tokens=10,
                prompt_tokens_details=MagicMock(cached_tokens=0),
            )
        }

        message_second = MagicMock()
        message_second.tool_calls = None
        message_second.function_call = None
        message_second.content = "done"

        choice_second = MagicMock(message=message_second)
        response_second = MagicMock()
        response_second.choices = [choice_second]
        response_second.model_extra = {
            "usage": MagicMock(
                prompt_tokens=5,
                completion_tokens=5,
                total_tokens=10,
                prompt_tokens_details=MagicMock(cached_tokens=0),
            )
        }

        token_usage = {
            "prompt_tokens": 5,
            "completion_tokens": 5,
            "total_tokens": 10,
            "model": "mock-model",
            "provider": "mock-provider",
            "cached_tokens": 0,
        }

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch('api.agent.core.event_processing.build_prompt_context', return_value=([{"role": "system", "content": "sys"}], 1000, None)), \
             patch('api.agent.core.event_processing.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
             patch('api.agent.core.event_processing._completion_with_failover', side_effect=[(response_first, token_usage), (response_second, token_usage)]) as mock_completion, \
             patch('api.agent.core.event_processing.execute_enabled_tool', return_value={"status": "warning", "message": "0 rows affected"}), \
             patch('api.agent.core.event_processing._ensure_credit_for_tool', return_value={"cost": None, "credit": None}):
            from api.agent.core import event_processing as ep
            with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 2):
                _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(mock_completion.call_count, 2, "Warning status should force a follow-up iteration.")

    def test_web_session_activation_retry_discards_completion_and_retries_once(self):
        """When switch is on and web session appears mid-iteration, discard and rerun once."""
        EmailAddress.objects.create(
            user=self.user,
            email=self.user.email,
            verified=True,
            primary=True,
        )
        tool_call = MagicMock()
        tool_call.function = MagicMock()
        tool_call.function.name = "sqlite_batch"
        tool_call.function.arguments = '{"sql":"SELECT 1","will_continue_work":false}'

        message_first = MagicMock()
        message_first.tool_calls = [tool_call]
        message_first.function_call = None
        message_first.content = None
        response_first = MagicMock()
        response_first.choices = [MagicMock(message=message_first)]
        response_first.model_extra = {}

        message_second = MagicMock()
        message_second.tool_calls = None
        message_second.function_call = None
        message_second.content = None
        response_second = MagicMock()
        response_second.choices = [MagicMock(message=message_second)]
        response_second.model_extra = {}

        token_usage = {"model": "mock-model", "provider": "mock-provider"}
        seen_tool_names = []

        def completion_side_effect(*args, **kwargs):
            names = [
                entry.get("function", {}).get("name")
                for entry in kwargs.get("tools", [])
                if isinstance(entry, dict)
            ]
            seen_tool_names.append(names)
            if len(seen_tool_names) == 1:
                start_web_session(self.agent, self.user)
                return response_first, token_usage
            return response_second, token_usage

        with override_switch("agent_retry_completion_on_web_session_activation", active=True):
            with patch('api.agent.core.event_processing.build_prompt_context', return_value=([{"role": "system", "content": "sys"}], 1000, None)), \
                 patch('api.agent.core.event_processing.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
                 patch('api.agent.core.event_processing._completion_with_failover', side_effect=completion_side_effect) as mock_completion, \
                 patch('api.agent.core.event_processing.execute_enabled_tool', return_value={"status": "ok"}) as mock_execute_tool, \
                 patch('api.agent.core.event_processing._ensure_credit_for_tool', return_value={"cost": None, "credit": None}), \
                 patch('api.agent.core.event_processing.Analytics.track_event') as mock_track_event:
                from api.agent.core import event_processing as ep
                with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 2):
                    _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(mock_completion.call_count, 2)
        self.assertEqual(mock_execute_tool.call_count, 0)
        self.assertNotIn("send_chat_message", seen_tool_names[0])
        self.assertIn("send_chat_message", seen_tool_names[1])
        self.assertEqual(mock_track_event.call_count, 1)
        self.assertEqual(
            mock_track_event.call_args.kwargs["event"],
            ep.AnalyticsEvent.PERSISTENT_AGENT_WEB_SESSION_ACTIVATED_POST_COMPLETION,
        )
        retry_props = mock_track_event.call_args.kwargs["properties"]
        self.assertEqual(retry_props.get("retry_reason"), "web_session_activated_mid_completion")
        self.assertEqual(retry_props.get("retry_strategy"), "discard_and_rerun_once")
        self.assertEqual(retry_props.get("retry_switch_active"), True)
        self.assertEqual(retry_props.get("retry_performed"), True)
        self.assertEqual(retry_props.get("had_deliverable_web_target_at_start"), False)

    def test_web_session_activation_retry_does_not_run_when_switch_off(self):
        """When switch is off, completion output is processed normally."""
        tool_call = MagicMock()
        tool_call.function = MagicMock()
        tool_call.function.name = "sqlite_batch"
        tool_call.function.arguments = '{"sql":"SELECT 1","will_continue_work":false}'

        message_first = MagicMock()
        message_first.tool_calls = [tool_call]
        message_first.function_call = None
        message_first.content = None
        response_first = MagicMock()
        response_first.choices = [MagicMock(message=message_first)]
        response_first.model_extra = {}

        token_usage = {"model": "mock-model", "provider": "mock-provider"}

        def completion_side_effect(*args, **kwargs):
            start_web_session(self.agent, self.user)
            return response_first, token_usage

        with override_switch("agent_retry_completion_on_web_session_activation", active=False):
            with patch('api.agent.core.event_processing.build_prompt_context', return_value=([{"role": "system", "content": "sys"}], 1000, None)), \
                 patch('api.agent.core.event_processing.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
                 patch('api.agent.core.event_processing._completion_with_failover', side_effect=completion_side_effect) as mock_completion, \
                 patch('api.agent.core.event_processing.execute_enabled_tool', return_value={"status": "ok", "auto_sleep_ok": True}) as mock_execute_tool, \
                 patch('api.agent.core.event_processing._ensure_credit_for_tool', return_value={"cost": None, "credit": None}), \
                 patch('api.agent.core.event_processing.Analytics.track_event') as mock_track_event:
                from api.agent.core import event_processing as ep
                with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 2):
                    _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(mock_completion.call_count, 1)
        self.assertEqual(mock_execute_tool.call_count, 1)
        self.assertEqual(mock_track_event.call_count, 1)
        self.assertEqual(
            mock_track_event.call_args.kwargs["event"],
            ep.AnalyticsEvent.PERSISTENT_AGENT_WEB_SESSION_ACTIVATED_POST_COMPLETION,
        )
        retry_props = mock_track_event.call_args.kwargs["properties"]
        self.assertEqual(retry_props.get("retry_strategy"), "none")
        self.assertEqual(retry_props.get("retry_switch_active"), False)
        self.assertEqual(retry_props.get("retry_performed"), False)

    def test_web_session_activation_retry_emits_expected_analytics_event(self):
        """Retry path should emit the dedicated analytics event with expected properties."""
        message_first = MagicMock()
        message_first.tool_calls = None
        message_first.function_call = None
        message_first.content = ""
        response_first = MagicMock()
        response_first.choices = [MagicMock(message=message_first)]
        response_first.model_extra = {}

        message_second = MagicMock()
        message_second.tool_calls = None
        message_second.function_call = None
        message_second.content = None
        response_second = MagicMock()
        response_second.choices = [MagicMock(message=message_second)]
        response_second.model_extra = {}

        token_usage = {"model": "mock-model", "provider": "mock-provider"}

        def completion_side_effect(*args, **kwargs):
            if not hasattr(completion_side_effect, "called"):
                completion_side_effect.called = True
                start_web_session(self.agent, self.user)
                return response_first, token_usage
            return response_second, token_usage

        with override_switch("agent_retry_completion_on_web_session_activation", active=True):
            with patch('api.agent.core.event_processing.build_prompt_context', return_value=([{"role": "system", "content": "sys"}], 1000, None)), \
                 patch('api.agent.core.event_processing.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
                 patch('api.agent.core.event_processing._completion_with_failover', side_effect=completion_side_effect), \
                 patch('api.agent.core.event_processing._ensure_credit_for_tool', return_value={"cost": None, "credit": None}), \
                 patch('api.agent.core.event_processing.Analytics.track_event') as mock_track_event:
                from api.agent.core import event_processing as ep
                with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 2):
                    _run_agent_loop(self.agent, is_first_run=False, run_sequence_number=7)

        self.assertEqual(mock_track_event.call_count, 1)
        kwargs = mock_track_event.call_args.kwargs
        self.assertEqual(
            kwargs["event"],
            ep.AnalyticsEvent.PERSISTENT_AGENT_WEB_SESSION_ACTIVATED_POST_COMPLETION,
        )
        self.assertEqual(kwargs["source"], ep.AnalyticsSource.AGENT)
        props = kwargs["properties"]
        self.assertEqual(props.get("agent_id"), str(self.agent.id))
        self.assertEqual(props.get("agent_name"), self.agent.name)
        self.assertEqual(props.get("run_sequence_number"), 7)
        self.assertEqual(props.get("iteration"), 1)
        self.assertEqual(props.get("retry_reason"), "web_session_activated_mid_completion")
        self.assertEqual(props.get("retry_strategy"), "discard_and_rerun_once")
        self.assertEqual(props.get("retry_switch_active"), True)
        self.assertEqual(props.get("retry_performed"), True)
        self.assertEqual(props.get("had_deliverable_web_target_at_start"), False)

    def test_web_session_activation_retry_skips_when_no_iterations_remaining(self):
        """If no loop iteration remains, process current completion instead of dropping it."""
        tool_call = MagicMock()
        tool_call.function = MagicMock()
        tool_call.function.name = "sqlite_batch"
        tool_call.function.arguments = '{"sql":"SELECT 1","will_continue_work":false}'

        message_first = MagicMock()
        message_first.tool_calls = [tool_call]
        message_first.function_call = None
        message_first.content = None
        response_first = MagicMock()
        response_first.choices = [MagicMock(message=message_first)]
        response_first.model_extra = {}

        token_usage = {"model": "mock-model", "provider": "mock-provider"}

        def completion_side_effect(*args, **kwargs):
            start_web_session(self.agent, self.user)
            return response_first, token_usage

        with override_switch("agent_retry_completion_on_web_session_activation", active=True):
            with patch('api.agent.core.event_processing.build_prompt_context', return_value=([{"role": "system", "content": "sys"}], 1000, None)), \
                 patch('api.agent.core.event_processing.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
                 patch('api.agent.core.event_processing._completion_with_failover', side_effect=completion_side_effect) as mock_completion, \
                 patch('api.agent.core.event_processing.execute_enabled_tool', return_value={"status": "ok"}) as mock_execute_tool, \
                 patch('api.agent.core.event_processing._ensure_credit_for_tool', return_value={"cost": None, "credit": None}):
                from api.agent.core import event_processing as ep
                with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 1):
                    _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(mock_completion.call_count, 1)
        self.assertEqual(mock_execute_tool.call_count, 1)


@tag("batch_event_processing")
class AgentRunSequenceHelperTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="sequence_tester@example.com",
            email="sequence_tester@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="SeqBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="SeqAgent",
            charter="Test run sequence helper",
            browser_use_agent=self.browser_agent,
        )

    def test_completed_run_count_ignores_non_processing_steps(self):
        """Helper should ignore credit gate PROCESS_EVENTS system steps."""
        skipped_step = PersistentAgentStep.objects.create(
            agent=self.agent,
            description="Skipped due to credits",
        )
        PersistentAgentSystemStep.objects.create(
            step=skipped_step,
            code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
            notes="credit_insufficient",
        )

        self.assertEqual(_get_completed_process_run_count(self.agent), 0)

        run_step = PersistentAgentStep.objects.create(
            agent=self.agent,
            description="Process events",
        )
        PersistentAgentSystemStep.objects.create(
            step=run_step,
            code=PersistentAgentSystemStep.Code.PROCESS_EVENTS,
            notes="simplified",
        )

        self.assertEqual(_get_completed_process_run_count(self.agent), 1)

@tag("batch_event_processing")
class CronTriggerTaskTests(TestCase):
    """Unit tests for the cron trigger task."""

    def setUp(self):
        self.user = User.objects.create_user(
            username="cron_tester@example.com",
            email="cron_tester@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="CronBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="CronAgent",
            charter="cron test agent",
            browser_use_agent=self.browser_agent,
        )

    @patch('api.agent.tasks.process_events.process_agent_events')
    def test_cron_trigger_task_creates_trigger_record(self, mock_process_events):
        """Test that process_agent_cron_trigger_task creates the cron trigger record."""
        cron_expression = "@daily"
        
        # Verify no cron triggers exist initially
        self.assertEqual(PersistentAgentCronTrigger.objects.count(), 0)
        
        # Call the cron trigger task
        process_agent_cron_trigger_task(str(self.agent.id), cron_expression)
        
        # Verify cron trigger was created
        self.assertEqual(PersistentAgentCronTrigger.objects.count(), 1)
        
        cron_trigger = PersistentAgentCronTrigger.objects.first()
        self.assertEqual(cron_trigger.cron_expression, cron_expression)
        self.assertEqual(cron_trigger.step.agent, self.agent)
        self.assertEqual(cron_trigger.step.description, f"Cron trigger: {cron_expression}")
        
        # Verify process_agent_events was called
        mock_process_events.assert_called_once_with(str(self.agent.id))

    @patch('api.agent.tasks.process_events.process_agent_events')
    def test_cron_trigger_task_with_complex_expression(self, mock_process_events):
        """Test that cron trigger task works with complex cron expressions."""
        cron_expression = "0 9 * * 1-5"  # Weekdays at 9am
        
        # Call the cron trigger task
        process_agent_cron_trigger_task(str(self.agent.id), cron_expression)
        
        # Verify cron trigger was created with correct expression
        cron_trigger = PersistentAgentCronTrigger.objects.first()
        self.assertEqual(cron_trigger.cron_expression, cron_expression)
        
        # Verify process_agent_events was called
        mock_process_events.assert_called_once_with(str(self.agent.id))

    @patch('api.agent.tasks.process_events.process_agent_events')
    def test_cron_trigger_task_skips_paused_owner(self, mock_process_events):
        UserBilling.objects.update_or_create(
            user=self.user,
            defaults={
                "execution_paused": True,
                "execution_pause_reason": "billing_delinquency",
                "execution_paused_at": timezone.now(),
            },
        )

        process_agent_cron_trigger_task(str(self.agent.id), "@daily")

        self.assertEqual(PersistentAgentCronTrigger.objects.count(), 0)
        mock_process_events.assert_not_called()

    @patch('api.agent.tasks.process_events.logger')
    @patch('api.agent.tasks.process_events.process_agent_events')
    def test_cron_trigger_task_logs_quota_validation_as_info(self, mock_process_events, mock_logger):
        """Quota ValidationErrors should be logged as info instead of failing the task."""
        quota_error = ValidationError(
            {"quota": ["Task quota exceeded. You have no remaining task credits and no active subscription."]}
        )
        mock_process_events.side_effect = quota_error
        cron_expression = "@hourly"

        process_agent_cron_trigger_task(str(self.agent.id), cron_expression)

        self.assertEqual(PersistentAgentCronTrigger.objects.count(), 1)
        mock_process_events.assert_called_once_with(str(self.agent.id))
        mock_logger.info.assert_any_call(
            "Skipping cron trigger for agent %s due to task quota: %s",
            str(self.agent.id),
            quota_error,
        )

    @patch('api.agent.tasks.process_events._remove_orphaned_celery_beat_task')
    @patch('api.agent.tasks.process_events.process_agent_events')
    def test_cron_trigger_task_handles_nonexistent_agent(self, mock_process_events, mock_remove_beat_task):
        """Test that cron trigger task handles non-existent agents by removing orphaned beat tasks."""
        # Use a non-existent agent ID
        nonexistent_agent_id = "00000000-0000-0000-0000-000000000000"
        cron_expression = "@daily"
        
        # Verify no cron triggers exist initially
        initial_count = PersistentAgentCronTrigger.objects.count()
        
        # Call the cron trigger task with non-existent agent ID
        process_agent_cron_trigger_task(nonexistent_agent_id, cron_expression)
        
        # Verify no cron trigger was created
        self.assertEqual(PersistentAgentCronTrigger.objects.count(), initial_count)
        
        # Verify process_agent_events was NOT called
        mock_process_events.assert_not_called()
        
        # Verify orphaned beat task removal was called
        mock_remove_beat_task.assert_called_once_with(nonexistent_agent_id)

    @patch('api.agent.tasks.process_events.switch_is_active', return_value=True)
    @patch('config.redis_client.get_redis_client')
    @patch('api.agent.tasks.process_events.process_agent_events')
    def test_cron_trigger_task_throttles_old_free_agents_and_marks_footer_pending(
        self,
        mock_process_events,
        mock_get_redis,
        _mock_switch,
    ):
        """Second cron trigger should be skipped when throttled, and a footer notice marked pending."""

        class _FakeRedis:
            def __init__(self):
                self._store = {}

            def get(self, key):
                return self._store.get(key)

            def set(self, key, value, ex=None, nx=None):
                if nx and key in self._store:
                    return False
                self._store[key] = value
                return True

            def delete(self, key):
                self._store.pop(key, None)
                return 1

            def exists(self, key):
                return 1 if key in self._store else 0

        fake_redis = _FakeRedis()
        mock_get_redis.return_value = fake_redis

        # Age the agent so throttling is active (defaults: start at 16 days)
        old_ts = timezone.now() - timedelta(days=20)
        PersistentAgent.objects.filter(pk=self.agent.pk).update(created_at=old_ts, schedule="@daily")
        self.agent.refresh_from_db()

        cron_expression = "@daily"

        # First run allowed
        process_agent_cron_trigger_task(str(self.agent.id), cron_expression)
        self.assertEqual(PersistentAgentCronTrigger.objects.count(), 1)
        mock_process_events.assert_called_once_with(str(self.agent.id))

        # Second run skipped due to redis gate (NX=False)
        process_agent_cron_trigger_task(str(self.agent.id), cron_expression)
        self.assertEqual(PersistentAgentCronTrigger.objects.count(), 1)
        mock_process_events.assert_called_once_with(str(self.agent.id))

        from api.services.cron_throttle import cron_throttle_pending_footer_key
        pending_key = cron_throttle_pending_footer_key(str(self.agent.id))
        self.assertTrue(fake_redis.get(pending_key))

    @patch('redbeat.RedBeatSchedulerEntry.from_key')
    @patch('celery.current_app')
    def test_remove_orphaned_celery_beat_task_success(self, mock_celery_app, mock_from_key):
        """Test successful removal of orphaned Celery beat task."""
        # Setup mocks
        mock_entry = mock_from_key.return_value
        
        agent_id = "test-agent-id"
        expected_task_name = f"persistent-agent-schedule:{agent_id}"
        expected_key = f"redbeat:{expected_task_name}"
        
        # Call the function
        with patch('api.agent.tasks.process_events.logger') as mock_logger:
            _remove_orphaned_celery_beat_task(agent_id)
        
        # Verify RedBeatSchedulerEntry.from_key was called with correct parameters
        mock_from_key.assert_called_once_with(expected_key, app=mock_celery_app)
        
        # Verify entry.delete() was called
        mock_entry.delete.assert_called_once()
        
        # Verify success was logged
        mock_logger.info.assert_called_once_with(
            "Removed orphaned Celery Beat task for non-existent agent %s", agent_id
        )

    @patch('redbeat.RedBeatSchedulerEntry.from_key')
    @patch('celery.current_app')
    def test_remove_orphaned_celery_beat_task_key_error(self, mock_celery_app, mock_from_key):
        """Test handling of KeyError when beat task doesn't exist."""
        # Setup mocks - simulate KeyError when task doesn't exist
        mock_from_key.side_effect = KeyError("Task not found")
        
        agent_id = "test-agent-id"
        
        # Call the function
        with patch('api.agent.tasks.process_events.logger') as mock_logger:
            _remove_orphaned_celery_beat_task(agent_id)
        
        # Verify appropriate message was logged
        mock_logger.info.assert_called_once_with(
            "No Celery Beat task found for non-existent agent %s", agent_id
        )

    @patch('redbeat.RedBeatSchedulerEntry.from_key')
    @patch('celery.current_app')
    def test_remove_orphaned_celery_beat_task_general_error(self, mock_celery_app, mock_from_key):
        """Test handling of general exceptions during beat task removal."""
        # Setup mocks - simulate general exception
        mock_from_key.side_effect = Exception("Redis connection failed")
        
        agent_id = "test-agent-id"
        
        # Call the function
        with patch('api.agent.tasks.process_events.logger') as mock_logger:
            _remove_orphaned_celery_beat_task(agent_id)
        
        # Verify error was logged
        mock_logger.error.assert_called_once_with(
            "Error removing orphaned Celery Beat task for agent %s: %s", 
            agent_id, 
            mock_from_key.side_effect
        ) 


@tag("batch_event_processing")
class UpdateScheduleMinimumIntervalTests(TestCase):
    """Unit tests for _execute_update_schedule minimum interval validation."""

    def setUp(self):
        self.user = User.objects.create_user(
            username="schedule_tester@example.com",
            email="schedule_tester@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="ScheduleBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="ScheduleAgent",
            charter="test schedule updates",
            browser_use_agent=self.browser_agent,
            schedule="@daily",  # Start with a valid schedule
        )
        invalidate_tool_settings_cache()
        self.tool_config, _ = ToolConfig.objects.get_or_create(
            plan_name=PlanNamesChoices.FREE,
            defaults={"min_cron_schedule_minutes": DEFAULT_MIN_CRON_SCHEDULE_MINUTES},
        )
        self.original_min_interval = self.tool_config.min_cron_schedule_minutes
        self.tool_config.min_cron_schedule_minutes = DEFAULT_MIN_CRON_SCHEDULE_MINUTES
        self.tool_config.save()
        invalidate_tool_settings_cache()
        self.min_interval_minutes = DEFAULT_MIN_CRON_SCHEDULE_MINUTES
        self.min_interval_seconds = self.min_interval_minutes * 60
        self.addCleanup(self._restore_tool_config)

    def _restore_tool_config(self):
        cfg, _ = ToolConfig.objects.get_or_create(
            plan_name=PlanNamesChoices.FREE,
            defaults={"min_cron_schedule_minutes": DEFAULT_MIN_CRON_SCHEDULE_MINUTES},
        )
        cfg.min_cron_schedule_minutes = self.original_min_interval
        cfg.save()
        invalidate_tool_settings_cache()

    def test_valid_schedules_accepted(self):
        """Test that schedules meeting minimum interval are accepted."""
        valid_schedules = [
            "@daily",          # Once per day
            f"@every {self.min_interval_minutes}m",  # Exactly the configured minimum
            f"@every {max(self.min_interval_minutes, 60)}m",  # At least an hour
            "@every 2h",       # 2 hours
            "0 */2 * * *",     # Every 2 hours (cron)
            "0 0 * * *",       # Daily at midnight (cron)
            "0 8,20 * * *",    # Twice daily, 12 hours apart (cron)
        ]
        if self.min_interval_minutes <= 60:
            valid_schedules.append("@hourly")         # Once per hour
        
        for schedule in valid_schedules:
            with self.subTest(schedule=schedule):
                original_schedule = self.agent.schedule
                result = _execute_update_schedule(self.agent, {"new_schedule": schedule})
                
                self.assertEqual(result["status"], "ok")
                self.agent.refresh_from_db()
                self.assertEqual(self.agent.schedule, schedule)
                
                # Reset for next test
                self.agent.schedule = original_schedule
                self.agent.save()

    def test_too_frequent_interval_schedules_rejected(self):
        """Test that interval schedules more frequent than the configured minimum are rejected."""
        below_limit = max(self.min_interval_minutes - 1, 1)
        actually_too_frequent = [
            f"@every {below_limit}m",  # Just under the configured limit
            "@every 15m",
            "@every 5m",
            "@every 1m",
            "@every 30s",
        ]

        for schedule in actually_too_frequent:
            with self.subTest(schedule=schedule):
                original_schedule = self.agent.schedule
                result = _execute_update_schedule(self.agent, {"new_schedule": schedule})
                
                self.assertEqual(result["status"], "error")
                self.assertIn("too frequent", result["message"])
                self.assertIn(str(self.min_interval_minutes), result["message"])
                
                # Verify schedule wasn't changed
                self.agent.refresh_from_db()
                self.assertEqual(self.agent.schedule, original_schedule)

    def test_plan_specific_min_interval_applied(self):
        """Plan-specific minimum cron frequency is enforced."""
        stricter_min = self.min_interval_minutes + 15
        cfg = ToolConfig.objects.get(plan_name=PlanNamesChoices.FREE)
        cfg.min_cron_schedule_minutes = stricter_min
        cfg.save()
        invalidate_tool_settings_cache()

        result = _execute_update_schedule(self.agent, {"new_schedule": f"@every {self.min_interval_minutes}m"})

        self.assertEqual(result["status"], "error")
        self.assertIn(str(stricter_min), result["message"])
        self.agent.refresh_from_db()
        self.assertEqual(self.agent.schedule, "@daily")

    def test_too_frequent_cron_schedules_rejected(self):
        """Test that cron schedules running more than twice per hour are rejected."""
        too_frequent_cron_schedules = [
            "*/10 * * * *",    # Every 10 minutes (6 times per hour)
            "*/15 * * * *",    # Every 15 minutes (4 times per hour)
            "*/20 * * * *",    # Every 20 minutes (3 times per hour)
            "0,20,40 * * * *", # At 0, 20, 40 minutes (3 times per hour)
            "0,15,30,45 * * * *", # Every 15 minutes (4 times per hour)
        ]
        
        for schedule in too_frequent_cron_schedules:
            with self.subTest(schedule=schedule):
                original_schedule = self.agent.schedule
                result = _execute_update_schedule(self.agent, {"new_schedule": schedule})
                
                self.assertEqual(result["status"], "error")
                self.assertIn("too frequent", result["message"])
                
                # Verify schedule wasn't changed
                self.agent.refresh_from_db()
                self.assertEqual(self.agent.schedule, original_schedule)

    def test_edge_case_cron_schedules(self):
        """Test edge cases for cron schedule validation."""
        # Test exactly 2 executions per hour with various intervals
        edge_cases = [
            ("0,30 * * * *", True),    # Every 30 minutes (2 times per hour) - should be accepted
            ("0,31 * * * *", False),   # At 0 and 31 minutes (29 minute gap from 31 to 0) - should be rejected
            ("15,45 * * * *", True),   # At 15 and 45 minutes (30 minute gap) - should be accepted
            ("10,35 * * * *", False),  # At 10 and 35 minutes (25 minute gap) - should be rejected
            ("5,40 * * * *", False),   # At 5 and 40 minutes (35 minute gap first, then 25 minute gap) - should be rejected
        ]
        
        for schedule, should_be_valid in edge_cases:
            with self.subTest(schedule=schedule, should_be_valid=should_be_valid):
                original_schedule = self.agent.schedule
                result = _execute_update_schedule(self.agent, {"new_schedule": schedule})
                
                if should_be_valid:
                    self.assertEqual(result["status"], "ok")
                    self.agent.refresh_from_db()
                    self.assertEqual(self.agent.schedule, schedule)
                    # Reset for next test
                    self.agent.schedule = original_schedule
                    self.agent.save()
                else:
                    self.assertEqual(result["status"], "error")
                    self.assertIn("too frequent", result["message"])
                    # Verify schedule wasn't changed
                    self.agent.refresh_from_db()
                    self.assertEqual(self.agent.schedule, original_schedule)

    def test_empty_and_null_schedules(self):
        """Test that empty and null schedules are accepted (disables scheduling)."""
        empty_schedules = [
            None,
            "",
            "   ",  # Whitespace only
        ]
        
        for schedule in empty_schedules:
            with self.subTest(schedule=schedule):
                result = _execute_update_schedule(self.agent, {"new_schedule": schedule})
                
                self.assertEqual(result["status"], "ok")
                self.assertEqual(result["message"], "Schedule has been disabled.")
                
                self.agent.refresh_from_db()
                self.assertIsNone(self.agent.schedule)
                
                # Reset for next test
                self.agent.schedule = "@daily"
                self.agent.save()

    def test_invalid_schedule_format_rejected(self):
        """Test that invalid schedule formats are rejected without affecting the agent."""
        invalid_schedules = [
            "invalid schedule",
            "@reboot",
            "@every 5x",
            "60 * * * *",      # Invalid minute value
            "* 25 * * *",      # Invalid hour value
        ]
        
        for schedule in invalid_schedules:
            with self.subTest(schedule=schedule):
                original_schedule = self.agent.schedule
                result = _execute_update_schedule(self.agent, {"new_schedule": schedule})
                
                self.assertEqual(result["status"], "error")
                self.assertIn("Invalid schedule format", result["message"])
                
                # Verify schedule wasn't changed
                self.agent.refresh_from_db()
                self.assertEqual(self.agent.schedule, original_schedule)

    def test_boundary_minute_interval(self):
        """Test that exactly the configured minimum interval is accepted."""
        result = _execute_update_schedule(self.agent, {"new_schedule": f"@every {self.min_interval_minutes}m"})
        
        self.assertEqual(result["status"], "ok")
        self.agent.refresh_from_db()
        self.assertEqual(self.agent.schedule, f"@every {self.min_interval_minutes}m")

    def test_just_under_minute_interval(self):
        """Test that intervals just under the minimum are rejected."""
        almost_min = max(self.min_interval_minutes - 1, 1)
        result = _execute_update_schedule(self.agent, {"new_schedule": f"@every {almost_min}m"})
        
        self.assertEqual(result["status"], "error")
        self.assertIn("too frequent", result["message"])
        
        # Verify schedule wasn't changed
        original_schedule = self.agent.schedule
        self.agent.refresh_from_db()
        self.assertEqual(self.agent.schedule, original_schedule)

    def test_complex_interval_combinations(self):
        """Test complex interval combinations that should be valid."""
        valid_complex_intervals = [
            "@every 1h 30m",   # 90 minutes
            "@every 2h 15m",   # 135 minutes  
            f"@every {self.min_interval_minutes}m 30s",  # Slightly above the minimum
        ]
        
        for schedule in valid_complex_intervals:
            with self.subTest(schedule=schedule):
                original_schedule = self.agent.schedule
                result = _execute_update_schedule(self.agent, {"new_schedule": schedule})
                
                self.assertEqual(result["status"], "ok")
                self.agent.refresh_from_db()
                self.assertEqual(self.agent.schedule, schedule)
                
                # Reset for next test
                self.agent.schedule = original_schedule
                self.agent.save()


@tag("batch_event_processing")
class HttpRequestSecretPlaceholderTests(TestCase):
    """Unit tests for http_request tool secret placeholder substitution."""

    def setUp(self):
        self.user = User.objects.create_user(
            username="http_tester@example.com", 
            email="http_tester@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="HttpBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="HttpAgent",
            charter="HTTP test agent",
            browser_use_agent=self.browser_agent,
        )

    def _create_secret(
        self,
        key,
        value,
        domain="https://api.example.com",
        name=None,
        secret_type=PersistentAgentSecret.SecretType.CREDENTIAL,
    ):
        """Helper to create a secret for the agent."""
        if secret_type == PersistentAgentSecret.SecretType.ENV_VAR:
            domain = PersistentAgentSecret.ENV_VAR_DOMAIN_SENTINEL
            key = str(key).upper()
        secret = PersistentAgentSecret(
            agent=self.agent,
            domain_pattern=domain,
            name=name or key,
            key=key,
            secret_type=secret_type,
        )
        secret.set_value(value)
        secret.save()
        return secret

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_secret_substitution_in_headers(self, mock_proxy, mock_request):
        """Test that secret placeholders in headers are properly substituted."""
        # Create a test secret
        self._create_secret("api_key", "secret-api-key-value")
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'application/json'},
            'iter_content': lambda self, chunk_size: [b'{"success": true}'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with placeholder in headers
        params = {
            "method": "GET",
            "url": "https://api.example.com/data",
            "headers": {
                "Authorization": "Bearer <<<api_key>>>",
                "X-API-Key": "<<<api_key>>>"
            }
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify request was made
        self.assertEqual(result["status"], "ok", result)
        mock_request.assert_called_once()
        
        # Verify headers were substituted correctly
        call_args = mock_request.call_args
        headers = call_args[1]["headers"]
        self.assertEqual(headers["Authorization"], "Bearer secret-api-key-value")
        self.assertEqual(headers["X-API-Key"], "secret-api-key-value")

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_secret_substitution_in_url(self, mock_proxy, mock_request):
        """Test that secret placeholders in URL are properly substituted."""
        # Create test secrets
        self._create_secret("base_url", "https://api.secret.com")
        self._create_secret("api_version", "v2")
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'success'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with placeholder in URL
        params = {
            "method": "GET",
            "url": "<<<base_url>>>/<<<api_version>>>/endpoint"
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify request was made with substituted URL
        self.assertEqual(result["status"], "ok", result)
        mock_request.assert_called_once()
        
        call_args = mock_request.call_args
        self.assertEqual(call_args[0][1], "https://api.secret.com/v2/endpoint")

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_http_request_uses_agent_preferred_proxy(self, mock_proxy_selector, mock_request):
        """Preferred proxies assigned to the browser agent should be reused for HTTP requests."""
        proxy = ProxyServer.objects.create(
            name="Dedicated",
            proxy_type=ProxyServer.ProxyType.HTTP,
            host="dedicated.proxy",
            port=8080,
            username="user",
            password="pass",
            is_active=True,
        )
        self.browser_agent.preferred_proxy = proxy
        self.browser_agent.save(update_fields=["preferred_proxy"])

        def _selector(agent, *args, **kwargs):
            self.assertEqual(agent.preferred_proxy, proxy)
            return proxy

        mock_proxy_selector.side_effect = _selector
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'ok'],
            'close': lambda self: None,
        })()
        mock_request.return_value = mock_response

        params = {
            "method": "GET",
            "url": "https://api.example.com/health",
        }

        result = _execute_http_request(self.agent, params)

        self.assertEqual(result["status"], "ok", result)
        mock_proxy_selector.assert_called_once()
        proxies = mock_request.call_args[1]["proxies"]
        self.assertEqual(proxies, {"http": proxy.proxy_url, "https": proxy.proxy_url})

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_secret_substitution_in_body_string(self, mock_proxy, mock_request):
        """Test that secret placeholders in body string are properly substituted."""
        # Create test secrets
        self._create_secret("username", "test_user")
        self._create_secret("password", "secret_pass")
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'application/json'},
            'iter_content': lambda self, chunk_size: [b'{"login": "success"}'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with placeholders in body
        params = {
            "method": "POST",
            "url": "https://api.example.com/login",
            "body": '{"username": "<<<username>>>", "password": "<<<password>>>"}'
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify request was made with substituted body
        self.assertEqual(result["status"], "ok")
        mock_request.assert_called_once()
        
        call_args = mock_request.call_args
        expected_body = '{"username": "test_user", "password": "secret_pass"}'
        self.assertEqual(call_args[1]["data"], expected_body)

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_secret_substitution_in_body_dict(self, mock_proxy, mock_request):
        """Test that secret placeholders in body dict are properly substituted and JSON-encoded."""
        # Create test secret
        self._create_secret("client_secret", "super-secret-value")
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'application/json'},
            'iter_content': lambda self, chunk_size: [b'{"token": "abc123"}'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with dict body containing placeholders
        params = {
            "method": "POST",
            "url": "https://oauth.example.com/token",
            "body": {
                "grant_type": "client_credentials",
                "client_secret": "<<<client_secret>>>",
                "scope": "read"
            }
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify request was made
        self.assertEqual(result["status"], "ok")
        mock_request.assert_called_once()
        
        # Verify body was substituted and JSON-encoded
        call_args = mock_request.call_args
        import json
        body_data = json.loads(call_args[1]["data"])
        self.assertEqual(body_data["client_secret"], "super-secret-value")
        self.assertEqual(body_data["grant_type"], "client_credentials")
        self.assertEqual(body_data["scope"], "read")

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_secret_substitution_with_whitespace(self, mock_proxy, mock_request):
        """Test that secret placeholders with whitespace are properly handled."""
        # Create test secret
        self._create_secret("api_token", "token-with-spaces")
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'ok'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with whitespace in placeholders
        params = {
            "method": "GET",
            "url": "https://api.example.com/data",
            "headers": {
                "Token": "<<<  api_token  >>>",  # Extra whitespace
                "Authorization": "Bearer <<<api_token>>>"  # No whitespace
            }
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify both placeholders were substituted correctly
        self.assertEqual(result["status"], "ok")
        call_args = mock_request.call_args
        headers = call_args[1]["headers"]
        self.assertEqual(headers["Token"], "token-with-spaces")
        self.assertEqual(headers["Authorization"], "Bearer token-with-spaces")

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_nonexistent_secret_placeholder_unchanged(self, mock_proxy, mock_request):
        """Test that placeholders for nonexistent secrets are left unchanged."""
        # Create one secret but reference a different one
        self._create_secret("real_secret", "real_value")
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'ok'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with nonexistent secret placeholder
        params = {
            "method": "GET",
            "url": "https://api.example.com/data",
            "headers": {
                "Real-Key": "<<<real_secret>>>",  # This should be replaced
                "Fake-Key": "<<<fake_secret>>>"   # This should remain unchanged
            }
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify only real secret was substituted
        self.assertEqual(result["status"], "ok")
        call_args = mock_request.call_args
        headers = call_args[1]["headers"]
        self.assertEqual(headers["Real-Key"], "real_value")
        self.assertEqual(headers["Fake-Key"], "<<<fake_secret>>>")  # Unchanged

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_env_var_secret_is_ignored_for_placeholder_substitution(self, mock_proxy, mock_request):
        self._create_secret(
            "API_TOKEN",
            "env-var-token-value",
            secret_type=PersistentAgentSecret.SecretType.ENV_VAR,
        )

        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'ok'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response

        params = {
            "method": "GET",
            "url": "https://api.example.com/data",
            "headers": {
                "Authorization": "Bearer <<<API_TOKEN>>>",
            },
        }

        result = _execute_http_request(self.agent, params)

        self.assertEqual(result["status"], "ok")
        headers = mock_request.call_args[1]["headers"]
        self.assertEqual(headers["Authorization"], "Bearer <<<API_TOKEN>>>")

    @patch('api.agent.tools.http_request.build_signed_filespace_download_url')
    @patch('api.agent.tools.http_request.write_bytes_to_dir')
    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_http_request_download_saves_to_filespace(
        self,
        mock_proxy,
        mock_request,
        mock_write,
        mock_signed_url,
    ):
        """Download requests should persist the response body to filespace."""
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {
                'Content-Type': 'application/pdf',
                'Content-Disposition': 'attachment; filename="report.pdf"',
            },
            'iter_content': lambda self, chunk_size: [b'file-bytes'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        mock_write.return_value = {
            "status": "ok",
            "path": "/downloads/report.pdf",
            "node_id": "node-123",
            "filename": "report.pdf",
        }
        mock_signed_url.return_value = "https://example.com/downloads/report.pdf"

        params = {
            "method": "GET",
            "url": "https://files.example.com/report.pdf",
            "download": True,
        }

        result = _execute_http_request(self.agent, params)

        self.assertEqual(result["status"], "ok", result)
        self.assertEqual(result["file"], "$[/downloads/report.pdf]")
        self.assertEqual(result["node_id"], "node-123")
        self.assertEqual(result["filename"], "report.pdf")
        mock_write.assert_called_once()
        self.assertEqual(mock_write.call_args.kwargs["content_bytes"], b"file-bytes")
        self.assertEqual(mock_write.call_args.kwargs["path"], "/downloads/report.pdf")
        self.assertEqual(mock_write.call_args.kwargs["extension"], ".pdf")
        self.assertEqual(mock_write.call_args.kwargs["mime_type"], "application/pdf")

    @override_settings(ALLOW_FILE_DOWNLOAD=False)
    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_http_request_download_disabled_returns_error(self, mock_proxy, mock_request):
        params = {
            "method": "GET",
            "url": "https://files.example.com/report.pdf",
            "download": True,
        }

        result = _execute_http_request(self.agent, params)

        self.assertEqual(result["status"], "error", result)
        self.assertEqual(result["message"], "File downloads are disabled.")
        mock_proxy.assert_not_called()
        mock_request.assert_not_called()

    @patch('api.agent.tools.http_request.write_bytes_to_dir')
    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_http_request_download_non_success_returns_error(self, mock_proxy, mock_request, mock_write):
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 404,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'not found'],
            'close': lambda self: None,
        })()
        mock_request.return_value = mock_response

        params = {
            "method": "GET",
            "url": "https://files.example.com/missing.pdf",
            "download": True,
        }

        result = _execute_http_request(self.agent, params)

        self.assertEqual(result["status"], "error", result)
        self.assertEqual(result["status_code"], 404)
        mock_request.assert_called_once()
        mock_write.assert_not_called()

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_no_secrets_no_substitution(self, mock_proxy, mock_request):
        """Test that when agent has no secrets, placeholders remain unchanged."""
        # Don't create any secrets
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'ok'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with placeholder when no secrets exist
        params = {
            "method": "GET",
            "url": "https://api.example.com/data",
            "headers": {
                "Authorization": "Bearer <<<api_key>>>"
            }
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify placeholder remains unchanged
        self.assertEqual(result["status"], "ok")
        call_args = mock_request.call_args
        headers = call_args[1]["headers"]
        self.assertEqual(headers["Authorization"], "Bearer <<<api_key>>>")

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_complex_nested_substitution(self, mock_proxy, mock_request):
        """Test secret substitution in complex nested data structures."""
        # Create test secrets
        self._create_secret("auth_token", "nested-auth-token")
        self._create_secret("api_endpoint", "https://nested-api.com")
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'application/json'},
            'iter_content': lambda self, chunk_size: [b'{"result": "success"}'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with complex nested body structure
        params = {
            "method": "POST",
            "url": "<<<api_endpoint>>>/webhook",
            "headers": {
                "Authorization": "Bearer <<<auth_token>>>",
                "Content-Type": "application/json"
            },
            "body": {
                "webhook": {
                    "url": "<<<api_endpoint>>>/callback",
                    "auth": {
                        "type": "bearer",
                        "token": "<<<auth_token>>>"
                    },
                    "events": ["user.created", "user.updated"]
                }
            }
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify all substitutions worked correctly
        self.assertEqual(result["status"], "ok")
        call_args = mock_request.call_args
        
        # Check URL substitution
        self.assertEqual(call_args[0][1], "https://nested-api.com/webhook")
        
        # Check header substitution
        headers = call_args[1]["headers"]
        self.assertEqual(headers["Authorization"], "Bearer nested-auth-token")
        
        # Check body substitution (should be JSON-encoded)
        import json
        body_data = json.loads(call_args[1]["data"])
        self.assertEqual(body_data["webhook"]["url"], "https://nested-api.com/callback")
        self.assertEqual(body_data["webhook"]["auth"]["token"], "nested-auth-token")
        self.assertEqual(body_data["webhook"]["events"], ["user.created", "user.updated"])

    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_whole_string_secret_key_substitution(self, mock_proxy, mock_request):
        """Test that a header value that exactly matches a secret key is substituted."""
        # Create test secret
        self._create_secret("bearer_token", "whole-string-token-value")
        
        # Mock proxy and response
        mock_proxy.return_value = type('ProxyServer', (), {'proxy_url': 'http://proxy:8080'})()
        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'ok'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response
        
        # Test with header value that exactly matches secret key
        params = {
            "method": "GET",
            "url": "https://api.example.com/data",
            "headers": {
                "Authorization": "bearer_token",  # Exact match to secret key
                "X-Token": "Bearer <<<bearer_token>>>"  # Regular placeholder
            }
        }
        
        result = _execute_http_request(self.agent, params)
        
        # Verify both substitutions worked
        self.assertEqual(result["status"], "ok")
        call_args = mock_request.call_args
        headers = call_args[1]["headers"]
        self.assertEqual(headers["Authorization"], "whole-string-token-value")  # Whole string match
        self.assertEqual(headers["X-Token"], "Bearer whole-string-token-value")  # Placeholder match 

    @override_settings(GOBII_PROPRIETARY_MODE=False)
    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_allows_direct_http_request_without_proxy_in_community_mode(self, mock_proxy, mock_request):
        """Community mode should fall back to direct requests when no proxy exists."""
        mock_proxy.side_effect = RuntimeError("No proxies configured")

        mock_response = type('Response', (), {
            'status_code': 200,
            'headers': {'Content-Type': 'text/plain'},
            'iter_content': lambda self, chunk_size: [b'ok'],
            'close': lambda self: None
        })()
        mock_request.return_value = mock_response

        params = {
            "method": "GET",
            "url": "https://api.example.com/community",
        }

        result = _execute_http_request(self.agent, params)

        self.assertEqual(result["status"], "ok")

        call_args = mock_request.call_args
        self.assertNotIn("proxies", call_args[1])

    @override_settings(GOBII_PROPRIETARY_MODE=True)
    @patch('requests.request')
    @patch('api.agent.tools.http_request.select_proxy_for_persistent_agent')
    def test_requires_proxy_in_proprietary_mode(self, mock_proxy, mock_request):
        """Proprietary mode must fail if no proxy is available."""
        mock_proxy.side_effect = RuntimeError("No proxies configured")

        params = {
            "method": "GET",
            "url": "https://api.example.com/proprietary",
        }

        result = _execute_http_request(self.agent, params)

        self.assertEqual(result["status"], "error")
        self.assertIn("No proxy server available", result["message"])
        mock_request.assert_not_called()

@tag("batch_event_processing")
class PromptConfigFunctionTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="prompt_limit_user@example.com",
            email="prompt_limit_user@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="PromptConfigBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="PromptConfigAgent",
            charter="",
            browser_use_agent=self.browser_agent,
        )

    def _configure_limits(self):
        config, _ = PromptConfig.objects.get_or_create(singleton_id=1)
        config.standard_prompt_token_budget = 500
        config.premium_prompt_token_budget = 1000
        config.max_prompt_token_budget = 1500
        config.standard_message_history_limit = 3
        config.premium_message_history_limit = 7
        config.max_message_history_limit = 9
        config.standard_tool_call_history_limit = 4
        config.premium_tool_call_history_limit = 8
        config.max_tool_call_history_limit = 10
        config.standard_skill_prompt_limit = 2
        config.premium_skill_prompt_limit = 4
        config.max_skill_prompt_limit = 6
        config.save()
        invalidate_prompt_settings_cache()
        return config

    def test_limits_follow_configuration(self):
        config = self._configure_limits()

        with patch("api.agent.core.prompt_context.get_agent_llm_tier", return_value=AgentLLMTier.STANDARD):
            self.assertEqual(get_prompt_token_budget(self.agent), config.standard_prompt_token_budget)
            self.assertEqual(message_history_limit(self.agent), config.standard_message_history_limit)
            self.assertEqual(tool_call_history_limit(self.agent), config.standard_tool_call_history_limit)
            self.assertEqual(skill_prompt_limit(self.agent), config.standard_skill_prompt_limit)

        with patch("api.agent.core.prompt_context.get_agent_llm_tier", return_value=AgentLLMTier.PREMIUM):
            self.assertEqual(get_prompt_token_budget(self.agent), config.premium_prompt_token_budget)
            self.assertEqual(message_history_limit(self.agent), config.premium_message_history_limit)
            self.assertEqual(tool_call_history_limit(self.agent), config.premium_tool_call_history_limit)
            self.assertEqual(skill_prompt_limit(self.agent), config.premium_skill_prompt_limit)

        with patch("api.agent.core.prompt_context.get_agent_llm_tier", return_value=AgentLLMTier.MAX):
            self.assertEqual(get_prompt_token_budget(self.agent), config.max_prompt_token_budget)
            self.assertEqual(message_history_limit(self.agent), config.max_message_history_limit)
            self.assertEqual(tool_call_history_limit(self.agent), config.max_tool_call_history_limit)
            self.assertEqual(skill_prompt_limit(self.agent), config.max_skill_prompt_limit)

    def test_skill_prompt_limit_defaults_to_three(self):
        invalidate_prompt_settings_cache()
        settings = get_prompt_settings()

        self.assertEqual(settings.standard_skill_prompt_limit, 3)
        self.assertEqual(settings.premium_skill_prompt_limit, 3)
        self.assertEqual(settings.max_skill_prompt_limit, 3)
        self.assertEqual(settings.ultra_skill_prompt_limit, 3)
        self.assertEqual(settings.ultra_max_skill_prompt_limit, 3)

    def test_internal_reasoning_history_limit_setting(self):
        invalidate_prompt_settings_cache()
        self.assertEqual(get_prompt_settings().internal_reasoning_history_limit, 3)

        config, _ = PromptConfig.objects.get_or_create(singleton_id=1)
        config.internal_reasoning_history_limit = 7
        config.save()
        invalidate_prompt_settings_cache()

        self.assertEqual(get_prompt_settings().internal_reasoning_history_limit, 7)

    def test_tool_search_auto_enable_apps_setting(self):
        config, _ = ToolConfig.objects.get_or_create(plan_name=PlanNamesChoices.FREE)
        config.tool_search_auto_enable_apps = False
        config.save()
        invalidate_tool_settings_cache()

        self.assertFalse(get_tool_settings_for_plan(PlanNamesChoices.FREE).tool_search_auto_enable_apps)

    def test_prompt_config_admin_omits_browser_task_unified_history_limit(self):
        admin_view = PromptConfigAdmin(PromptConfig, django_admin.site)

        self.assertIn("internal_reasoning_history_limit", admin_view.list_display)
        self.assertIn("standard_skill_prompt_limit", admin_view.list_display)
        self.assertNotIn("browser_task_unified_history_limit", admin_view.list_display)

        unified_fields = next(
            fields["fields"]
            for title, fields in admin_view.fieldsets
            if title == "Unified history limits"
        )
        self.assertIn("internal_reasoning_history_limit", unified_fields)
        self.assertNotIn("browser_task_unified_history_limit", unified_fields)

        skill_fields = next(
            fields["fields"]
            for title, fields in admin_view.fieldsets
            if title == "Skill prompt limits"
        )
        self.assertIn("standard_skill_prompt_limit", skill_fields)
        self.assertIn("ultra_max_skill_prompt_limit", skill_fields)

    def test_tool_config_admin_exposes_tool_search_auto_enable_apps(self):
        admin_view = ToolConfigAdmin(ToolConfig, django_admin.site)

        self.assertIn("tool_search_auto_enable_apps", admin_view.list_display)

        tool_search_fields = next(
            fields["fields"]
            for title, fields in admin_view.fieldsets
            if title == "Tool search"
        )
        self.assertIn("tool_search_auto_enable_apps", tool_search_fields)


@tag("batch_event_processing")
class EventProcessingRuntimeGuardTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="runtime_guard@example.com",
            email="runtime_guard@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="RuntimeBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="RuntimeAgent",
            charter="Test runtime guard",
            browser_use_agent=self.browser_agent,
        )
        self.endpoint = PersistentAgentCommsEndpoint.objects.create(
            owner_agent=self.agent,
            channel="email",
            address="runtime-agent@example.com",
            is_primary=True,
        )
        self.external_endpoint = PersistentAgentCommsEndpoint.objects.create(
            channel="email",
            address="runtime-user@example.com",
        )

    def _create_inbound_email_message(self, *, body: str, timestamp) -> PersistentAgentMessage:
        message = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.external_endpoint,
            to_endpoint=self.endpoint,
            is_outbound=False,
            body=body,
            raw_payload={},
            seq=f"RG{int(timestamp.timestamp() * 1_000_000):024d}"[:26],
        )
        PersistentAgentMessage.objects.filter(pk=message.pk).update(timestamp=timestamp)
        message.refresh_from_db()
        return message

    @override_settings(GOBII_PROPRIETARY_MODE=True)
    def test_tool_call_runtime_rejects_tier_blacklisted_static_tool(self):
        tier = get_intelligence_tier("standard")
        tier.blacklisted_tools = ["search_tools"]
        tier.save(update_fields=["blacklisted_tools"])
        self.agent.preferred_llm_tier = tier
        self.agent.save(update_fields=["preferred_llm_tier"])

        result, updated_tools = _execute_tool_call_runtime(
            self.agent,
            tool_name="search_tools",
            exec_params={"query": "anything"},
            budget_ctx=None,
            eval_run_id=None,
        )

        self.assertIsNone(updated_tools)
        self.assertEqual(result["status"], "error")
        self.assertIn("intelligence tier", result["message"])

    @override_settings(CELERY_TASK_ALWAYS_EAGER=False)
    @patch("api.agent.tasks.process_events.process_agent_events_task.apply_async")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing._runtime_exceeded", return_value=True)
    @patch("api.agent.core.event_processing.build_prompt_context")
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[])
    @patch("api.agent.core.event_processing.get_redis_client")
    def test_run_agent_loop_aborts_when_runtime_limit_hit(
        self,
        mock_get_redis,
        _mock_tools,
        mock_build_context,
        _mock_runtime,
        mock_get_pending_settings,
        mock_apply_async,
    ):
        class _FakeRedis:
            def get(self, _key):
                return None

            def exists(self, _key):
                return 0

        mock_get_redis.return_value = _FakeRedis()
        mock_get_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )

        with patch(
            "api.agent.core.event_processing.settings.AGENT_EVENT_PROCESSING_MAX_RUNTIME_SECONDS",
            1,
        ):
            usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertFalse(mock_build_context.called)
        self.assertTrue(
            self.agent.steps.filter(description__icontains="runtime limit").exists()
        )
        mock_apply_async.assert_called_once_with(
            args=[str(self.agent.id)],
            countdown=10,
        )
        self.assertEqual(usage.get("total_tokens"), 0)

    @patch("api.agent.core.event_processing._execute_prepared_tool_call")
    def test_execute_prepared_tool_batch_stops_after_signup_preview_reply(
        self,
        mock_execute_prepared_tool_call,
    ):
        first_call = _PreparedToolExecution(
            idx=0,
            tool_name="send_chat_message",
            tool_params={"body": "Preview reply"},
            exec_params={"body": "Preview reply"},
            pending_step=None,
            credits_consumed=None,
            consumed_credit=None,
            call_id="call-1",
            explicit_continue=None,
            inferred_continue=False,
            parallel_safe=False,
            parallel_ineligible_reason="unsafe_tool:send_chat_message",
        )
        second_call = _PreparedToolExecution(
            idx=1,
            tool_name="sqlite_batch",
            tool_params={"sql": "select 1"},
            exec_params={"sql": "select 1"},
            pending_step=None,
            credits_consumed=None,
            consumed_credit=None,
            call_id="call-2",
            explicit_continue=None,
            inferred_continue=False,
            parallel_safe=False,
            parallel_ineligible_reason="unsafe_tool:sqlite_batch",
        )
        prepared_batch = _PreparedToolBatch(
            prepared_calls=[first_call, second_call],
            followup_required=False,
            all_calls_sleep=False,
            abort_after_execution=False,
            parallel_ineligible_reason="unsafe_tool:send_chat_message",
        )

        def _execute_side_effect(agent, prepared, **_kwargs):
            if prepared.tool_name == "send_chat_message":
                PersistentAgent.objects.filter(id=agent.id).update(
                    signup_preview_state=PersistentAgent.SignupPreviewState.AWAITING_SIGNUP_COMPLETION,
                )
            return _ToolExecutionOutcome(
                prepared=prepared,
                result={"status": "ok"},
                duration_ms=1,
                updated_tools=None,
                variable_map={},
            )

        mock_execute_prepared_tool_call.side_effect = _execute_side_effect

        executed_batch = _execute_prepared_tool_batch(
            self.agent,
            prepared_batch,
            budget_ctx=None,
            eval_run_id=None,
            tools=[],
            heartbeat=None,
            lock_extender=None,
        )

        self.assertTrue(executed_batch.abort_after_execution)
        self.assertEqual(mock_execute_prepared_tool_call.call_count, 1)

    def test_finalize_search_tools_always_requires_followup(self):
        prepared = _PreparedToolExecution(
            idx=0,
            tool_name="search_tools",
            tool_params={"query": "competitor research"},
            exec_params={"query": "competitor research"},
            pending_step=None,
            credits_consumed=None,
            consumed_credit=None,
            call_id="call-search",
            explicit_continue=False,
            inferred_continue=False,
            parallel_safe=False,
            parallel_ineligible_reason="unsafe_tool:search_tools",
        )
        outcome = _ToolExecutionOutcome(
            prepared=prepared,
            result={"status": "success", "message": "Enabled tools"},
            duration_ms=1,
            updated_tools=None,
            variable_map={},
        )

        finalized = _finalize_tool_batch(
            self.agent,
            [outcome],
            attach_completion=lambda kwargs: None,
            attach_prompt_archive=lambda step: None,
        )

        self.assertTrue(finalized.followup_required)

    @tag("batch_event_processing")
    def test_finalize_explicit_stop_without_auto_sleep_requires_followup(self):
        prepared = _PreparedToolExecution(
            idx=0,
            tool_name="http_request",
            tool_params={
                "method": "GET",
                "url": "https://briefing.example.test/api/daily.json",
                "will_continue_work": False,
            },
            exec_params={},
            pending_step=None,
            credits_consumed=None,
            consumed_credit=None,
            call_id="call-fetch",
            explicit_continue=False,
            inferred_continue=False,
            parallel_safe=False,
            parallel_ineligible_reason="unsafe_tool:http_request",
        )
        outcome = _ToolExecutionOutcome(
            prepared=prepared,
            result={"status": "ok", "content": {"items": []}},
            duration_ms=1,
            updated_tools=None,
            variable_map={},
        )

        finalized = _finalize_tool_batch(
            self.agent,
            [outcome],
            attach_completion=lambda kwargs: None,
            attach_prompt_archive=lambda step: None,
        )

        self.assertTrue(finalized.followup_required)
        self.assertIs(finalized.last_explicit_continue, False)

    def test_non_message_stop_continues_when_latest_inbound_is_unanswered(self):
        self._create_inbound_email_message(
            body="what's the weather in frederick md?",
            timestamp=timezone.now(),
        )
        finalized = _FinalizedToolBatch(
            executed_calls=1,
            followup_required=False,
            message_delivery_ok=False,
            last_explicit_continue=False,
            inferred_message_continue_this_iteration=False,
            executed_non_message_action=True,
        )

        self.assertTrue(_latest_inbound_message_needs_reply(self.agent))
        self.assertTrue(_should_continue_for_unanswered_inbound_after_tools(self.agent, finalized))

    def test_non_message_stop_does_not_continue_after_user_facing_reply(self):
        now = timezone.now()
        self._create_inbound_email_message(body="thanks", timestamp=now)
        outbound = PersistentAgentMessage.objects.create(
            owner_agent=self.agent,
            from_endpoint=self.endpoint,
            to_endpoint=self.external_endpoint,
            is_outbound=True,
            body="You're welcome.",
            raw_payload={},
            seq=f"OUT{int(now.timestamp() * 1_000_000):023d}"[:26],
        )
        PersistentAgentMessage.objects.filter(pk=outbound.pk).update(timestamp=now + timedelta(seconds=1))
        finalized = _FinalizedToolBatch(
            executed_calls=1,
            followup_required=False,
            message_delivery_ok=False,
            last_explicit_continue=False,
            inferred_message_continue_this_iteration=False,
            executed_non_message_action=True,
        )

        self.assertFalse(_latest_inbound_message_needs_reply(self.agent))
        self.assertFalse(_should_continue_for_unanswered_inbound_after_tools(self.agent, finalized))

    def test_pending_progress_reply_continues_after_non_message_stop(self):
        finalized = _FinalizedToolBatch(
            executed_calls=1,
            followup_required=False,
            message_delivery_ok=False,
            last_explicit_continue=False,
            inferred_message_continue_this_iteration=False,
            executed_non_message_action=True,
        )

        self.assertTrue(_should_continue_for_pending_progress_reply(True, finalized))

    def test_pending_progress_reply_stops_for_human_input_request(self):
        finalized = _FinalizedToolBatch(
            executed_calls=1,
            followup_required=False,
            message_delivery_ok=False,
            last_explicit_continue=False,
            inferred_message_continue_this_iteration=False,
            executed_non_message_action=True,
            human_input_request_ok=True,
        )

        self.assertFalse(_should_continue_for_pending_progress_reply(True, finalized))

    def test_signup_preview_processing_pause_is_suppressed_during_planning(self):
        self.agent.signup_preview_state = PersistentAgent.SignupPreviewState.AWAITING_SIGNUP_COMPLETION
        self.agent.planning_state = PersistentAgent.PlanningState.PLANNING
        self.agent.save(update_fields=["signup_preview_state", "planning_state", "updated_at"])

        self.assertFalse(is_signup_preview_processing_paused(self.agent))

        self.agent.planning_state = PersistentAgent.PlanningState.COMPLETED
        self.agent.save(update_fields=["planning_state", "updated_at"])

        self.assertTrue(is_signup_preview_processing_paused(self.agent))


@tag("batch_event_processing")
class HumanInboundGenerationTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="human_generation@example.com",
            email="human_generation@example.com",
            password="secret",
        )
        EmailAddress.objects.create(user=self.user, email=self.user.email, verified=True, primary=True)
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="HumanGenerationBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="HumanGenerationAgent",
            charter="Track interrupting human input",
            browser_use_agent=self.browser_agent,
        )
        self.web_agent_address = build_web_agent_address(self.agent.id)
        self.web_user_address = build_web_user_address(self.user.id, self.agent.id)

    def _owned_endpoint(self, channel: str, address: str) -> PersistentAgentCommsEndpoint:
        return PersistentAgentCommsEndpoint.objects.create(
            owner_agent=self.agent,
            channel=channel,
            address=address,
            is_primary=True,
        )

    def _ingest(self, channel: str, sender: str, recipient: str, body: str = "hello") -> None:
        parsed = ParsedMessage(
            sender=sender,
            recipient=recipient,
            subject=None,
            body=body,
            attachments=[],
            raw_payload={"source": "unit_test"},
            msg_channel=channel,
        )
        ingest_inbound_message(channel, parsed)

    def test_web_message_bumps_generation_and_queues_with_generation(self):
        self._owned_endpoint(CommsChannel.WEB, self.web_agent_address)
        before = get_human_inbound_generation(self.agent.id)
        expected = before + 1

        with patch("api.agent.tasks.process_agent_events_task.delay") as mock_delay:
            with self.captureOnCommitCallbacks(execute=True):
                self._ingest(CommsChannel.WEB, self.web_user_address, self.web_agent_address)

        self.assertEqual(get_human_inbound_generation(self.agent.id), expected)
        mock_delay.assert_called_once_with(
            str(self.agent.id),
            inbound_generation=expected,
        )

    def test_email_and_sms_messages_queue_without_bumping_generation(self):
        cases = [
            (CommsChannel.EMAIL, self.user.email, "human-generation-agent@example.com"),
            (CommsChannel.SMS, "+15555550100", "+15555550101"),
        ]

        for channel, sender, recipient in cases:
            with self.subTest(channel=channel):
                self._owned_endpoint(channel, recipient)
                before = get_human_inbound_generation(self.agent.id)

                with patch("api.agent.tasks.process_agent_events_task.delay") as mock_delay:
                    with self.captureOnCommitCallbacks(execute=True):
                        self._ingest(channel, sender, recipient)

                self.assertEqual(get_human_inbound_generation(self.agent.id), before)
                mock_delay.assert_called_once_with(str(self.agent.id))

    def test_web_human_input_panel_response_bumps_generation_and_queues_with_generation(self):
        conversation = PersistentAgentConversation.objects.create(
            owner_agent=self.agent,
            channel=CommsChannel.WEB,
            address=self.web_user_address,
        )
        request_obj = PersistentAgentHumanInputRequest.objects.create(
            agent=self.agent,
            conversation=conversation,
            question="Which result format should I use?",
            options_json=[
                {"key": "summary", "title": "Summary", "description": "Reply in chat."},
                {"key": "csv", "title": "CSV", "description": "Attach a spreadsheet."},
            ],
            input_mode=PersistentAgentHumanInputRequest.InputMode.OPTIONS_PLUS_TEXT,
            requested_via_channel=CommsChannel.WEB,
        )
        before = get_human_inbound_generation(self.agent.id)
        expected = before + 1

        with patch("api.agent.tasks.process_agent_events_task.delay") as mock_delay:
            with self.captureOnCommitCallbacks(execute=True):
                submit_human_input_responses_batch(
                    self.agent,
                    [{"request_id": str(request_obj.id), "selected_option_key": "summary"}],
                    actor_user_id=self.user.id,
                )

        self.assertEqual(get_human_inbound_generation(self.agent.id), expected)
        mock_delay.assert_called_once_with(
            str(self.agent.id),
            inbound_generation=expected,
        )

    def test_inbound_webhook_does_not_bump_human_generation(self):
        webhook = PersistentAgentInboundWebhook.objects.create(agent=self.agent, name="Deploy Hook")
        before = get_human_inbound_generation(self.agent.id)

        with patch("api.agent.tasks.process_agent_events_task.delay") as mock_delay:
            with self.captureOnCommitCallbacks(execute=True):
                ingest_inbound_webhook_message(
                    webhook,
                    body='{"status":"ok"}',
                    raw_payload={"source": "inbound_webhook", "source_kind": "webhook"},
                )

        self.assertEqual(get_human_inbound_generation(self.agent.id), before)
        mock_delay.assert_called_once_with(str(self.agent.id))

    def test_peer_agent_message_does_not_bump_human_generation(self):
        peer_browser_agent = BrowserUseAgent.objects.create(user=self.user, name="PeerGenerationBA")
        peer_agent = PersistentAgent.objects.create(
            user=self.user,
            name="PeerGenerationAgent",
            charter="Receive peer work",
            browser_use_agent=peer_browser_agent,
        )
        AgentPeerLink.objects.create(agent_a=self.agent, agent_b=peer_agent, created_by=self.user)

        with patch("api.agent.tasks.process_agent_events_task") as task_mock, patch(
            "api.agent.peer_comm.transaction.on_commit",
            lambda cb, **kwargs: cb(),
        ):
            task_mock.delay = MagicMock()
            PeerMessagingService(self.agent, peer_agent).send_message("handoff")

        self.assertEqual(get_human_inbound_generation(peer_agent.id), 0)
        task_mock.delay.assert_called_once_with(str(peer_agent.id))

    def test_redundant_queued_task_skips_after_generation_is_consumed(self):
        generation = bump_human_inbound_generation(self.agent.id)
        mark_human_inbound_generation_consumed(self.agent.id, generation)

        with patch("api.agent.tasks.process_events.process_agent_events") as mock_process:
            process_agent_events_task.push_request(id="generation-skip", delivery_info={})
            try:
                process_agent_events_task.run(str(self.agent.id), inbound_generation=generation)
            finally:
                process_agent_events_task.pop_request()

        mock_process.assert_not_called()

    def test_new_human_generation_advances_when_current_key_expired_before_consumed(self):
        generation = bump_human_inbound_generation(self.agent.id)
        mark_human_inbound_generation_consumed(self.agent.id, generation)
        get_redis_client().delete(_human_inbound_generation_key(self.agent.id))

        next_generation = bump_human_inbound_generation(self.agent.id)

        self.assertEqual(next_generation, generation + 1)
        self.assertFalse(is_human_inbound_generation_consumed(self.agent.id, next_generation))

    def test_process_agent_events_task_logs_task_quota_error(self):
        exc = ValidationError({"quota": "Task quota exceeded. You have no remaining task credits."})

        with patch("api.agent.tasks.process_events.process_agent_events", side_effect=exc):
            process_agent_events_task.push_request(id="quota-task", delivery_info={})
            try:
                with self.assertRaises(ValidationError):
                    process_agent_events_task.run(str(self.agent.id))
            finally:
                process_agent_events_task.pop_request()

        error = PersistentAgentError.objects.get(agent=self.agent)
        self.assertEqual(error.category, PersistentAgentError.Category.TASK_QUOTA_EXCEEDED)
        self.assertEqual(error.source, "api.agent.tasks.process_events.process_agent_events_task")
        self.assertEqual(error.context["task_id"], "quota-task")

    @patch("api.agent.tasks.process_events.switch_is_active", return_value=False)
    def test_process_agent_cron_trigger_logs_task_quota_error(self, _mock_switch):
        exc = ValidationError({"quota": "Task quota exceeded. You have no remaining task credits."})

        with patch("api.agent.tasks.process_events.process_agent_events", side_effect=exc):
            process_agent_cron_trigger_task.push_request(id="quota-cron", delivery_info={})
            try:
                process_agent_cron_trigger_task.run(str(self.agent.id), "* * * * *")
            finally:
                process_agent_cron_trigger_task.pop_request()

        error = PersistentAgentError.objects.get(agent=self.agent)
        self.assertEqual(error.category, PersistentAgentError.Category.TASK_QUOTA_EXCEEDED)
        self.assertEqual(error.source, "api.agent.tasks.process_events.process_agent_cron_trigger_task")
        self.assertEqual(error.context["task_id"], "quota-cron")


@tag("batch_event_processing")
class OrchestratorHumanInputInterruptTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="interrupt@example.com",
            email="interrupt@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="InterruptBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="InterruptAgent",
            charter="Handle prompt interrupts",
            browser_use_agent=self.browser_agent,
        )

    def _prompt_context(self):
        return ([{"role": "system", "content": "sys"}], 1000, None, {})

    @patch("api.agent.core.event_processing._schedule_agent_follow_up")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.maybe_run_agent_judge")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value=BurnRateAction.PAUSED)
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[])
    @patch("api.agent.core.event_processing.build_prompt_context")
    def test_burn_rate_pause_triggers_judge_before_exiting_loop(
        self,
        mock_build_prompt,
        _mock_tools,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_burn_control,
        mock_judge,
        mock_pending_settings,
        _mock_follow_up,
    ):
        mock_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )

        usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(usage.get("total_tokens"), 0)
        mock_build_prompt.assert_not_called()
        mock_judge.assert_called_once_with(
            self.agent,
            tools=[],
            extra_trigger_reasons=["burn_rate_throttled"],
        )

    @patch("api.agent.core.event_processing._schedule_agent_follow_up")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.maybe_run_agent_judge")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value=BurnRateAction.STEPPED_DOWN)
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[])
    @patch("api.agent.core.event_processing._completion_with_failover")
    @patch("api.agent.core.event_processing.build_prompt_context")
    def test_burn_rate_tier_step_down_triggers_judge(
        self,
        mock_build_prompt,
        mock_completion,
        _mock_tools,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_burn_control,
        mock_judge,
        mock_pending_settings,
        _mock_follow_up,
    ):
        mock_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )

        def _build_prompt_and_interrupt(*_args, **_kwargs):
            bump_human_inbound_generation(self.agent.id)
            return self._prompt_context()

        mock_build_prompt.side_effect = _build_prompt_and_interrupt

        from api.agent.core import event_processing as ep

        with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1):
            usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(usage.get("total_tokens"), 0)
        mock_completion.assert_not_called()
        mock_judge.assert_called_once_with(
            self.agent,
            tools=[],
            extra_trigger_reasons=["burn_rate_tier_step_down"],
        )

    @patch("api.agent.core.event_processing._schedule_agent_follow_up")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value="none")
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[])
    @patch("api.agent.core.event_processing._completion_with_failover")
    @patch("api.agent.core.event_processing.build_prompt_context")
    def test_prompt_build_retries_without_llm_when_generation_changes_before_send(
        self,
        mock_build_prompt,
        mock_completion,
        _mock_tools,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_burn_control,
        mock_pending_settings,
        _mock_follow_up,
    ):
        mock_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )

        def _build_prompt_and_interrupt(*_args, **_kwargs):
            bump_human_inbound_generation(self.agent.id)
            return self._prompt_context()

        mock_build_prompt.side_effect = _build_prompt_and_interrupt

        from api.agent.core import event_processing as ep

        with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1):
            usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(usage.get("total_tokens"), 0)
        mock_completion.assert_not_called()

    @patch("api.agent.core.event_processing._schedule_agent_follow_up")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value="none")
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[])
    @patch("api.agent.core.event_processing._completion_with_failover")
    @patch("api.agent.core.event_processing.build_prompt_context")
    def test_generation_is_not_consumed_when_completion_fails(
        self,
        mock_build_prompt,
        mock_completion,
        _mock_tools,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_burn_control,
        mock_pending_settings,
        _mock_follow_up,
    ):
        mock_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )
        generation = bump_human_inbound_generation(self.agent.id)
        mock_build_prompt.return_value = self._prompt_context()
        mock_completion.side_effect = RuntimeError("provider unavailable")

        from api.agent.core import event_processing as ep

        with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1):
            usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(usage.get("total_tokens"), 0)
        self.assertEqual(get_consumed_human_inbound_generation(self.agent.id), 0)
        self.assertEqual(get_human_inbound_generation(self.agent.id), generation)
        error = PersistentAgentError.objects.get(agent=self.agent)
        self.assertEqual(error.category, PersistentAgentError.Category.LLM_COMPLETION)
        self.assertEqual(error.exception_class, "RuntimeError")
        self.assertEqual(error.context["iteration"], 1)

    @patch("api.agent.core.event_processing._schedule_agent_follow_up")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value="none")
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[])
    @patch("api.agent.core.event_processing._completion_with_failover")
    @patch("api.agent.core.event_processing.build_prompt_context")
    def test_empty_completion_retries_agent_loop_before_logging_error(
        self,
        mock_build_prompt,
        mock_completion,
        _mock_tools,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_burn_control,
        mock_pending_settings,
        _mock_follow_up,
    ):
        mock_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )
        mock_build_prompt.return_value = self._prompt_context()
        response = make_completion_response(content="Done")
        token_usage = {
            "prompt_tokens": 10,
            "completion_tokens": 5,
            "total_tokens": 15,
            "model": "m",
            "provider": "p",
        }
        mock_completion.side_effect = [
            EmptyLiteLLMResponseError("empty response"),
            (response, token_usage),
        ]

        from api.agent.core import event_processing as ep

        with (
            patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 2),
            patch.object(ep.settings, "AGENT_EMPTY_LLM_RESPONSE_LOOP_RETRIES", 1),
        ):
            usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(mock_completion.call_count, 2)
        self.assertEqual(usage.get("total_tokens"), 15)
        self.assertFalse(PersistentAgentError.objects.filter(agent=self.agent).exists())

    @patch("api.agent.core.event_processing._schedule_agent_follow_up")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value="none")
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[])
    @patch("api.agent.core.event_processing.build_prompt_context")
    def test_prompt_construction_failure_records_error(
        self,
        mock_build_prompt,
        _mock_tools,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_burn_control,
        mock_pending_settings,
        _mock_follow_up,
    ):
        mock_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )
        mock_build_prompt.side_effect = RuntimeError("template exploded")

        from api.agent.core import event_processing as ep

        with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1), self.assertRaises(RuntimeError):
            _run_agent_loop(self.agent, is_first_run=False, run_sequence_number=4)

        error = PersistentAgentError.objects.get(agent=self.agent)
        self.assertEqual(error.category, PersistentAgentError.Category.PROMPT_CONSTRUCTION)
        self.assertEqual(error.exception_class, "RuntimeError")
        self.assertEqual(error.context["iteration"], 1)
        self.assertEqual(error.context["run_sequence_number"], 4)

    def test_tool_persistence_failure_records_error(self):
        with patch(
            "api.models.PersistentAgentToolCall.objects.create",
            side_effect=DatabaseError("tool call insert failed"),
        ):
            step = _persist_tool_call_step(
                self.agent,
                "sqlite_query",
                {"sql": "select 1"},
                "ok",
                42,
                "complete",
                None,
                None,
                lambda step_kwargs: None,
                lambda created_step: None,
            )

        self.assertIsNone(step)
        error = PersistentAgentError.objects.get(agent=self.agent)
        self.assertEqual(error.category, PersistentAgentError.Category.TOOL_PERSISTENCE)
        self.assertEqual(error.exception_class, "DatabaseError")
        self.assertEqual(error.context["tool_name"], "sqlite_query")
        self.assertEqual(error.context["param_keys"], ["sql"])
        self.assertEqual(error.context["execution_duration_ms"], 42)

    @patch("api.agent.core.event_processing.run_completion")
    def test_streaming_completion_cancels_when_generation_changes_mid_stream(self, mock_run_completion):
        accepted_generation = get_human_inbound_generation(self.agent.id)

        def _chunk(content: str):
            return SimpleNamespace(
                choices=[
                    SimpleNamespace(
                        delta=SimpleNamespace(content=content),
                        finish_reason=None,
                    )
                ],
            )

        def _stream():
            yield _chunk("partial")
            bump_human_inbound_generation(self.agent.id)
            yield _chunk("stale")

        mock_run_completion.return_value = _stream()
        broadcaster = MagicMock()

        with self.assertRaises(OrchestratorPromptStale):
            _completion_with_failover(
                messages=[{"role": "system", "content": "sys"}],
                tools=[],
                failover_configs=[("mock", "mock-model", {})],
                agent_id=str(self.agent.id),
                stream_broadcaster=broadcaster,
                stale_prompt_checker=lambda: get_human_inbound_generation(self.agent.id) > accepted_generation,
            )

        broadcaster.start.assert_called_once()
        broadcaster.push_delta.assert_called_once_with(None, "partial")
        broadcaster.cancel.assert_called()
        broadcaster.finish.assert_not_called()

    @patch("api.agent.core.event_processing._schedule_agent_follow_up")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value="none")
    @patch("api.agent.core.event_processing.resolve_web_stream_target", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[{"type": "function", "function": {"name": "enable_tools", "parameters": {"type": "object", "properties": {}}}}])
    @patch("api.agent.core.event_processing.get_llm_config_with_failover", return_value=[("mock", "mock-model", {})])
    @patch("api.agent.core.event_processing.build_prompt_context")
    @patch("api.agent.core.event_processing.execute_enabled_tool")
    @patch("api.agent.core.event_processing.run_completion")
    def test_stale_non_streaming_response_is_discarded_before_tool_execution(
        self,
        mock_run_completion,
        mock_execute_tool,
        mock_build_prompt,
        _mock_failover,
        _mock_tools,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_stream_target,
        _mock_burn_control,
        mock_pending_settings,
        _mock_follow_up,
    ):
        mock_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )
        mock_build_prompt.return_value = self._prompt_context()

        def _return_stale_response(*_args, **_kwargs):
            bump_human_inbound_generation(self.agent.id)
            return make_completion_response(tool_names=["sqlite_batch"])

        mock_run_completion.side_effect = _return_stale_response

        from api.agent.core import event_processing as ep

        with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1):
            usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(usage.get("total_tokens"), 0)
        mock_execute_tool.assert_not_called()
        self.assertFalse(PersistentAgentCompletion.objects.filter(agent=self.agent).exists())

    @patch("api.agent.core.event_processing._schedule_agent_follow_up")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value="none")
    @patch("api.agent.core.event_processing.resolve_web_stream_target", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing.get_llm_config_with_failover", return_value=[("mock", "mock-model", {})])
    @patch("api.agent.core.event_processing.get_agent_tools")
    @patch("api.agent.core.event_processing._completion_with_failover")
    @patch("api.agent.core.event_processing.build_prompt_context")
    def test_skip_planning_refreshes_agent_and_tools_after_stale_completion(
        self,
        mock_build_prompt,
        mock_completion,
        mock_get_tools,
        _mock_failover,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_stream_target,
        _mock_burn_control,
        mock_pending_settings,
        _mock_follow_up,
    ):
        self.agent.planning_state = PersistentAgent.PlanningState.PLANNING
        self.agent.save(update_fields=["planning_state", "updated_at"])
        mock_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )
        planning_tools = [
            {"type": "function", "function": {"name": "end_planning", "parameters": {"type": "object"}}},
        ]
        skipped_tools = [
            {"type": "function", "function": {"name": "update_plan", "parameters": {"type": "object"}}},
        ]
        prompt_planning_states = []

        def _tools_for_agent(agent):
            if agent.planning_state == PersistentAgent.PlanningState.PLANNING:
                return planning_tools
            return skipped_tools

        def _build_prompt(agent, *_args, **_kwargs):
            prompt_planning_states.append(agent.planning_state)
            return self._prompt_context()

        def _completion(*_args, **kwargs):
            if mock_completion.call_count == 1:
                PersistentAgent.objects.filter(pk=self.agent.pk).update(
                    planning_state=PersistentAgent.PlanningState.SKIPPED,
                    updated_at=timezone.now(),
                )
                bump_human_inbound_generation(self.agent.id)
                stale_prompt_checker = kwargs["stale_prompt_checker"]
                self.assertTrue(stale_prompt_checker())
                raise OrchestratorPromptStale("Prompt became stale during test.")
            response_message = SimpleNamespace(
                content="",
                tool_calls=None,
                function_call=None,
                reasoning_content=None,
            )
            return (
                SimpleNamespace(
                    choices=[SimpleNamespace(message=response_message)],
                    model="mock-model",
                    provider="mock-provider",
                    model_extra={},
                ),
                {
                    "prompt_tokens": 1,
                    "completion_tokens": 1,
                    "total_tokens": 2,
                    "cached_tokens": 0,
                    "model": "mock-model",
                    "provider": "mock-provider",
                },
            )

        mock_get_tools.side_effect = _tools_for_agent
        mock_build_prompt.side_effect = _build_prompt
        mock_completion.side_effect = _completion

        from api.agent.core import event_processing as ep

        with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 2):
            usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(prompt_planning_states, [
            PersistentAgent.PlanningState.PLANNING,
            PersistentAgent.PlanningState.SKIPPED,
        ])
        first_tools = mock_completion.call_args_list[0].kwargs["tools"]
        second_tools = mock_completion.call_args_list[1].kwargs["tools"]
        self.assertIn("end_planning", _tool_definition_names_for_completion(first_tools))
        second_tool_names = _tool_definition_names_for_completion(second_tools)
        self.assertNotIn("end_planning", second_tool_names)
        self.assertIn("update_plan", second_tool_names)
        self.assertEqual(usage.get("total_tokens"), 2)

    @patch("api.agent.core.event_processing._close_active_cycle_for_skipped_agent")
    @patch("api.agent.core.event_processing.clear_processing_work_state")
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[])
    def test_loop_refresh_missing_agent_clears_processing_state(
        self,
        _mock_get_tools,
        mock_clear_processing_state,
        mock_close_cycle,
    ):
        agent_id = self.agent.id

        with patch.object(self.agent, "refresh_from_db", side_effect=PersistentAgent.DoesNotExist):
            usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(usage.get("total_tokens"), 0)
        mock_clear_processing_state.assert_called_once_with(agent_id, client=ANY)
        mock_close_cycle.assert_called_once()
        _, close_kwargs = mock_close_cycle.call_args
        self.assertEqual(close_kwargs["budget_id"], None)
        self.assertEqual(close_kwargs["check_context"], "loop_refresh_missing")


@tag("batch_event_processing")
class EventProcessingDeletedAgentAbortTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="deleted_abort@example.com",
            email="deleted_abort@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="DeletedAbortBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="DeletedAbortAgent",
            charter="Stop when deleted",
            browser_use_agent=self.browser_agent,
        )

    @patch("api.agent.core.event_processing._attempt_cycle_close_for_sleep")
    @patch("api.agent.core.event_processing._ensure_credit_for_tool", return_value={"cost": None, "credit": None})
    @patch("api.agent.core.event_processing.execute_enabled_tool")
    @patch("api.agent.core.event_processing.apply_sqlite_agent_config_updates", return_value=MagicMock(errors=[]))
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing._enforce_tool_rate_limit", return_value=True)
    @patch("api.agent.core.event_processing.extract_reasoning_content", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[{"type": "function", "function": {"name": "sqlite_batch", "parameters": {"type": "object", "properties": {}}}}])
    @patch("api.agent.core.event_processing.get_llm_config_with_failover", return_value=[("mock", "mock-model", {})])
    @patch("api.agent.core.event_processing.build_prompt_context", return_value=([{"role": "system", "content": "sys"}], 1000, None))
    def test_run_agent_loop_stops_when_agent_deleted_mid_batch(
        self,
        _mock_prompt,
        _mock_failover,
        _mock_tools,
        _mock_rate,
        _mock_seed_config,
        _mock_apply_config,
        _mock_reasoning,
        mock_execute_tool,
        _mock_credit,
        mock_close_cycle,
    ):
        tool_call_one = {
            "id": "call-1",
            "function": {"name": "sqlite_batch", "arguments": '{"sql":"SELECT 1","will_continue_work":true}'},
        }
        tool_call_two = {
            "id": "call-2",
            "function": {"name": "sqlite_batch", "arguments": '{"sql":"SELECT 2","will_continue_work":true}'},
        }

        response_message = MagicMock()
        response_message.tool_calls = [tool_call_one, tool_call_two]
        response_message.function_call = None
        response_message.content = None
        response_choice = MagicMock(message=response_message)
        response = MagicMock(choices=[response_choice], model_extra={})
        token_usage = {
            "prompt_tokens": 5,
            "completion_tokens": 5,
            "total_tokens": 10,
            "model": "mock-model",
            "provider": "mock-provider",
            "cached_tokens": 0,
        }

        def _first_call_deletes_agent(*_args, **_kwargs):
            self.agent.soft_delete()
            return {"status": "ok"}

        mock_execute_tool.side_effect = _first_call_deletes_agent

        with patch("api.agent.core.event_processing._completion_with_failover", return_value=(response, token_usage)):
            from api.agent.core import event_processing as ep

            with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1):
                usage = _run_agent_loop(self.agent, is_first_run=False)

        self.agent.refresh_from_db()
        self.assertTrue(self.agent.is_deleted)
        self.assertEqual(mock_execute_tool.call_count, 1)
        mock_close_cycle.assert_called_once()
        self.assertEqual(usage.get("total_tokens"), 10)

    @patch("api.agent.core.event_processing._attempt_cycle_close_for_sleep")
    @patch("api.agent.core.event_processing._ensure_credit_for_tool", return_value={"cost": None, "credit": None})
    @patch("api.agent.core.event_processing.execute_enabled_tool")
    @patch("api.agent.core.event_processing.apply_sqlite_agent_config_updates", return_value=MagicMock(errors=[]))
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing._enforce_tool_rate_limit", return_value=True)
    @patch("api.agent.core.event_processing.extract_reasoning_content", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[{"type": "function", "function": {"name": "sqlite_batch", "parameters": {"type": "object", "properties": {}}}}])
    @patch("api.agent.core.event_processing.get_llm_config_with_failover", return_value=[("mock", "mock-model", {})])
    @patch("api.agent.core.event_processing.build_prompt_context", return_value=([{"role": "system", "content": "sys"}], 1000, None))
    def test_run_agent_loop_stops_when_agent_deactivated_mid_batch(
        self,
        _mock_prompt,
        _mock_failover,
        _mock_tools,
        _mock_rate,
        _mock_seed_config,
        _mock_apply_config,
        _mock_reasoning,
        mock_execute_tool,
        _mock_credit,
        mock_close_cycle,
    ):
        tool_call_one = {
            "id": "call-1",
            "function": {"name": "sqlite_batch", "arguments": '{"sql":"SELECT 1","will_continue_work":true}'},
        }
        tool_call_two = {
            "id": "call-2",
            "function": {"name": "sqlite_batch", "arguments": '{"sql":"SELECT 2","will_continue_work":true}'},
        }

        response_message = MagicMock()
        response_message.tool_calls = [tool_call_one, tool_call_two]
        response_message.function_call = None
        response_message.content = None
        response_choice = MagicMock(message=response_message)
        response = MagicMock(choices=[response_choice], model_extra={})
        token_usage = {
            "prompt_tokens": 5,
            "completion_tokens": 5,
            "total_tokens": 10,
            "model": "mock-model",
            "provider": "mock-provider",
            "cached_tokens": 0,
        }

        def _first_call_deactivates_agent(*_args, **_kwargs):
            PersistentAgent.objects.filter(pk=self.agent.pk).update(is_active=False)
            return {"status": "ok"}

        mock_execute_tool.side_effect = _first_call_deactivates_agent

        with patch("api.agent.core.event_processing._completion_with_failover", return_value=(response, token_usage)):
            from api.agent.core import event_processing as ep

            with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1):
                usage = _run_agent_loop(self.agent, is_first_run=False)

        self.agent.refresh_from_db()
        self.assertFalse(self.agent.is_active)
        self.assertEqual(mock_execute_tool.call_count, 1)
        mock_close_cycle.assert_called_once()
        self.assertEqual(usage.get("total_tokens"), 10)


@tag("batch_event_processing")
class EventProcessingStopRequestTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="stop_request@example.com",
            email="stop_request@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="StopRequestBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="StopRequestAgent",
            charter="Stop when requested",
            browser_use_agent=self.browser_agent,
        )
        self.addCleanup(lambda: clear_processing_stop_requested(self.agent.id))

    @patch("api.agent.core.event_processing.Redlock")
    @patch("api.agent.core.event_processing._process_agent_events_locked")
    def test_process_agent_events_skips_when_stop_requested_before_start(self, mock_process_locked, mock_redlock):
        set_processing_stop_requested(self.agent.id)
        set_processing_queued_flag(self.agent.id)
        mock_lock = MagicMock()
        mock_lock.acquire.return_value = True
        mock_redlock.return_value = mock_lock

        process_agent_events(self.agent.id)

        mock_process_locked.assert_not_called()
        self.assertFalse(is_processing_stop_requested(self.agent.id))

    @patch("api.agent.core.event_processing._schedule_pending_drain")
    @patch("api.agent.core.event_processing.enqueue_pending_agent")
    @patch("api.agent.core.event_processing._maybe_clear_stale_lock", return_value=False)
    @patch("api.agent.core.event_processing.Redlock")
    @patch("api.agent.core.event_processing._process_agent_events_locked")
    def test_process_agent_events_preserves_stop_request_when_lock_not_acquired(
        self,
        mock_process_locked,
        mock_redlock,
        _mock_clear_stale_lock,
        mock_enqueue_pending,
        _mock_schedule_pending_drain,
    ):
        set_processing_stop_requested(self.agent.id)
        mock_lock = MagicMock()
        mock_lock.acquire.return_value = False
        mock_redlock.return_value = mock_lock

        process_agent_events(self.agent.id)

        mock_process_locked.assert_not_called()
        mock_enqueue_pending.assert_called_once()
        self.assertTrue(is_processing_stop_requested(self.agent.id))

    @patch("api.agent.core.event_processing._attempt_cycle_close_for_sleep")
    @patch("api.agent.core.event_processing._ensure_credit_for_tool", return_value={"cost": None, "credit": None})
    @patch("api.agent.core.event_processing.execute_enabled_tool")
    @patch("api.agent.core.event_processing.apply_sqlite_agent_config_updates", return_value=MagicMock(errors=[]))
    @patch("api.agent.core.event_processing.seed_sqlite_skills", return_value=None)
    @patch("api.agent.core.event_processing.seed_sqlite_agent_config", return_value=None)
    @patch("api.agent.core.event_processing._enforce_tool_rate_limit", return_value=True)
    @patch("api.agent.core.event_processing.extract_reasoning_content", return_value=None)
    @patch("api.agent.core.event_processing.get_agent_tools", return_value=[{"type": "function", "function": {"name": "sqlite_batch", "parameters": {"type": "object", "properties": {}}}}])
    @patch("api.agent.core.event_processing.get_llm_config_with_failover", return_value=[("mock", "mock-model", {})])
    @patch("api.agent.core.event_processing.build_prompt_context", return_value=([{"role": "system", "content": "sys"}], 1000, None))
    def test_run_agent_loop_stops_after_current_tool_when_stop_requested_mid_batch(
        self,
        _mock_prompt,
        _mock_failover,
        _mock_tools,
        _mock_rate,
        _mock_seed_config,
        _mock_seed_skills,
        _mock_apply_config,
        _mock_reasoning,
        mock_execute_tool,
        _mock_credit,
        mock_close_cycle,
    ):
        tool_call_one = {
            "id": "call-1",
            "function": {"name": "sqlite_batch", "arguments": '{"sql":"SELECT 1","will_continue_work":true}'},
        }
        tool_call_two = {
            "id": "call-2",
            "function": {"name": "sqlite_batch", "arguments": '{"sql":"SELECT 2","will_continue_work":true}'},
        }

        response_message = MagicMock()
        response_message.tool_calls = [tool_call_one, tool_call_two]
        response_message.function_call = None
        response_message.content = None
        response_choice = MagicMock(message=response_message)
        response = MagicMock(choices=[response_choice], model_extra={})
        token_usage = {
            "prompt_tokens": 5,
            "completion_tokens": 5,
            "total_tokens": 10,
            "model": "mock-model",
            "provider": "mock-provider",
            "cached_tokens": 0,
        }

        def _first_call_requests_stop(*_args, **_kwargs):
            set_processing_stop_requested(self.agent.id)
            return {"status": "ok"}

        mock_execute_tool.side_effect = _first_call_requests_stop

        with patch("api.agent.core.event_processing._completion_with_failover", return_value=(response, token_usage)):
            from api.agent.core import event_processing as ep

            with patch.object(ep, "MAX_AGENT_LOOP_ITERATIONS", 1):
                usage = _run_agent_loop(self.agent, is_first_run=False)

        self.assertEqual(mock_execute_tool.call_count, 1)
        mock_close_cycle.assert_called_once()
        self.assertFalse(is_processing_stop_requested(self.agent.id))
        self.assertEqual(usage.get("total_tokens"), 10)


@tag("batch_event_processing")
class EventProcessingMaxIterationsFollowUpTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="max_iterations@example.com",
            email="max_iterations@example.com",
            password="secret",
        )
        self.browser_agent = BrowserUseAgent.objects.create(user=self.user, name="MaxIterBA")
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            name="MaxIterAgent",
            charter="Test max iterations follow-up",
            browser_use_agent=self.browser_agent,
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=False)
    @patch("api.agent.tasks.process_events.process_agent_events_task.apply_async")
    @patch("api.agent.core.event_processing.get_pending_drain_settings")
    @patch("api.agent.core.event_processing.handle_burn_rate_limit", return_value="none")
    @patch("api.agent.core.event_processing.get_redis_client")
    def test_run_agent_loop_queues_follow_up_when_max_iterations_reached(
        self,
        mock_get_redis,
        _mock_burn_control,
        mock_get_pending_settings,
        mock_apply_async,
    ):
        enable_tools(self.agent, ["sqlite_batch"])

        class _FakeRedis:
            def get(self, _key):
                return None

            def exists(self, _key):
                return 0

        mock_get_redis.return_value = _FakeRedis()
        mock_get_pending_settings.return_value = PendingDrainSettings(
            pending_set_ttl_seconds=123,
            pending_drain_delay_seconds=10,
            pending_drain_limit=50,
            pending_drain_schedule_ttl_seconds=60,
        )

        tool_call = MagicMock()
        tool_call.function = MagicMock()
        tool_call.function.name = "sqlite_batch"
        tool_call.function.arguments = '{"sql": "UPDATE t SET id = 1", "will_continue_work": false}'

        response_message = MagicMock()
        response_message.tool_calls = [tool_call]
        response_message.function_call = None
        response_message.content = None
        response_message.reasoning_content = None

        response_choice = MagicMock(message=response_message)
        response = MagicMock()
        response.choices = [response_choice]
        response.model_extra = {
            "usage": MagicMock(
                prompt_tokens=5,
                completion_tokens=5,
                total_tokens=10,
                prompt_tokens_details=MagicMock(cached_tokens=0),
            )
        }

        token_usage = {
            "prompt_tokens": 5,
            "completion_tokens": 5,
            "total_tokens": 10,
            "model": "mock-model",
            "provider": "mock-provider",
            "cached_tokens": 0,
        }

        with patch('api.agent.core.prompt_context.ensure_steps_compacted'), \
             patch('api.agent.core.prompt_context.ensure_comms_compacted'), \
             patch('api.agent.core.event_processing.build_prompt_context', return_value=([{"role": "system", "content": "sys"}], 1000, None)), \
             patch('api.agent.core.event_processing.get_llm_config_with_failover', return_value=[("mock", "mock-model", {})]), \
             patch('api.agent.core.event_processing._completion_with_failover', return_value=(response, token_usage)), \
             patch('api.agent.core.event_processing.execute_enabled_tool', return_value={"status": "warning", "message": "0 rows affected"}), \
             patch('api.agent.core.event_processing._ensure_credit_for_tool', return_value={"cost": None, "credit": None}):
            from api.agent.core import event_processing as ep
            with patch.object(ep, 'MAX_AGENT_LOOP_ITERATIONS', 1):
                _run_agent_loop(self.agent, is_first_run=False)

        self.assertTrue(
            self.agent.steps.filter(description__icontains="max iterations").exists()
        )
        expected_delay_seconds = max(
            int(ep.MAX_ITERATIONS_FOLLOWUP_DELAY_SECONDS),
            int(mock_get_pending_settings.return_value.pending_drain_delay_seconds),
        )
        mock_apply_async.assert_called_once_with(
            args=[str(self.agent.id)],
            countdown=expected_delay_seconds,
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    @patch("api.agent.tasks.process_events.process_agent_events_task.apply_async")
    def test_delayed_agent_follow_up_does_not_run_immediately_in_eager_mode(
        self,
        mock_apply_async,
    ):
        from api.agent.core import event_processing as ep

        ep._schedule_agent_follow_up(
            agent_id=self.agent.id,
            delay_seconds=60,
            reason="Test",
        )

        mock_apply_async.assert_not_called()

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    @patch("api.agent.tasks.process_events.process_pending_agent_events_task.apply_async")
    @patch("api.agent.core.event_processing.claim_pending_drain_slot")
    def test_delayed_pending_drain_does_not_run_immediately_in_eager_mode(
        self,
        mock_claim_pending_drain_slot,
        mock_apply_async,
    ):
        from api.agent.core import event_processing as ep

        ep._schedule_pending_drain(
            delay_seconds=60,
            schedule_ttl_seconds=120,
        )

        mock_claim_pending_drain_slot.assert_not_called()
        mock_apply_async.assert_not_called()
