#601 AI 조교 호출 횟수 제한 및 히스토리 영구 저장 기능 구현 (#611)

* #601 AI 조교 호출 횟수 제한 및 히스토리 영구 저장 기능 구현

- 백엔드: ProblemAIHintLog 모델 추가 및 영구 저장 로직 구현
- 백엔드: 일일 30회, 문제당 5회 횟수 제한 및 관리자(Admin) 우회 로직 적용
- 백엔드: 과거 AI 힌트 내역 및 사용 횟수 조회 API(AIHintHistoryAPI) 추가
- 프론트엔드: BottomDrag.vue에 백엔드 API 연동 및 남은 횟수 UI 렌더링 적용
- 프론트엔드: 기존 localStorage 기반의 임시 저장 로직 제거
- 테스트: 횟수 제한 및 API 관련 백엔드 단위 테스트 추가 및 100% 통과

* #611 copilot 리뷰사항 반영

- 남은 횟수 UI표기 변경
- 페이지가 열릴 때가 아닌 AI 조교 패널이 열릴 때 history 불러오기로 수정
- streaming chunk 누적 방식 string -> list로 변경
- 날짜 탐색 방식 범위 검색 방식으로 변경
- v-html="renderText" 보안 위협으로 인한 제거

* #611 renderText 삭제

- 사용하지 않는 renderText를 삭제하였습니다.

* #611 AI 조교 횟수 제한 동시성 이슈 해결

- 브라우저 탭을 여러개 사용하여 횟수 제한을 우회할 수 있는 버그 수정
  - 스트리밍 시작 전 빈 로그를 먼저 생성하여 즉시 횟수 차감 처리 (Transaction 적용)
  - vLLM 통신 에러 및 스트림 중단 시 생성된 빈 로그를 삭제하여 횟수 롤백 로직 추가
- AI 조교 일일 사용 횟수(30회) 제한 제거, 문제당 5회 제한만 유지
- ProblemAIHintLog 모델 성능 최적화: user, created_at 인덱스 추가

* #611 코파일럿 리뷰 반영

- select_for_update()를 활용하여 원자성 충족 -> 동시 요청 병렬적 처리 방지 (DB락)
- daily_count 하드 코딩 제거
- 불필요한 import 삭제
- 일일 제한 테스트 코드 삭제

* Update frontend/src/pages/oj/views/problem/problemSolving/problemSolvingComponent/BottomDrag.vue

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* #611 코파일럿 리뷰 반영

- 내용이 조금이라도 있으면 저장, 1바이트도 없으면 횟수 환불
- 바깥에서 미리 만들어둔 hint_log 객체의 내용만 바꾼 뒤 저장

* #611 user_id, problem_id, created_at 인덱스 생성

- user_id, problem_id, created_at 인덱스 생성

* #611 AI 힌트 공백 응답 시 횟수 차감 버그 및 히스토리 API 오류 수정

- AI 힌트 스트리밍 빈 응답 시 횟수 차감 방지 로직 추가
- 히스토리 조회 API의 비공개/대회용 문제 접근 차단 및 조인 최적화
- 동시성 제어 중 발생하는 SimpleLazyObject 에러 해결 (get_user_model)
- 힌트 생성 및 조회 관련 엣지 케이스 테스트 코드 추가

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: wlsgur11

backend/problem/llm_hint.py

@@ -5,6 +5,7 @@
 
 import requests
 from django.utils.html import strip_tags
+import time
 
 LOCAL_VLLM_CHAT_COMPLETIONS_URL = "http://localhost:8000/v1/chat/completions"
 CLUSTER_VLLM_CHAT_COMPLETIONS_URL = "http://vllm:8000/v1/chat/completions"
@@ -62,9 +63,7 @@ def _format_samples(samples):
     for index, sample in enumerate(samples, start=1):
         input_text = (sample.get("input") or "").strip() or "(비어 있음)"
         output_text = (sample.get("output") or "").strip() or "(비어 있음)"
-        rendered_samples.append(
-            f"[샘플 입력 {index}]\n{input_text}\n[샘플 출력 {index}]\n{output_text}"
-        )
+        rendered_samples.append(f"[샘플 입력 {index}]\n{input_text}\n[샘플 출력 {index}]\n{output_text}")
     return "\n\n".join(rendered_samples)
 
 
@@ -104,8 +103,14 @@ def build_hint_payload(problem, stream=False):
     return {
         "model": VLLM_MODEL,
         "messages": [
-            {"role": "system", "content": SYSTEM_PROMPT},
-            {"role": "user", "content": build_problem_prompt(problem)},
+            {
+                "role": "system",
+                "content": SYSTEM_PROMPT
+            },
+            {
+                "role": "user",
+                "content": build_problem_prompt(problem)
+            },
         ],
         "temperature": 0.55,
         "max_tokens": 512,
@@ -124,14 +129,18 @@ def _extract_stream_delta(response_json):
         return ""
 
     if isinstance(content, list):
-        return "".join(
-            item.get("text", "") if isinstance(item, dict) else str(item)
-            for item in content
-        )
+        return "".join(item.get("text", "") if isinstance(item, dict) else str(item) for item in content)
     return str(content)
 
 
 def stream_problem_hint(problem):
+    if os.getenv("IS_LOCAL_TEST") == "True":
+        mock_response = "이것은 로컬 테스트용 힌트입니다. 문제의 입력을 다시 확인해보세요."
+        for char in mock_response:
+            yield char
+            time.sleep(0.05)    # 실제 스트리밍 느낌을 위해 딜레이 추가
+        return
+
     response = None
     try:
         response = requests.post(

backend/problem/migrations/0002_problemaihintlog.py

@@ -0,0 +1,30 @@
+# Generated by Django 3.2.25 on 2026-04-14 05:33
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('problem', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ProblemAIHintLog',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('hint_content', models.TextField()),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('problem', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='problem.problem')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'db_table': 'problem_ai_hint_log',
+                'ordering': ['-created_at'],
+            },
+        ),
+    ]

backend/problem/migrations/0003_auto_20260421_1122.py

@@ -0,0 +1,25 @@
+# Generated by Django 3.2.25 on 2026-04-21 02:22
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('problem', '0002_problemaihintlog'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='problemaihintlog',
+            options={},
+        ),
+        migrations.AddIndex(
+            model_name='problemaihintlog',
+            index=models.Index(fields=['user', 'created_at'], name='problem_ai__user_id_a502c8_idx'),
+        ),
+        migrations.AddIndex(
+            model_name='problemaihintlog',
+            index=models.Index(fields=['problem', 'created_at'], name='problem_ai__problem_3abff5_idx'),
+        ),
+    ]

backend/problem/migrations/0004_problemaihintlog_problem_ai__user_id_7fac82_idx.py

@@ -0,0 +1,17 @@
+# Generated by Django 3.2.25 on 2026-04-21 05:14
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('problem', '0003_auto_20260421_1122'),
+    ]
+
+    operations = [
+        migrations.AddIndex(
+            model_name='problemaihintlog',
+            index=models.Index(fields=['user', 'problem', 'created_at'], name='problem_ai__user_id_7fac82_idx'),
+        ),
+    ]

backend/problem/models.py

@@ -6,6 +6,22 @@
 from utils.constants import Choices
 
 
+class ProblemAIHintLog(models.Model):
+    user = models.ForeignKey("account.User", on_delete=models.CASCADE)
+    problem = models.ForeignKey("Problem", on_delete=models.CASCADE)
+    hint_content = models.TextField()
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    class Meta:
+        db_table = "problem_ai_hint_log"
+
+        indexes = [
+            models.Index(fields=['user', 'created_at']),
+            models.Index(fields=['problem', 'created_at']),
+            models.Index(fields=['user', 'problem', 'created_at']),
+        ]
+
+
 class ProblemTag(models.Model):
     name = models.TextField()
 

backend/problem/serializers.py

@@ -7,10 +7,17 @@
 from utils.constants import Difficulty
 from utils.serializers import LanguageNameMultiChoiceField, SPJLanguageNameChoiceField, LanguageNameChoiceField
 
-from .models import Problem, ProblemRuleType, ProblemTag, ProblemIOMode, ProblemDifficulty
+from .models import Problem, ProblemRuleType, ProblemTag, ProblemIOMode, ProblemDifficulty, ProblemAIHintLog
 from .utils import parse_problem_template
 
 
+class AIHintLogSerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = ProblemAIHintLog
+        fields = ["id", "hint_content", "created_at"]
+
+
 class TestCaseUploadForm(forms.Form):
     spj = forms.CharField(max_length=12)
     file = forms.FileField()

backend/problem/tests.py

@@ -16,13 +16,12 @@
 from unittest import mock
 
 from .models import ProblemTag, ProblemIOMode, get_default_week_info
-from .models import Problem, ProblemRuleType
+from .models import Problem, ProblemRuleType, ProblemAIHintLog
 from .tasks import update_weekly_stats, update_bonus_problem
 from contest.models import Contest
 from contest.tests import DEFAULT_CONTEST_DATA
-from .llm_hint import (CLUSTER_VLLM_CHAT_COMPLETIONS_URL, LOCAL_VLLM_CHAT_COMPLETIONS_URL,
-                       VLLM_CONNECT_TIMEOUT_SEC, VLLM_MODEL, VLLM_STREAM_READ_TIMEOUT_SEC,
-                       get_vllm_chat_completions_url)
+from .llm_hint import (CLUSTER_VLLM_CHAT_COMPLETIONS_URL, LOCAL_VLLM_CHAT_COMPLETIONS_URL, VLLM_CONNECT_TIMEOUT_SEC,
+                       VLLM_MODEL, VLLM_STREAM_READ_TIMEOUT_SEC, get_vllm_chat_completions_url)
 
 from .views.admin import TestCaseAPI
 from .utils import parse_problem_template
@@ -248,19 +247,21 @@ def test_get_one_problem(self):
 class ProblemLLMHintAPITest(ProblemCreateTestBase):
 
     def setUp(self):
+        os.environ["IS_LOCAL_TEST"] = "False"
+
         self.create_school_fixtures(college_id=1, college_name="Test", department_id=1, department_name="Test")
         self.url = self.reverse("problem_llm_hint_api")
         self.admin = self.create_admin(login=False)
         self.problem = self.add_problem(DEFAULT_PROBLEM_DATA, self.admin)
         self.hidden_problem = self.create_problem_with_custom_field(self.admin, _id="A-211", visible=False)
-        self.create_user(email="test@test.com", username="test", password="test1234!")
+        # self.create_user(email="test@test.com", username="test", password="test1234!")
+        self.user = self.create_user(email="test@test.com", username="test", password="test1234!")
+        self.client.force_login(self.user)
 
     @staticmethod
     def _streaming_body(response):
         return "".join(
-            chunk.decode("utf-8") if isinstance(chunk, bytes) else chunk
-            for chunk in response.streaming_content
-        )
+            chunk.decode("utf-8") if isinstance(chunk, bytes) else chunk for chunk in response.streaming_content)
 
     @staticmethod
     def _mock_streaming_response(lines):
@@ -286,22 +287,81 @@ def test_stream_llm_hint(self, mocked_post):
         self.assertIn(json.dumps({"text": "힌트 "}, ensure_ascii=False), body)
         self.assertIn(json.dumps({"text": "스트림"}, ensure_ascii=False), body)
         self.assertIn('event: done', body)
+        self.assertEqual(ProblemAIHintLog.objects.filter(user=self.user, problem=self.problem).count(), 1)
+        saved_log = ProblemAIHintLog.objects.filter(user=self.user, problem=self.problem).first()
+        self.assertEqual(saved_log.hint_content, "힌트 스트림")
+
         mocked_post.assert_called_once()
         self.assertEqual(mocked_post.call_args.args[0], get_vllm_chat_completions_url())
         self.assertEqual(mocked_post.call_args.kwargs["json"]["model"], VLLM_MODEL)
-        self.assertIn("힌트는 정확히 1개만 제공하라.", mocked_post.call_args.kwargs["json"]["messages"][0]["content"])
-        self.assertIn("너무 추상적인 조언 대신", mocked_post.call_args.kwargs["json"]["messages"][0]["content"])
+        self.assertIn("힌트는 정확히 1개의 핵심 아이디어만 제공하라.", mocked_post.call_args.kwargs["json"]["messages"][0]["content"])
+        self.assertIn("너무 추상적인 조언은 피하고,", mocked_post.call_args.kwargs["json"]["messages"][0]["content"])
         self.assertIn(self.problem._id, mocked_post.call_args.kwargs["json"]["messages"][1]["content"])
         self.assertNotIn("<p>", mocked_post.call_args.kwargs["json"]["messages"][1]["content"])
         self.assertIn("짧은 힌트 하나만 작성해라.", mocked_post.call_args.kwargs["json"]["messages"][1]["content"])
         self.assertIn("관찰 포인트를 짚어라.", mocked_post.call_args.kwargs["json"]["messages"][1]["content"])
-        self.assertEqual(mocked_post.call_args.kwargs["json"]["temperature"], 0.65)
+        self.assertEqual(mocked_post.call_args.kwargs["json"]["temperature"], 0.55)
         self.assertEqual(mocked_post.call_args.kwargs["stream"], True)
         self.assertEqual(
             mocked_post.call_args.kwargs["timeout"],
             (VLLM_CONNECT_TIMEOUT_SEC, VLLM_STREAM_READ_TIMEOUT_SEC),
         )
 
+    @mock.patch("problem.llm_hint.requests.post")
+    def test_stream_llm_hint_problem_limit(self, mocked_post):
+        """한 문제당 5회 제한에 걸리는지 테스트"""
+        # 해당 문제에 대해 이미 5개의 힌트 로그가 존재하도록 세팅
+        for i in range(5):
+            ProblemAIHintLog.objects.create(user=self.user, problem=self.problem, hint_content=f"더미 {i}")
+
+        resp = self.client.get(f"{self.url}?problem_id={self.problem._id}")
+        body = self._streaming_body(resp)
+
+        self.assertIn('event: app-error', body)
+        self.assertIn("limit-exceeded", body)
+        self.assertIn("소진", body)
+        mocked_post.assert_not_called()    # 제한에 걸리면 LLM 호출이 아예 안 일어나야 함
+
+    @mock.patch("problem.llm_hint.requests.post")
+    def test_stream_llm_hint_admin_bypass(self, mocked_post):
+        """관리자는 횟수 제한 없이 무제한으로 사용 가능한지 테스트"""
+        mocked_post.return_value = self._mock_streaming_response([
+            'data: {"choices":[{"delta":{"content":"어드민 패스"}}]}',
+            "data: [DONE]",
+        ])
+
+        # Admin 계정으로 5회 꽉 채움
+        for i in range(5):
+            ProblemAIHintLog.objects.create(user=self.admin, problem=self.problem, hint_content=f"더미 {i}")
+
+        # 일반 유저를 로그아웃시키고 Admin으로 로그인
+        self.client.force_login(self.admin)
+        resp = self.client.get(f"{self.url}?problem_id={self.problem._id}")
+        body = self._streaming_body(resp)
+
+        # 제한에 걸리지 않고 정상 응답이 와야 함
+        self.assertNotIn("limit-exceeded", body)
+        self.assertIn("어드민 패스", body)
+        mocked_post.assert_called_once()
+
+    @mock.patch("problem.llm_hint.requests.post")
+    def test_stream_llm_hint_empty_response(self, mocked_post):
+        """정상적으로 스트리밍이 종료되었으나 텍스트가 비어있는 경우, 선제 생성된 로그가 삭제되는지 테스트"""
+        # 공백만 리턴하는 모델 응답 Mocking
+        mocked_post.return_value = self._mock_streaming_response([
+            'data: {"choices":[{"delta":{"content":"   "}}]}',
+            "data: [DONE]",
+        ])
+
+        resp = self.client.get(f"{self.url}?problem_id={self.problem._id}")
+        body = self._streaming_body(resp)
+
+        # 1. 정상적으로 done 이벤트가 와야 함
+        self.assertIn('event: done', body)
+
+        # 2. 가장 중요: 빈 텍스트였기 때문에 DB에 로그가 남아있으면 안 됨 (횟수 차감 방어 성공)
+        self.assertEqual(ProblemAIHintLog.objects.filter(user=self.user, problem=self.problem).count(), 0)
+
     def test_stream_llm_hint_requires_login(self):
         self.client.logout()
 
@@ -372,6 +432,80 @@ def test_get_vllm_chat_completions_url_for_kubernetes(self):
             self.assertEqual(get_vllm_chat_completions_url(), CLUSTER_VLLM_CHAT_COMPLETIONS_URL)
 
 
+class AIHintHistoryAPITest(ProblemCreateTestBase):
+
+    def setUp(self):
+        self.create_school_fixtures(college_id=1, college_name="Test", department_id=1, department_name="Test")
+        self.url = self.reverse("problem_ai_hint_history_api")    # urls/oj.py 에 등록했던 name과 일치해야 함
+        self.admin = self.create_admin(login=False)
+        self.problem = self.add_problem(DEFAULT_PROBLEM_DATA, self.admin)
+        self.user = self.create_user(email="history@test.com", username="history_test", password="test1234!")
+        self.hidden_problem = self.create_problem_with_custom_field(self.admin, _id="A-211", visible=False)
+
+    def test_get_history_require_login(self):
+        """로그인하지 않은 유저가 조회 요청 시 에러"""
+        self.client.logout()
+        resp = self.client.get(f"{self.url}?problem_id={self.problem._id}")
+        # 일반 APIView 응답 구조에 맞춰 검증 (에러 메시지 포함 여부)
+        self.assertIsNotNone(resp.data.get("error"))
+
+    def test_get_history_success_and_structure(self):
+        """정상적으로 로그와 횟수 데이터가 넘어오는지 구조 검증"""
+        self.client.force_login(self.user)
+
+        # 더미 데이터 생성
+        ProblemAIHintLog.objects.create(user=self.user, problem=self.problem, hint_content="과거의 힌트 1")
+        ProblemAIHintLog.objects.create(user=self.user, problem=self.problem, hint_content="과거의 힌트 2")
+
+        resp = self.client.get(f"{self.url}?problem_id={self.problem._id}")
+        self.assertEqual(resp.status_code, 200)
+        self.assertIsNone(resp.data.get("error"))
+
+        payload = resp.data.get("data")
+
+        # 응답 구조 검증
+        self.assertIn("logs", payload)
+
+        # 값 검증
+        self.assertEqual(len(payload["logs"]), 2)
+        self.assertEqual(payload["logs"][0]["hint_content"], "과거의 힌트 1")
+
+    def test_get_history_with_hidden_problem(self):
+        """비공개 문제의 힌트 히스토리는 조회할 수 없어야 함"""
+        self.client.force_login(self.user)
+
+        # setUp에서 미리 만들어둔 hidden_problem 사용
+        resp = self.client.get(f"{self.url}?problem_id={self.hidden_problem._id}")
+
+        self.assertIsNotNone(resp.data.get("error"))
+        self.assertIn("문제를 찾을 수 없습니다.", resp.data.get("data", ""))
+
+    def test_get_history_with_contest_problem(self):
+        """대회용 문제의 힌트 히스토리는 일반 API로 조회할 수 없어야 함"""
+        self.client.force_login(self.user)
+
+        # 대회 및 대회용 문제 임시 생성
+        from contest.models import Contest
+        from contest.tests import DEFAULT_CONTEST_DATA
+        contest = Contest.objects.create(created_by=self.admin, **DEFAULT_CONTEST_DATA)
+        contest_problem = self.create_problem_with_custom_field(self.admin, _id="C-999")
+        contest_problem.contest = contest
+        contest_problem.save(update_fields=["contest"])
+
+        resp = self.client.get(f"{self.url}?problem_id={contest_problem._id}")
+
+        self.assertIsNotNone(resp.data.get("error"))
+        self.assertIn("문제를 찾을 수 없습니다.", resp.data.get("data", ""))
+
+    def test_get_history_with_invalid_problem_id(self):
+        """존재하지 않는 문제 ID를 넣었을 때 처리"""
+        self.client.force_login(self.user)
+        resp = self.client.get(f"{self.url}?problem_id=INVALID_ID")
+
+        self.assertIsNotNone(resp.data.get("error"))
+        self.assertIn("문제를 찾을 수 없습니다.", resp.data.get("data", ""))
+
+
 class ContestProblemAdminTest(APITestCase):
 
     def setUp(self):