feat(backend): Django + OpenCode 백엔드 추가 — 5_django_backend/

앱이 기대하는 base-backend 계약(envelope·JWT·/chat/stream SSE)을 그대로 구현. Django 는 로그인·세션 미러(SQLite/PG)·OpenCode 이벤트 번역만 맡고, 답변은 전용 OpenCode 인스턴스(opencode/ workspace, codeassist 에이전트)가 만듦. - accounts: 이메일 로그인, access 60분 / refresh 14일, entra/config 는 501 - chat: 세션 목록·검색·메시지·취소 + POST /chat/stream 어댑터(part.delta→token, idle→usage/done, 클라 끊겨도 턴 감시 태스크가 DB 마무리, 첫 이벤트 60초 타임아웃) - OpenCode 1.18 멀티 프로젝트라 모든 요청에 ?directory= 부착 - docs-lib 에 OpenCode SDK 1.18.6 타입 원본, pytest 32 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-16 21:28:31 +09:00
parent 08fdc1a258
commit 2869985139
49 changed files with 5884 additions and 5 deletions
@@ -0,0 +1,17 @@
+# CodeAssist 전용 OpenCode 서버 (opencode/ 폴더에서 `opencode serve --port 4096`)
+OPENCODE_BASE_URL=http://localhost:4096
+OPENCODE_AGENT=codeassist
+
+# 토큰 서명 키 (비우면 data/.auth_secret 자동 생성)
+AUTH_SECRET=
+
+# DB — 비우면 SQLite(data/app.db)
+DB_HOST=
+DB_PORT=5432
+DB_USER=
+DB_PASSWORD=
+DB_NAME=
+DB_SCHEMA=codeassist
+
+# 프론트 origin (dev 는 Vite proxy 라 안 탐. Tauri 빌드용)
+CORS_ORIGINS=http://localhost:15173,http://tauri.localhost,tauri://localhost
@@ -0,0 +1,7 @@
+.venv/
+data/
+.env
+__pycache__/
+*.pyc
+staticfiles/
+.pytest_cache/
@@ -0,0 +1,77 @@
+# 5_django_backend — CodeAssist 백엔드 (Django + OpenCode)
+
+프론트(`2_frontend`)가 기대하는 base-backend 계약을 그대로 구현한 Django 서버. AI 답변은 CodeAssist 전용 OpenCode 서버(`opencode/`)가 만들고, 여기는 인증·세션 미러·SSE 어댑터를 맡음.
+ABAP_OPENCODE `web/BE` 에서 뼈대(settings·JWT·OpenCode 클라)를 복사해 왔고, 그쪽 서버와는 독립.
+
+```
+CodeAssist(Tauri/React) ──/api/v1──▶ Django(uvicorn :8001) ──HTTP/SSE──▶ opencode serve(:4096) ──▶ LLM
+   Bearer JWT (localStorage)            ├─ accounts  로그인·refresh
+                                        ├─ chat      세션·검색 미러(SQLite/PG) + /chat/stream 어댑터
+                                        └─ opencode/ workspace (AGENTS.md·agent·opencode.json)
+```
+
+## 띄우기
+
+```bash
+# 1) OpenCode (터미널 1)
+cd 5_django_backend/opencode
+export OPENROUTER_API_KEY=sk-or-...
+npx -y opencode-ai@1.18.6 serve --hostname 127.0.0.1 --port 4096
+
+# 2) Django (터미널 2)
+cd 5_django_backend
+python3.12 -m venv .venv && .venv/bin/pip install -r requirements.txt
+cp .env.example .env            # 필요하면 값 수정
+.venv/bin/python manage.py migrate      # 빈 DB 면 admin@codeassist.local / guest@codeassist.local (test1234) 시드
+.venv/bin/uvicorn config.asgi:application --host 0.0.0.0 --port 8001 --reload
+
+# 3) 프론트 (터미널 3) — .env 의 VITE_DEV_API_TARGET 기본이 :8001 이라 그대로
+cd 2_frontend && npm run dev
+```
+
+`runserver` 말고 **uvicorn(ASGI)** 이어야 SSE 가 버퍼링 없이 흐름.
+
+## API (전부 envelope `CommonResponse`)
+
+| 엔드포인트 | 설명 |
+|---|---|
+| `GET /api/v1/health` | 상태 (무인증) |
+| `POST /api/v1/auth/login` `{email,password}` | `TokenResponse` — access 60분 / refresh 14일 |
+| `POST /api/v1/auth/refresh` `{refreshToken}` | access 재발급 (refresh 회전 안 함). 없으면 401 `REFRESH_TOKEN_MISSING` |
+| `POST /api/v1/auth/logout` | null (클라가 토큰 버림) |
+| `GET /api/v1/users/me` | `UserResponse` |
+| `GET /api/v1/auth/entra/config` | 항상 501 — 프론트가 Entra 버튼 숨김 |
+| `GET /api/v1/chat/sessions?page&limit` | 내 세션 목록, 최근순 + meta |
+| `POST /api/v1/chat/sessions` | OpenCode 세션 생성 + 미러 |
+| `GET /api/v1/chat/sessions/search?query&page&limit` | 내 메시지 본문 검색 |
+| `GET /api/v1/chat/sessions/{id}/messages` | 세션 + 메시지 (`isGenerating` 폴링용) |
+| `POST /api/v1/chat/sessions/{id}/cancel` | OpenCode abort + isGenerating 해제 |
+| `POST /api/v1/chat/stream` | SSE — `token{delta}`… `title{title}` `usage{used,limit,ratio,elapsed_ms}` `done{}` / `error{message,code}` |
+
+에러 code: `INVALID_CREDENTIALS` `REFRESH_TOKEN_MISSING` `REFRESH_TOKEN_INVALID` `CHAT_GENERATION_IN_PROGRESS`(409) `UPSTREAM_UNAVAILABLE`(503) `LLM_ERROR` `LLM_ABORTED`.
+
+## 스트림이 도는 법 (`apps/chat/stream.py`)
+
+1. Bearer 검사 → 내 세션 → `is_generating` 이면 409.
+2. user 메시지 미러 저장, `is_generating=true`.
+3. `apps/chat/events.py` 버스(프로세스당 하나가 OpenCode `/event` 구독) 에 세션 큐 등록 → `POST /session/{id}/prompt_async`.
+4. `message.part.updated`(text, delta) → `token`, `session.updated` 제목 → `title`, `session.idle` → OpenCode 에서 최종 메시지·tokens·cost 가져와 assistant 미러 저장 → `usage` → `done`. `session.error` → `error` + 부분 답변 저장.
+5. 이 감시는 클라이언트와 별개 태스크라 창을 닫아도 idle 까지 돌고 DB 를 마무리함 → 프론트 `isGenerating` 폴링이 답변 복구.
+
+## 테스트
+
+```bash
+.venv/bin/python -m pytest -q     # OpenCode 는 가짜(tests/test_stream.py FakeOpencode)
+```
+
+## 폴더
+
+```
+config/      settings · urls · asgi
+common/      envelope.py(응답 포장·CodedError) · opencode_service.py(OpenCode HTTP/SSE)
+apps/accounts  User(email 로그인) · JWT · 시드
+apps/chat      models(ChatSession/ChatMessage) · views · events(버스) · stream(어댑터)
+opencode/    OpenCode workspace — AGENTS.md · opencode.json · .opencode/agent/codeassist.md
+docs-lib/    OpenCode SDK 1.18.6 타입 (API 진실원천)
+tests/       pytest
+```
@@ -0,0 +1,15 @@
+from django.apps import AppConfig
+
+
+class AccountsConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "apps.accounts"
+    verbose_name = "사용자"
+
+    def ready(self) -> None:
+        # migrate 직후 빈 DB 면 기본 계정을 심음
+        from django.db.models.signals import post_migrate
+
+        from .seed import seed_default_users
+
+        post_migrate.connect(seed_default_users, sender=self)
@@ -0,0 +1,43 @@
+"""JWT 인증 — Authorization: Bearer 헤더 또는 ?token= 쿼리(EventSource·다운로드용).
+토큰 발급은 issue_tokens() — 프론트 TokenResponse 모양."""
+
+from rest_framework_simplejwt.authentication import JWTAuthentication
+from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
+from rest_framework_simplejwt.tokens import AccessToken, RefreshToken
+
+
+class BearerOrQueryJWTAuthentication(JWTAuthentication):
+    def authenticate(self, request):
+        result = super().authenticate(request)
+        if result is not None:
+            return result
+        raw = request.query_params.get("token") if hasattr(request, "query_params") else None
+        raw = raw or request.GET.get("token")
+        if not raw:
+            return None
+        validated = self.get_validated_token(raw)
+        return self.get_user(validated), validated
+
+
+def issue_tokens(user, refresh: RefreshToken | None = None) -> dict:
+    """TokenResponse. refresh 를 주면 그걸 유지하고 access 만 새로 찍음(회전 안 함)."""
+    refresh = refresh or RefreshToken.for_user(user)
+    access = refresh.access_token
+    return {
+        "token": str(access),
+        "tokenExpirationTime": int(access["exp"]),
+        "refreshToken": str(refresh),
+        "refreshTokenExpirationTime": int(refresh["exp"]),
+        "tokenType": "bearer",
+        "user": user.as_payload(),
+    }
+
+
+def user_from_token(raw: str):
+    """순수 Django(async) 뷰용 — access 토큰 → 사용자, 실패 시 None."""
+    if not raw:
+        return None
+    try:
+        return BearerOrQueryJWTAuthentication().get_user(AccessToken(raw))
+    except (InvalidToken, TokenError, Exception):
+        return None
@@ -0,0 +1,145 @@
+# Generated by Django 5.1.15 on 2026-09-16 08:38
+
+import django.contrib.auth.models
+import django.contrib.auth.validators
+import django.utils.timezone
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ("auth", "0012_alter_user_first_name_max_length"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="User",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("password", models.CharField(max_length=128, verbose_name="password")),
+                (
+                    "last_login",
+                    models.DateTimeField(
+                        blank=True, null=True, verbose_name="last login"
+                    ),
+                ),
+                (
+                    "is_superuser",
+                    models.BooleanField(
+                        default=False,
+                        help_text="Designates that this user has all permissions without explicitly assigning them.",
+                        verbose_name="superuser status",
+                    ),
+                ),
+                (
+                    "username",
+                    models.CharField(
+                        error_messages={
+                            "unique": "A user with that username already exists."
+                        },
+                        help_text="Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.",
+                        max_length=150,
+                        unique=True,
+                        validators=[
+                            django.contrib.auth.validators.UnicodeUsernameValidator()
+                        ],
+                        verbose_name="username",
+                    ),
+                ),
+                (
+                    "first_name",
+                    models.CharField(
+                        blank=True, max_length=150, verbose_name="first name"
+                    ),
+                ),
+                (
+                    "last_name",
+                    models.CharField(
+                        blank=True, max_length=150, verbose_name="last name"
+                    ),
+                ),
+                (
+                    "is_staff",
+                    models.BooleanField(
+                        default=False,
+                        help_text="Designates whether the user can log into this admin site.",
+                        verbose_name="staff status",
+                    ),
+                ),
+                (
+                    "is_active",
+                    models.BooleanField(
+                        default=True,
+                        help_text="Designates whether this user should be treated as active. Unselect this instead of deleting accounts.",
+                        verbose_name="active",
+                    ),
+                ),
+                (
+                    "date_joined",
+                    models.DateTimeField(
+                        default=django.utils.timezone.now, verbose_name="date joined"
+                    ),
+                ),
+                (
+                    "email",
+                    models.EmailField(
+                        max_length=254, unique=True, verbose_name="이메일"
+                    ),
+                ),
+                (
+                    "user_name",
+                    models.CharField(
+                        blank=True, max_length=40, verbose_name="표시 이름"
+                    ),
+                ),
+                (
+                    "employee_id",
+                    models.CharField(blank=True, max_length=40, verbose_name="사번"),
+                ),
+                (
+                    "department",
+                    models.CharField(blank=True, max_length=80, verbose_name="부서"),
+                ),
+                (
+                    "groups",
+                    models.ManyToManyField(
+                        blank=True,
+                        help_text="The groups this user belongs to. A user will get all permissions granted to each of their groups.",
+                        related_name="user_set",
+                        related_query_name="user",
+                        to="auth.group",
+                        verbose_name="groups",
+                    ),
+                ),
+                (
+                    "user_permissions",
+                    models.ManyToManyField(
+                        blank=True,
+                        help_text="Specific permissions for this user.",
+                        related_name="user_set",
+                        related_query_name="user",
+                        to="auth.permission",
+                        verbose_name="user permissions",
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "사용자",
+                "verbose_name_plural": "사용자",
+            },
+            managers=[
+                ("objects", django.contrib.auth.models.UserManager()),
+            ],
+        ),
+    ]
@@ -0,0 +1,41 @@
+"""사용자 — 프론트 `types/api.ts` UserPayload/UserResponse 모양으로 내보냄.
+로그인 키는 email. username 은 email 과 같은 값으로 채움(Django 필수 칸이라)."""
+
+from django.contrib.auth.models import AbstractUser
+from django.db import models
+
+
+class User(AbstractUser):
+    email = models.EmailField("이메일", unique=True)
+    user_name = models.CharField("표시 이름", max_length=40, blank=True)
+    employee_id = models.CharField("사번", max_length=40, blank=True)
+    department = models.CharField("부서", max_length=80, blank=True)
+
+    class Meta:
+        verbose_name = "사용자"
+        verbose_name_plural = "사용자"
+
+    @property
+    def role(self) -> str:
+        return "ADMIN" if self.is_superuser else "USER"
+
+    def as_payload(self) -> dict:
+        """UserPayload — 로그인 응답 안의 user."""
+        return {
+            "id": str(self.id),
+            "email": self.email,
+            "userName": self.user_name or None,
+            "role": self.role,
+            "employeeId": self.employee_id or None,
+            "department": self.department or None,
+            "authProvider": "local",
+        }
+
+    def as_response(self) -> dict:
+        """UserResponse — GET /users/me."""
+        return {
+            **self.as_payload(),
+            "isActive": self.is_active,
+            "createdAt": self.date_joined.isoformat(),
+            "updatedAt": (self.last_login or self.date_joined).isoformat(),
+        }
@@ -0,0 +1,22 @@
+"""빈 DB 첫 기동 시 기본 계정 — 어느 PC 에서든 바로 로그인되게. 계정 있으면 건너뜀."""
+
+import logging
+
+log = logging.getLogger(__name__)
+
+DEFAULT_USERS = [
+    # (email, password, 표시이름, 관리자)
+    ("admin@codeassist.local", "test1234", "관리자", True),
+    ("guest@codeassist.local", "test1234", "게스트", False),
+]
+
+
+def seed_default_users(**_kwargs) -> None:
+    from .models import User
+
+    if User.objects.exists():
+        return
+    for email, pw, name, admin in DEFAULT_USERS:
+        maker = User.objects.create_superuser if admin else User.objects.create_user
+        maker(username=email, email=email, password=pw, user_name=name)
+    log.info("기본 계정 시드: %s", ", ".join(u[0] for u in DEFAULT_USERS))
@@ -0,0 +1,10 @@
+from rest_framework import serializers
+
+
+class LoginSerializer(serializers.Serializer):
+    email = serializers.EmailField()
+    password = serializers.CharField()
+
+
+class RefreshSerializer(serializers.Serializer):
+    refreshToken = serializers.CharField(required=False, allow_blank=True, default="")
@@ -0,0 +1,10 @@
+from django.urls import path
+
+from .views import EntraConfigView, LoginView, LogoutView, RefreshView
+
+urlpatterns = [
+    path("login", LoginView.as_view()),
+    path("refresh", RefreshView.as_view()),
+    path("logout", LogoutView.as_view()),
+    path("entra/config", EntraConfigView.as_view()),
+]
@@ -0,0 +1,77 @@
+"""
+인증 API — 프론트 `features/auth/api/auth.api.ts` 계약.
+
+  POST /api/v1/auth/login        {email,password} → TokenResponse
+  POST /api/v1/auth/refresh      {refreshToken}   → TokenResponse (access 갱신)
+  POST /api/v1/auth/logout       → null (클라가 토큰 버림. 서버는 회전·블랙리스트 안 함)
+  GET  /api/v1/users/me          → UserResponse
+  GET  /api/v1/auth/entra/config → 501 (Entra 안 씀 — 프론트가 버튼 숨김)
+"""
+
+from rest_framework.permissions import AllowAny
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from rest_framework_simplejwt.exceptions import TokenError
+from rest_framework_simplejwt.tokens import RefreshToken
+
+from common.envelope import CodedError
+
+from .authentication import issue_tokens
+from .models import User
+from .serializers import LoginSerializer, RefreshSerializer
+
+
+class LoginView(APIView):
+    permission_classes = [AllowAny]
+    authentication_classes = []  # 옛 토큰이 붙어 있어도 로그인은 통과시킴
+
+    def post(self, request):
+        ser = LoginSerializer(data=request.data)
+        ser.is_valid(raise_exception=True)
+        user = User.objects.filter(email__iexact=ser.validated_data["email"]).first()
+        if user is None or not user.check_password(ser.validated_data["password"]):
+            raise CodedError(401, "INVALID_CREDENTIALS", "이메일 또는 비밀번호가 올바르지 않아")
+        if not user.is_active:
+            raise CodedError(403, "USER_INACTIVE", "비활성 계정이야")
+        return Response(issue_tokens(user))
+
+
+class RefreshView(APIView):
+    permission_classes = [AllowAny]
+    authentication_classes = []
+
+    def post(self, request):
+        ser = RefreshSerializer(data=request.data or {})
+        ser.is_valid(raise_exception=True)
+        raw = ser.validated_data.get("refreshToken") or ""
+        if not raw:
+            raise CodedError(401, "REFRESH_TOKEN_MISSING", "refresh 토큰이 없어")
+        try:
+            refresh = RefreshToken(raw)
+        except TokenError:
+            raise CodedError(401, "REFRESH_TOKEN_INVALID", "refresh 토큰이 만료됐거나 잘못됐어")
+        user = User.objects.filter(id=refresh["uid"], is_active=True).first()
+        if user is None:
+            raise CodedError(401, "REFRESH_TOKEN_INVALID", "사용자를 찾을 수 없어")
+        return Response(issue_tokens(user, refresh))
+
+
+class LogoutView(APIView):
+    permission_classes = [AllowAny]
+    authentication_classes = []
+
+    def post(self, _request):
+        return Response(None)
+
+
+class MeView(APIView):
+    def get(self, request):
+        return Response(request.user.as_response())
+
+
+class EntraConfigView(APIView):
+    permission_classes = [AllowAny]
+    authentication_classes = []
+
+    def get(self, _request):
+        raise CodedError(501, "NOT_IMPLEMENTED", "Entra 로그인은 지원하지 않아")
@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class ChatConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "apps.chat"
+    verbose_name = "채팅"
@@ -0,0 +1,76 @@
+"""
+OpenCode `/event` 구독 버스 — 프로세스당 하나.
+
+OpenCode 는 서버 전체 이벤트를 SSE 하나로 흘림. 여기서 한 번만 구독하고
+세션별 asyncio.Queue 로 나눠줌. 구독자 없으면 태스크도 안 돌림.
+큐 item: dict(이벤트) 또는 None(하트비트 — 60초 무활동/재연결).
+"""
+
+import asyncio
+import json
+import logging
+
+from common.opencode_service import opencode_service
+
+log = logging.getLogger(__name__)
+
+
+def session_id_of(event: dict) -> str | None:
+    props = event.get("properties") or {}
+    info = props.get("info") or {}
+    part = props.get("part") or {}
+    if str(event.get("type", "")).startswith("session."):
+        return info.get("id") or props.get("sessionID")
+    return part.get("sessionID") or props.get("sessionID") or info.get("sessionID")
+
+
+class EventBus:
+    def __init__(self, source=None):
+        # source: async iterator factory — 테스트에서 가짜로 바꿈
+        self._source = source or opencode_service.iter_event_jsons
+        self._subs: dict[str, set[asyncio.Queue]] = {}
+        self._task: asyncio.Task | None = None
+
+    def subscribe(self, session_id: str) -> asyncio.Queue:
+        q: asyncio.Queue = asyncio.Queue()
+        self._subs.setdefault(session_id, set()).add(q)
+        self._ensure_running()
+        return q
+
+    def unsubscribe(self, session_id: str, q: asyncio.Queue) -> None:
+        subs = self._subs.get(session_id)
+        if subs:
+            subs.discard(q)
+            if not subs:
+                del self._subs[session_id]
+
+    def _ensure_running(self) -> None:
+        if self._task is None or self._task.done():
+            self._task = asyncio.get_running_loop().create_task(self._pump(), name="opencode-event-pump")
+
+    async def _pump(self) -> None:
+        try:
+            async for payload in self._source():
+                if payload is None:
+                    for subs in list(self._subs.values()):
+                        for q in list(subs):
+                            q.put_nowait(None)
+                    continue
+                try:
+                    event = json.loads(payload)
+                except ValueError:
+                    continue
+                sid = session_id_of(event)
+                if not sid:
+                    continue
+                for q in list(self._subs.get(sid, ())):
+                    q.put_nowait(event)
+        except asyncio.CancelledError:
+            raise
+        except Exception:  # noqa: BLE001
+            log.exception("이벤트 펌프 죽음 — 다음 구독 때 재시작")
+        finally:
+            self._task = None
+
+
+event_bus = EventBus()
@@ -0,0 +1,84 @@
+# Generated by Django 5.1.15 on 2026-09-16 08:40
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="ChatSession",
+            fields=[
+                (
+                    "id",
+                    models.CharField(max_length=64, primary_key=True, serialize=False),
+                ),
+                ("title", models.CharField(blank=True, max_length=200)),
+                ("title_llm", models.CharField(blank=True, max_length=200)),
+                ("is_generating", models.BooleanField(default=False)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                (
+                    "user",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="chat_sessions",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "ordering": ["-updated_at"],
+            },
+        ),
+        migrations.CreateModel(
+            name="ChatMessage",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "role",
+                    models.CharField(
+                        choices=[
+                            ("user", "user"),
+                            ("assistant", "assistant"),
+                            ("system", "system"),
+                        ],
+                        max_length=16,
+                    ),
+                ),
+                ("content", models.TextField()),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("input_tokens", models.IntegerField(blank=True, null=True)),
+                ("output_tokens", models.IntegerField(blank=True, null=True)),
+                ("cost_usd", models.FloatField(blank=True, null=True)),
+                ("elapsed_ms", models.IntegerField(blank=True, null=True)),
+                (
+                    "session",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="messages",
+                        to="chat.chatsession",
+                    ),
+                ),
+            ],
+            options={
+                "ordering": ["created_at", "id"],
+            },
+        ),
+    ]
@@ -0,0 +1,61 @@
+"""
+세션·메시지 미러 — 프론트 `features/snap/contract/types.ts` 와 1:1.
+
+대화 원본(컨텍스트)은 OpenCode 가 들고 있고, 여긴 목록·검색·usage·isGenerating 용 사본.
+OpenCode 는 페이지네이션·본문 검색·메시지별 토큰을 안 주니까 여기서 함.
+"""
+
+from django.conf import settings
+from django.db import models
+
+
+class ChatSession(models.Model):
+    id = models.CharField(primary_key=True, max_length=64)  # OpenCode session id 그대로
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="chat_sessions")
+    title = models.CharField(max_length=200, blank=True)  # 사용자가 지은 제목 (지금은 안 씀)
+    title_llm = models.CharField(max_length=200, blank=True)  # OpenCode 가 지은 제목
+    is_generating = models.BooleanField(default=False)
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        ordering = ["-updated_at"]
+
+    def as_dto(self) -> dict:
+        return {
+            "id": self.id,
+            "title": self.title or None,
+            "titleLlm": self.title_llm or None,
+            "isGenerating": self.is_generating,
+            "createdAt": self.created_at.isoformat(),
+            "updatedAt": self.updated_at.isoformat(),
+        }
+
+
+class ChatMessage(models.Model):
+    ROLES = [("user", "user"), ("assistant", "assistant"), ("system", "system")]
+
+    session = models.ForeignKey(ChatSession, on_delete=models.CASCADE, related_name="messages")
+    role = models.CharField(max_length=16, choices=ROLES)
+    content = models.TextField()  # markdown
+    created_at = models.DateTimeField(auto_now_add=True)
+    # assistant 행에만 채움
+    input_tokens = models.IntegerField(null=True, blank=True)
+    output_tokens = models.IntegerField(null=True, blank=True)
+    cost_usd = models.FloatField(null=True, blank=True)
+    elapsed_ms = models.IntegerField(null=True, blank=True)
+
+    class Meta:
+        ordering = ["created_at", "id"]
+
+    def as_dto(self) -> dict:
+        return {
+            "sessionId": self.session_id,
+            "role": self.role,
+            "content": self.content,
+            "createdAt": self.created_at.isoformat(),
+            "inputTokens": self.input_tokens,
+            "outputTokens": self.output_tokens,
+            "costUsd": self.cost_usd,
+            "elapsedMs": self.elapsed_ms,
+        }
@@ -0,0 +1,381 @@
+"""
+POST /api/v1/chat/stream — 프론트 `lib/streaming/streamLLM.ts` SSE 계약.
+
+요청 {sessionId, content, images?[{mediaType,data}], forcedSkill?, explain?}
+이벤트  token{delta} ×N → usage{used,limit,ratio,elapsed_ms} → done{}
+        중간 title{title}, 실패 error{message,code}
+
+흐름
+  1. 인증(Bearer) → 본인 세션 → 생성 중이면 409
+  2. 미러: user 메시지 저장 + is_generating=true
+  3. 이벤트 버스 구독 → OpenCode prompt_async (블록 없음)
+  4. 턴 감시 태스크(run_turn)가 /event 를 우리 이벤트로 바꿔 큐에 넣음.
+     클라이언트가 끊겨도 태스크는 idle 까지 돌고 DB 를 마무리함 → 프론트 isGenerating 폴링이 복구.
+  5. session.idle 이면 OpenCode 에서 최종 메시지·tokens·cost 가져와 저장 → usage → done
+
+순수 Django async 뷰 — DRF 는 async 스트리밍을 못 함. 반드시 ASGI(uvicorn).
+"""
+
+import asyncio
+import json
+import logging
+import time
+
+from django.conf import settings
+from django.http import StreamingHttpResponse
+from django.views.decorators.csrf import csrf_exempt
+
+from apps.accounts.authentication import user_from_token
+from asgiref.sync import sync_to_async
+from common.envelope import json_error
+from common.opencode_service import opencode_service
+
+from .events import event_bus
+from .models import ChatMessage, ChatSession
+
+log = logging.getLogger(__name__)
+
+ALLOWED_IMAGE_TYPES = {"image/png", "image/jpeg", "image/webp"}
+MAX_IMAGES = 4
+TURN_TIMEOUT_S = 600  # OpenCode 답변 상한
+# prompt 를 받았는데 이 세션 이벤트가 하나도 안 오면 OpenCode 가 조용히 실패한 것
+# (예: 다른 프로젝트의 세션 — 로그에만 "prompt_async failed" 남고 session.error 안 옴)
+FIRST_EVENT_TIMEOUT_S = 60
+KEEPALIVE_S = 15
+_END = object()  # 큐 종료 표시
+
+
+# ── SSE 프레임 ──────────────────────────────────────────────────
+def sse(event: str, data) -> bytes:
+    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode()
+
+
+def _token_from(request) -> str:
+    auth = request.headers.get("Authorization", "")
+    return auth[7:] if auth.startswith("Bearer ") else request.GET.get("token", "")
+
+
+def _parse_body(request) -> tuple[dict | None, str | None]:
+    try:
+        body = json.loads(request.body or b"{}")
+    except ValueError:
+        return None, "JSON 이 아니야"
+    if not isinstance(body, dict) or not body.get("sessionId"):
+        return None, "sessionId 가 없어"
+    content = body.get("content") or ""
+    images = body.get("images") or []
+    if not isinstance(images, list) or len(images) > MAX_IMAGES:
+        return None, f"이미지는 최대 {MAX_IMAGES}장"
+    for img in images:
+        if not isinstance(img, dict) or img.get("mediaType") not in ALLOWED_IMAGE_TYPES or not img.get("data"):
+            return None, "이미지 형식이 잘못됐어 (png/jpeg/webp data URL)"
+    if not content.strip() and not images:
+        return None, "content 나 images 중 하나는 있어야 해"
+    return {"sessionId": body["sessionId"], "content": content, "images": images, "explain": bool(body.get("explain"))}, None
+
+
+def _prompt_payload(req: dict) -> dict:
+    text = req["content"]
+    if req["explain"]:
+        text = f"[설명 모드] {text}".strip()
+    parts: list[dict] = []
+    if text.strip():
+        parts.append({"type": "text", "text": text})
+    for i, img in enumerate(req["images"]):
+        ext = img["mediaType"].split("/")[1]
+        parts.append({"type": "file", "mime": img["mediaType"], "filename": f"image-{i + 1}.{ext}", "url": img["data"]})
+    return {"agent": settings.OPENCODE_AGENT, "parts": parts}
+
+
+# ── OpenCode 이벤트 → 우리 이벤트 ────────────────────────────────
+class TurnState:
+    """한 턴 동안 delta 계산에 필요한 것."""
+
+    def __init__(self, session_id: str, user_text: str):
+        self.session_id = session_id
+        self.user_text = user_text
+        self.user_message_ids: set[str] = set()
+        self.assistant_message_ids: set[str] = set()
+        self.seen_len: dict[str, int] = {}  # partID → 지금까지 보낸 글자 수
+        self.part_types: dict[str, str] = {}  # partID → "text"/"reasoning"/… (part.updated 가 delta 보다 먼저 옴)
+        self.text_by_part: dict[str, str] = {}
+        self.part_order: list[str] = []
+        self.title: str | None = None
+
+    def accumulated(self) -> str:
+        return "".join(self.text_by_part.get(p, "") for p in self.part_order)
+
+    def handle(self, event: dict) -> list[tuple[str, dict]]:
+        """이벤트 하나 → 프론트로 보낼 (event, data) 목록. idle/error 는 호출자가 봄."""
+        out: list[tuple[str, dict]] = []
+        etype = event.get("type")
+        props = event.get("properties") or {}
+
+        if etype == "message.updated":
+            info = props.get("info") or {}
+            if info.get("role") == "user":
+                self.user_message_ids.add(info.get("id", ""))
+            elif info.get("role") == "assistant":
+                self.assistant_message_ids.add(info.get("id", ""))
+            return out
+
+        if etype == "message.part.delta":
+            # 실서버(1.18.6)의 글자 단위 스트리밍은 이 이벤트로 옴 — SDK 타입엔 없지만 실제로 174개/답변.
+            # {sessionID, messageID, partID, field:"text", delta}. 파트 타입은 안 실려서 part.updated 로 알아둔 것으로 거름.
+            if props.get("field") != "text" or not props.get("delta"):
+                return out
+            mid = props.get("messageID", "")
+            if mid in self.user_message_ids and mid not in self.assistant_message_ids:
+                return out
+            pid = props.get("partID", "")
+            if self.part_types.get(pid) != "text":
+                return out  # reasoning/tool 파트, 또는 아직 타입 모름(최종 스냅샷이 메워줌)
+            delta = props["delta"]
+            if pid not in self.text_by_part:
+                self.part_order.append(pid)
+            self.text_by_part[pid] = self.text_by_part.get(pid, "") + delta
+            self.seen_len[pid] = len(self.text_by_part[pid])
+            out.append(("token", {"delta": delta}))
+            return out
+
+        if etype == "message.part.updated":
+            part = props.get("part") or {}
+            if part.get("id"):
+                self.part_types[part["id"]] = part.get("type", "")
+            if part.get("type") != "text" or part.get("synthetic") or part.get("ignored"):
+                return out
+            mid = part.get("messageID", "")
+            # 사용자 메시지 파트(우리가 보낸 질문의 echo)는 건너뜀
+            if mid in self.user_message_ids and mid not in self.assistant_message_ids:
+                return out
+            pid = part.get("id", "")
+            text = part.get("text") or ""
+            delta = props.get("delta")
+            if pid not in self.text_by_part:
+                self.part_order.append(pid)
+            if delta:
+                self.text_by_part[pid] = self.text_by_part.get(pid, "") + delta
+                self.seen_len[pid] = len(self.text_by_part[pid])
+                out.append(("token", {"delta": delta}))
+            else:
+                seen = self.seen_len.get(pid, 0)
+                if len(text) > seen and text.startswith(self.text_by_part.get(pid, "")):
+                    out.append(("token", {"delta": text[seen:]}))
+                    self.seen_len[pid] = len(text)
+                self.text_by_part[pid] = text if len(text) >= seen else self.text_by_part.get(pid, "")
+            return out
+
+        if etype == "session.updated":
+            title = _real_title((props.get("info") or {}).get("title"))
+            if title and title != self.title:
+                self.title = title
+                out.append(("title", {"title": title}))
+            return out
+
+        return out
+
+
+def _real_title(raw) -> str | None:
+    """OpenCode 기본 제목("New session - 2026-…")은 제목이 아님 → None."""
+    t = (raw or "").strip()
+    if not t or t.startswith("New session"):
+        return None
+    return t
+
+
+def _error_message(err: dict | None) -> str:
+    if not err:
+        return "OpenCode 오류"
+    data = err.get("data") or {}
+    return data.get("message") or err.get("name") or "OpenCode 오류"
+
+
+async def _finalize(session: ChatSession, state: TurnState, started: float, *, failed: str | None = None):
+    """idle/에러/타임아웃 후 미러 DB 마무리. 실패해도 is_generating 은 반드시 풀림."""
+    content = state.accumulated()
+    usage: dict | None = None
+    title = state.title
+    try:
+        msgs = await opencode_service.list_messages_a(session.id) or []
+        assistant = [m for m in msgs if (m.get("info") or {}).get("role") == "assistant"]
+        if assistant:
+            info = assistant[-1]["info"]
+            parts = assistant[-1].get("parts") or []
+            full = "".join(p.get("text", "") for p in parts if p.get("type") == "text" and not p.get("synthetic"))
+            if full:
+                content = full
+            tokens = info.get("tokens") or {}
+            t = info.get("time") or {}
+            elapsed = (t.get("completed") or int(time.time() * 1000)) - (t.get("created") or int(started * 1000))
+            usage = {
+                "input": int(tokens.get("input") or 0),
+                "output": int(tokens.get("output") or 0) + int(tokens.get("reasoning") or 0),
+                "cost": float(info.get("cost") or 0.0),
+                "elapsed_ms": max(0, int(elapsed)),
+            }
+        if not title:
+            info = await opencode_service.get_session_a(session.id) or {}
+            title = _real_title(info.get("title"))
+    except Exception:  # noqa: BLE001
+        log.warning("턴 마무리 중 OpenCode 조회 실패 — 누적 텍스트로 저장: %s", session.id)
+
+    if content or failed is None:
+        await ChatMessage.objects.acreate(
+            session=session,
+            role="assistant",
+            content=content or (f"(오류: {failed})" if failed else ""),
+            input_tokens=usage["input"] if usage else None,
+            output_tokens=usage["output"] if usage else None,
+            cost_usd=usage["cost"] if usage else None,
+            elapsed_ms=usage["elapsed_ms"] if usage else None,
+        )
+    session.is_generating = False
+    if title:
+        session.title_llm = title[:200]
+    await session.asave(update_fields=["is_generating", "title_llm", "updated_at"])
+    return usage, title
+
+
+TITLE_WAIT_S = 20  # OpenCode 제목 생성은 별도 LLM 호출이라 idle 뒤에 오기도 함
+
+
+async def _wait_late_title(session: ChatSession, q: asyncio.Queue) -> None:
+    """idle 뒤 늦게 오는 session.updated 제목을 미러에만 반영 (목록 재조회 때 보임)."""
+    deadline = time.time() + TITLE_WAIT_S
+    while (remaining := deadline - time.time()) > 0:
+        try:
+            event = await asyncio.wait_for(q.get(), timeout=remaining)
+        except asyncio.TimeoutError:
+            return
+        if not event or event.get("type") != "session.updated":
+            continue
+        title = _real_title(((event.get("properties") or {}).get("info") or {}).get("title"))
+        if title:
+            session.title_llm = title[:200]
+            await session.asave(update_fields=["title_llm", "updated_at"])
+            return
+
+
+async def run_turn(session: ChatSession, req: dict, out: asyncio.Queue) -> None:
+    """한 턴 감시 — 클라이언트와 무관하게 idle 까지 돌고 DB 마무리. 결과는 out 큐로."""
+    state = TurnState(session.id, req["content"])
+    started = time.time()
+    q = event_bus.subscribe(session.id)
+    try:
+        await opencode_service.prompt_async(session.id, _prompt_payload(req))
+    except Exception as e:  # noqa: BLE001
+        event_bus.unsubscribe(session.id, q)
+        log.warning("prompt_async 실패: %s", e)
+        await _finalize(session, state, started, failed=str(e))
+        out.put_nowait(("error", {"message": "OpenCode 에 질문을 못 보냈어", "code": "LLM_ERROR"}))
+        out.put_nowait(_END)
+        return
+
+    try:
+        deadline = started + TURN_TIMEOUT_S
+        got_any = False
+        while True:
+            now = time.time()
+            remaining = deadline - now
+            silent_too_long = not got_any and now - started > FIRST_EVENT_TIMEOUT_S
+            if remaining <= 0 or silent_too_long:
+                try:
+                    await opencode_service.abort_session_a(session.id)
+                except Exception:  # noqa: BLE001
+                    pass
+                reason = "no-events" if silent_too_long else "timeout"
+                await _finalize(session, state, started, failed=reason)
+                msg = "OpenCode 가 응답을 시작하지 않았어" if silent_too_long else "답변 시간이 너무 길어서 중단했어"
+                out.put_nowait(("error", {"message": msg, "code": "LLM_ERROR"}))
+                break
+            try:
+                event = await asyncio.wait_for(q.get(), timeout=min(KEEPALIVE_S, remaining))
+            except asyncio.TimeoutError:
+                out.put_nowait(None)  # keepalive
+                continue
+            if event is None:
+                out.put_nowait(None)
+                continue
+            got_any = True
+
+            etype = event.get("type")
+            props = event.get("properties") or {}
+            if etype == "session.error" or (
+                etype == "message.updated"
+                and (props.get("info") or {}).get("role") == "assistant"
+                and (props.get("info") or {}).get("error")
+            ):
+                err = props.get("error") or (props.get("info") or {}).get("error")
+                msg = _error_message(err)
+                await _finalize(session, state, started, failed=msg)
+                code = "LLM_ABORTED" if (err or {}).get("name") == "MessageAbortedError" else "LLM_ERROR"
+                out.put_nowait(("error", {"message": msg, "code": code}))
+                break
+
+            for ev in state.handle(event):
+                out.put_nowait(ev)
+
+            if etype == "session.idle":
+                usage, title = await _finalize(session, state, started)
+                if title and title != state.title:
+                    out.put_nowait(("title", {"title": title}))
+                if usage:
+                    used = usage["input"] + usage["output"]
+                    limit = settings.CONTEXT_LIMIT_TOKENS
+                    out.put_nowait(
+                        ("usage", {"used": used, "limit": limit, "ratio": round(used / limit, 4) if limit else 0, "elapsed_ms": usage["elapsed_ms"]})
+                    )
+                out.put_nowait(("done", {}))
+                out.put_nowait(_END)  # 클라이언트는 여기서 끝. 제목은 뒤에서 조용히 기다림
+                if not title:
+                    await _wait_late_title(session, q)
+                break
+    except Exception as e:  # noqa: BLE001
+        log.exception("턴 감시 실패: %s", session.id)
+        await _finalize(session, state, started, failed=str(e))
+        out.put_nowait(("error", {"message": "서버 오류로 답변을 못 받았어", "code": "LLM_ERROR"}))
+    finally:
+        event_bus.unsubscribe(session.id, q)
+        out.put_nowait(_END)
+
+
+# ── 뷰 ──────────────────────────────────────────────────────────
+@csrf_exempt
+async def stream_chat(request):
+    if request.method != "POST":
+        return json_error(405, "METHOD_NOT_ALLOWED", "POST 만 받아")
+    user = await sync_to_async(user_from_token)(_token_from(request))
+    if user is None:
+        return json_error(401, "UNAUTHORIZED", "로그인이 필요해")
+    req, err = _parse_body(request)
+    if err:
+        return json_error(400, "VALIDATION_ERROR", err)
+
+    try:
+        session = await ChatSession.objects.aget(id=req["sessionId"], user=user)
+    except ChatSession.DoesNotExist:
+        return json_error(404, "NOT_FOUND", "세션이 없어")
+    if session.is_generating:
+        return json_error(409, "CHAT_GENERATION_IN_PROGRESS", "아직 답변 생성 중이야")
+
+    session.is_generating = True
+    await session.asave(update_fields=["is_generating", "updated_at"])
+    await ChatMessage.objects.acreate(session=session, role="user", content=req["content"])
+
+    out: asyncio.Queue = asyncio.Queue()
+    asyncio.get_running_loop().create_task(run_turn(session, req, out), name=f"turn-{session.id}")
+
+    async def gen():
+        while True:
+            item = await out.get()
+            if item is _END:
+                break
+            if item is None:
+                yield b": keepalive\n\n"
+                continue
+            yield sse(*item)
+
+    resp = StreamingHttpResponse(gen(), content_type="text/event-stream")
+    resp["Cache-Control"] = "no-cache"
+    resp["X-Accel-Buffering"] = "no"
+    return resp
@@ -0,0 +1,12 @@
+from django.urls import path
+
+from .stream import stream_chat
+from .views import CancelView, SearchView, SessionListView, SessionMessagesView
+
+urlpatterns = [
+    path("stream", stream_chat),
+    path("sessions", SessionListView.as_view()),
+    path("sessions/search", SearchView.as_view()),  # <str:session_id> 보다 먼저
+    path("sessions/<str:session_id>/messages", SessionMessagesView.as_view()),
+    path("sessions/<str:session_id>/cancel", CancelView.as_view()),
+]
@@ -0,0 +1,103 @@
+"""
+세션 API — 프론트 `features/snap/api/snap.api.ts` 계약. 전부 본인 세션만.
+
+  GET  /api/v1/chat/sessions?page&limit          → SnapSession[] + meta
+  POST /api/v1/chat/sessions                     → SnapSession (OpenCode 세션 생성 + 미러)
+  GET  /api/v1/chat/sessions/search?query&page&limit → SnapMessage[] + meta (본문 icontains)
+  GET  /api/v1/chat/sessions/{id}/messages       → SnapSessionDetail
+  POST /api/v1/chat/sessions/{id}/cancel         → null (OpenCode abort + isGenerating 해제)
+스트림(POST /chat/stream)은 stream.py — async 라 DRF 밖.
+"""
+
+import logging
+
+import httpx
+from django.shortcuts import get_object_or_404
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from common.envelope import CodedError, paged
+from common.opencode_service import opencode_service
+
+from .models import ChatMessage, ChatSession
+
+log = logging.getLogger(__name__)
+
+MAX_LIMIT = 100
+
+
+def _page_args(request) -> tuple[int, int]:
+    try:
+        page = max(1, int(request.query_params.get("page", 1)))
+        limit = min(MAX_LIMIT, max(1, int(request.query_params.get("limit", 20))))
+    except ValueError:
+        raise CodedError(400, "VALIDATION_ERROR", "page/limit 은 정수여야 해")
+    return page, limit
+
+
+def _slice(qs, page: int, limit: int):
+    total = qs.count()
+    items = list(qs[(page - 1) * limit : page * limit])
+    return items, total
+
+
+def _own_session(request, session_id: str) -> ChatSession:
+    return get_object_or_404(ChatSession, id=session_id, user=request.user)
+
+
+def _upstream(e: Exception) -> CodedError:
+    if isinstance(e, httpx.ConnectError):
+        return CodedError(503, "UPSTREAM_UNAVAILABLE", "OpenCode 서버에 연결할 수 없어")
+    if isinstance(e, httpx.HTTPStatusError):
+        return CodedError(502, "UPSTREAM_ERROR", f"OpenCode 오류: {e.response.text[:300]}")
+    log.exception("OpenCode 호출 실패")
+    return CodedError(502, "UPSTREAM_ERROR", f"OpenCode 중계 오류: {e}")
+
+
+class SessionListView(APIView):
+    def get(self, request):
+        page, limit = _page_args(request)
+        items, total = _slice(ChatSession.objects.filter(user=request.user), page, limit)
+        return Response(paged([s.as_dto() for s in items], page, limit, total))
+
+    def post(self, request):
+        try:
+            created = opencode_service.create_session()
+        except Exception as e:  # noqa: BLE001
+            raise _upstream(e)
+        session = ChatSession.objects.create(id=created["id"], user=request.user)
+        return Response(session.as_dto(), status=201)
+
+
+class SearchView(APIView):
+    def get(self, request):
+        query = (request.query_params.get("query") or "").strip()
+        if not query:
+            raise CodedError(400, "VALIDATION_ERROR", "query 가 비었어")
+        page, limit = _page_args(request)
+        qs = (
+            ChatMessage.objects.filter(session__user=request.user, content__icontains=query)
+            .select_related("session")
+            .order_by("-created_at", "-id")
+        )
+        items, total = _slice(qs, page, limit)
+        return Response(paged([m.as_dto() for m in items], page, limit, total))
+
+
+class SessionMessagesView(APIView):
+    def get(self, request, session_id: str):
+        session = _own_session(request, session_id)
+        return Response({**session.as_dto(), "messages": [m.as_dto() for m in session.messages.all()]})
+
+
+class CancelView(APIView):
+    def post(self, request, session_id: str):
+        session = _own_session(request, session_id)
+        try:
+            opencode_service.abort_session(session.id)
+        except Exception:  # noqa: BLE001 — best-effort. 이미 끝났으면 실패해도 됨
+            log.warning("abort 실패 (이미 끝났을 수 있음): %s", session.id)
+        if session.is_generating:
+            session.is_generating = False
+            session.save(update_fields=["is_generating", "updated_at"])
+        return Response(None)
@@ -0,0 +1,154 @@
+"""
+응답 envelope — 프론트 `types/api.ts` 의 CommonResponse 와 1:1.
+
+{
+  success, statusCode, code, message, data, counts, errors, timestamp, meta
+}
+
+- DRF 뷰: 그냥 `Response(data)` 하면 EnvelopeRenderer 가 감쌈.
+  목록은 `Response(paged(items, page, limit, total))` — data=items, meta 채워짐.
+- DRF 예외: envelope_exception_handler 가 success:false 로 감쌈. code 는 CodedError 로 지정.
+- 순수 Django async 뷰(스트림): `json_error()` 로 직접 만듦.
+"""
+
+from datetime import datetime, timezone
+
+from django.http import JsonResponse
+from rest_framework.exceptions import APIException
+from rest_framework.renderers import JSONRenderer
+
+_LIST_MARK = "__paged__"
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def envelope(
+    *,
+    success: bool,
+    status_code: int,
+    data=None,
+    code: str | None = None,
+    message: str | None = None,
+    errors: list[str] | None = None,
+    counts: int | None = None,
+    meta: dict | None = None,
+) -> dict:
+    return {
+        "success": success,
+        "statusCode": status_code,
+        "code": code,
+        "message": message,
+        "data": data,
+        "counts": counts,
+        "errors": errors or [],
+        "timestamp": _now(),
+        "meta": meta,
+    }
+
+
+def paged(items: list, page: int, limit: int, total: int) -> dict:
+    """목록 응답용. Renderer 가 data=items, meta=페이지 정보로 풀어줌."""
+    pages = (total + limit - 1) // limit if limit > 0 else 0
+    return {
+        _LIST_MARK: True,
+        "items": items,
+        "counts": total,
+        "meta": {
+            "currentPage": page,
+            "pageSize": limit,
+            "totalItems": total,
+            "totalPages": pages,
+            "hasNextPage": page < pages,
+            "hasPreviousPage": page > 1,
+        },
+    }
+
+
+def _is_wrapped(data) -> bool:
+    return isinstance(data, dict) and "success" in data and "statusCode" in data
+
+
+class EnvelopeRenderer(JSONRenderer):
+    def render(self, data, accepted_media_type=None, renderer_context=None):
+        response = (renderer_context or {}).get("response")
+        status_code = getattr(response, "status_code", 200)
+        if _is_wrapped(data):
+            body = data
+        elif isinstance(data, dict) and data.get(_LIST_MARK):
+            body = envelope(
+                success=True,
+                status_code=status_code,
+                data=data["items"],
+                counts=data["counts"],
+                meta=data["meta"],
+            )
+        else:
+            body = envelope(success=True, status_code=status_code, data=data)
+        return super().render(body, accepted_media_type, renderer_context)
+
+
+class CodedError(APIException):
+    """상태코드 + 프론트가 분기하는 code 문자열을 같이 던짐."""
+
+    def __init__(self, status_code: int, code: str, message: str):
+        super().__init__(detail=message)
+        self.status_code = status_code
+        self.code_name = code
+
+
+_DEFAULT_CODES = {
+    400: "VALIDATION_ERROR",
+    401: "UNAUTHORIZED",
+    403: "FORBIDDEN",
+    404: "NOT_FOUND",
+    405: "METHOD_NOT_ALLOWED",
+    409: "CONFLICT",
+    501: "NOT_IMPLEMENTED",
+    502: "UPSTREAM_ERROR",
+    503: "UPSTREAM_UNAVAILABLE",
+}
+
+
+def _flatten(data) -> str:
+    """serializer 검증 오류(dict/list)를 한 줄로."""
+    if isinstance(data, str):
+        return data
+    if isinstance(data, list):
+        return " ".join(_flatten(d) for d in data)
+    if isinstance(data, dict):
+        parts = []
+        for k, v in data.items():
+            msg = _flatten(v)
+            parts.append(msg if k in ("detail", "non_field_errors") else f"{k}: {msg}")
+        return " ".join(parts)
+    return str(data)
+
+
+def envelope_exception_handler(exc, context):
+    # rest_framework.views 는 settings 의 renderer(=이 모듈)를 import 하므로 여기서 늦게 import (순환 방지)
+    from rest_framework.views import exception_handler
+
+    response = exception_handler(exc, context)
+    if response is None:
+        return None
+    code = getattr(exc, "code_name", None) or _DEFAULT_CODES.get(response.status_code, "ERROR")
+    message = _flatten(response.data)
+    response.data = envelope(
+        success=False,
+        status_code=response.status_code,
+        code=code,
+        message=message,
+        errors=[message],
+    )
+    return response
+
+
+def json_error(status_code: int, code: str, message: str) -> JsonResponse:
+    """DRF 안 거치는 async 뷰용."""
+    return JsonResponse(
+        envelope(success=False, status_code=status_code, code=code, message=message, errors=[message]),
+        status=status_code,
+        json_dumps_params={"ensure_ascii": False},
+    )
@@ -0,0 +1,167 @@
+"""
+===================================================================
+모듈: common/opencode_service.py
+설명: OpenCode 서버(opencode serve)와의 HTTP 통신
+
+    - JSON API (세션/메시지/프로바이더/에이전트) → 동기 httpx (DRF 뷰용)
+    - SSE 이벤트 스트림 → 비동기 (SSE 뷰·파일 워처용)
+===================================================================
+"""
+
+import asyncio
+import logging
+from typing import Any, AsyncIterator
+
+import httpx
+from django.conf import settings
+
+log = logging.getLogger(__name__)
+
+# 메시지 전송은 LLM 응답 완료까지 걸리므로 타임아웃을 길게 잡는다 (10분)
+LONG_TIMEOUT = httpx.Timeout(600.0, connect=10.0)
+DEFAULT_TIMEOUT = httpx.Timeout(30.0, connect=10.0)
+
+
+class OpencodeService:
+    """OpenCode 서버 프록시."""
+
+    def __init__(self) -> None:
+        self.base_url = settings.OPENCODE_BASE_URL
+        # 모든 요청에 ?directory= 를 붙여 CodeAssist workspace 프로젝트로 고정 (1.18 멀티 프로젝트 서버)
+        self.directory = getattr(settings, "OPENCODE_DIRECTORY", "") or None
+
+    def _params(self) -> dict | None:
+        return {"directory": self.directory} if self.directory else None
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        json: Any | None = None,
+        timeout: httpx.Timeout = DEFAULT_TIMEOUT,
+    ) -> Any:
+        with httpx.Client(base_url=self.base_url, timeout=timeout) as client:
+            response = client.request(method, path, json=json, params=self._params())
+            response.raise_for_status()
+            return response.json() if response.content else None
+
+    # ── 세션
+    def list_sessions(self) -> Any:
+        return self._request("GET", "/session")
+
+    def create_session(self) -> Any:
+        return self._request("POST", "/session", json={})
+
+    def delete_session(self, session_id: str) -> Any:
+        return self._request("DELETE", f"/session/{session_id}")
+
+    def rename_session(self, session_id: str, title: str) -> Any:
+        return self._request("PATCH", f"/session/{session_id}", json={"title": title})
+
+    # ── 메시지
+    def list_messages(self, session_id: str) -> Any:
+        return self._request("GET", f"/session/{session_id}/message")
+
+    def abort_session(self, session_id: str) -> Any:
+        log.info("응답 중단: session=%s", session_id)
+        return self._request("POST", f"/session/{session_id}/abort")
+
+    def send_message(self, session_id: str, payload: dict) -> Any:
+        log.info(
+            "메시지 전송: session=%s provider=%s model=%s",
+            session_id, payload.get("providerID"), payload.get("modelID"),
+        )
+        # OpenCode 1.18+ 는 최상위 providerID/modelID 를 무시한다(세션 기본 모델 사용).
+        # model:{providerID,modelID} 오브젝트 형식으로 변환해서 보내야 선택이 먹는다.
+        provider_id = payload.pop("providerID", None)
+        model_id = payload.pop("modelID", None)
+        if provider_id and model_id:
+            payload["model"] = {"providerID": provider_id, "modelID": model_id}
+        return self._request(
+            "POST", f"/session/{session_id}/message", json=payload, timeout=LONG_TIMEOUT
+        )
+
+    # ── 설정
+    def get_providers(self) -> Any:
+        return self._request("GET", "/config/providers")
+
+    def list_agents(self) -> Any:
+        return self._request("GET", "/agent")
+
+    # ── 비동기 버전 (스트림 어댑터용 — 이벤트 루프 안에서 블록 없이)
+    async def _arequest(self, method: str, path: str, json: Any | None = None) -> Any:
+        async with httpx.AsyncClient(base_url=self.base_url, timeout=DEFAULT_TIMEOUT) as client:
+            response = await client.request(method, path, json=json, params=self._params())
+            response.raise_for_status()
+            return response.json() if response.content else None
+
+    async def prompt_async(self, session_id: str, payload: dict) -> None:
+        """POST /session/{id}/prompt_async — 204 즉시 반환, 답변은 /event 로 흘러옴."""
+        await self._arequest("POST", f"/session/{session_id}/prompt_async", json=payload)
+
+    async def get_session_a(self, session_id: str) -> Any:
+        return await self._arequest("GET", f"/session/{session_id}")
+
+    async def list_messages_a(self, session_id: str) -> Any:
+        return await self._arequest("GET", f"/session/{session_id}/message")
+
+    async def abort_session_a(self, session_id: str) -> Any:
+        return await self._arequest("POST", f"/session/{session_id}/abort")
+
+    # ── SSE 이벤트 스트림 (비동기)
+    async def iter_event_jsons(self) -> AsyncIterator[str | None]:
+        """SSE 프레임을 파싱해 이벤트 JSON 문자열 단위로 순회한다.
+
+        자가 치유: 업스트림이 60초간 조용하면 연결을 재수립한다 —
+        서버 재시작 등으로 반쯤 죽은 소켓이 남아도 스스로 복구된다.
+        무활동/재연결 시점에는 None(하트비트)을 내보낸다."""
+        stream_timeout = httpx.Timeout(None, connect=10.0)
+        while True:
+            try:
+                async with httpx.AsyncClient(
+                    base_url=self.base_url, timeout=stream_timeout
+                ) as client:
+                    async with client.stream("GET", "/event", params=self._params()) as response:
+                        response.raise_for_status()
+                        data_lines: list[str] = []
+                        lines = response.aiter_lines()
+                        while True:
+                            try:
+                                line = await asyncio.wait_for(lines.__anext__(), timeout=60.0)
+                            except asyncio.TimeoutError:
+                                yield None  # 60초 무활동 — 하트비트 후 재연결
+                                break
+                            except StopAsyncIteration:
+                                break
+                            if line == "":
+                                if data_lines:
+                                    yield "\n".join(data_lines)
+                                    data_lines = []
+                            elif line.startswith("data:"):
+                                data_lines.append(line[5:].lstrip())
+                            # 그 외 필드(id:, event:, 코멘트)는 무시
+            except Exception:
+                log.warning("OpenCode SSE 업스트림 끊김 — 2초 후 재연결")
+            yield None
+            await asyncio.sleep(2)
+
+
+opencode_service = OpencodeService()
+
+
+def upstream_error(e: Exception):
+    """OpenCode 서버 에러 → DRF APIException (FE 는 detail 필드를 읽는다)."""
+    from rest_framework.exceptions import APIException
+
+    if isinstance(e, httpx.ConnectError):
+        exc = APIException("OpenCode 서버에 연결할 수 없습니다. opencode serve 실행 여부를 확인하세요.")
+        exc.status_code = 503
+        return exc
+    if isinstance(e, httpx.HTTPStatusError):
+        exc = APIException(f"OpenCode 서버 오류: {e.response.text[:500]}")
+        exc.status_code = e.response.status_code
+        return exc
+    log.exception("OpenCode 중계 오류")
+    exc = APIException(f"OpenCode 중계 오류: {e}")
+    exc.status_code = 502
+    return exc
@@ -0,0 +1,11 @@
+"""ASGI 진입점 — SSE 때문에 반드시 uvicorn 으로.
+    uvicorn config.asgi:application --host 0.0.0.0 --port 8001
+"""
+
+import os
+
+from django.core.asgi import get_asgi_application
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
+
+application = get_asgi_application()
@@ -0,0 +1,208 @@
+"""
+서버 설정표 — 켤 때 한 번 읽음.
+
+값 우선순위: 환경변수 > 5_django_backend/.env > 여기 기본값.
+DB 는 DB_HOST 있으면 PostgreSQL, 없으면 data/app.db (SQLite).
+
+ABAP_OPENCODE web/BE/code/config/settings.py 에서 CodeAssist 에 필요한 것만 남김
+(산출물·ITO 게이트웨이·bare 배포 관련은 뺌).
+"""
+
+import os
+import secrets
+from datetime import timedelta
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+
+
+def _load_dotenv(path: Path) -> None:
+    """이름=값 줄을 환경변수로 올림. 이미 있는 환경변수가 우선."""
+    if not path.exists():
+        return
+    for line in path.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        k, v = line.split("=", 1)
+        os.environ.setdefault(k.strip(), v.strip().strip('"\x27'))
+
+
+_load_dotenv(BASE_DIR / ".env")
+
+DATA_DIR = BASE_DIR / "data"
+DATA_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def _env(name: str, default: str = "") -> str:
+    return os.environ.get(name, default).strip()
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    v = _env(name)
+    return default if not v else v.lower() in ("1", "true", "yes", "on")
+
+
+APP_NAME = _env("APP_NAME", "CodeAssist Backend")
+APP_VERSION = "0.1.0"
+
+
+def _secret_key() -> str:
+    """토큰 서명 키. 바뀌면 전원 재로그인이라 파일에 저장해 재사용."""
+    env = _env("AUTH_SECRET")
+    if env:
+        return env
+    p = DATA_DIR / ".auth_secret"
+    if p.exists():
+        return p.read_text(encoding="utf-8").strip() or secrets.token_hex(32)
+    generated = secrets.token_hex(32)
+    p.write_text(generated, encoding="utf-8")
+    return generated
+
+
+SECRET_KEY = _secret_key()
+DEBUG = _env_bool("DEBUG", True)
+ALLOWED_HOSTS = ["*"]
+
+# ── 바깥 것들 ─────────────────────────────────────────────────────
+# CodeAssist 전용 OpenCode 인스턴스 (opencode/ workspace 로 띄운 것)
+OPENCODE_BASE_URL = _env("OPENCODE_BASE_URL", "http://localhost:4096").rstrip("/")
+# OpenCode 1.18 은 서버 하나가 여러 프로젝트(directory)를 다루고 요청마다 ?directory= 로 고름.
+# 안 넘기면 마지막에 열린 프로젝트로 가서 세션이 엉뚱한 workspace 에 생김 → 항상 넘긴다.
+# 기본은 이 repo 의 opencode/ workspace. 서버를 다른 경로에서 띄우면 그 절대경로로.
+OPENCODE_DIRECTORY = _env("OPENCODE_DIRECTORY") or str((BASE_DIR / "opencode").resolve())
+# OpenCode 에 넘길 에이전트 이름 (opencode/.opencode/agent/<이름>.md)
+OPENCODE_AGENT = _env("OPENCODE_AGENT", "codeassist")
+# 세션 컨텍스트 하드 한도 — usage.limit 로 프론트 게이지에 감
+CONTEXT_LIMIT_TOKENS = int(_env("CONTEXT_LIMIT_TOKENS", "128000") or "128000")
+
+INSTALLED_APPS = [
+    "django.contrib.admin",
+    "django.contrib.auth",
+    "django.contrib.contenttypes",
+    "django.contrib.sessions",
+    "django.contrib.messages",
+    "django.contrib.staticfiles",
+    "rest_framework",
+    "corsheaders",
+    "apps.accounts",
+    "apps.chat",
+]
+
+MIDDLEWARE = [
+    "corsheaders.middleware.CorsMiddleware",
+    "django.middleware.security.SecurityMiddleware",
+    "django.contrib.sessions.middleware.SessionMiddleware",
+    "django.middleware.common.CommonMiddleware",
+    "django.middleware.csrf.CsrfViewMiddleware",
+    "django.contrib.auth.middleware.AuthenticationMiddleware",
+    "django.contrib.messages.middleware.MessageMiddleware",
+]
+
+ROOT_URLCONF = "config.urls"
+ASGI_APPLICATION = "config.asgi.application"
+
+TEMPLATES = [
+    {
+        "BACKEND": "django.template.backends.django.DjangoTemplates",
+        "DIRS": [],
+        "APP_DIRS": True,
+        "OPTIONS": {
+            "context_processors": [
+                "django.template.context_processors.request",
+                "django.contrib.auth.context_processors.auth",
+                "django.contrib.messages.context_processors.messages",
+            ],
+        },
+    },
+]
+
+# ── DB ────────────────────────────────────────────────────────────
+_pg_host = _env("DB_HOST")
+if _pg_host:
+    _schema = _env("DB_SCHEMA", "codeassist") or "codeassist"
+    DATABASES = {
+        "default": {
+            "ENGINE": "django.db.backends.postgresql",
+            "HOST": _pg_host,
+            "PORT": _env("DB_PORT", "5432") or "5432",
+            "USER": _env("DB_USER"),
+            "PASSWORD": _env("DB_PASSWORD"),
+            "NAME": _env("DB_NAME"),
+            # ASGI 스레드마다 연결 눌러앉는 것 방지 — 요청 끝나면 바로 반납
+            "CONN_MAX_AGE": 0,
+            "OPTIONS": {
+                "sslmode": "require",
+                "connect_timeout": 10,
+                "options": f"-c search_path={_schema}",
+            },
+        }
+    }
+else:
+    DATABASES = {
+        "default": {
+            "ENGINE": "django.db.backends.sqlite3",
+            "NAME": DATA_DIR / "app.db",
+        }
+    }
+
+DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
+
+# ── 인증 ─────────────────────────────────────────────────────────
+AUTH_USER_MODEL = "accounts.User"
+AUTH_PASSWORD_VALIDATORS = []
+
+REST_FRAMEWORK = {
+    "DEFAULT_AUTHENTICATION_CLASSES": [
+        "apps.accounts.authentication.BearerOrQueryJWTAuthentication",
+    ],
+    "DEFAULT_PERMISSION_CLASSES": ["rest_framework.permissions.IsAuthenticated"],
+    # 모든 응답을 프론트 CommonResponse envelope 로 감쌈
+    "DEFAULT_RENDERER_CLASSES": ["common.envelope.EnvelopeRenderer"],
+    "DEFAULT_PARSER_CLASSES": ["rest_framework.parsers.JSONParser"],
+    "EXCEPTION_HANDLER": "common.envelope.envelope_exception_handler",
+    "UNAUTHENTICATED_USER": None,
+}
+
+# 프론트가 401 → /auth/refresh 자동 재시도를 하므로 access 는 짧게, refresh 는 길게.
+SIMPLE_JWT = {
+    "ACCESS_TOKEN_LIFETIME": timedelta(minutes=60),
+    "REFRESH_TOKEN_LIFETIME": timedelta(days=14),
+    "AUTH_HEADER_TYPES": ("Bearer",),
+    "USER_ID_FIELD": "id",
+    "USER_ID_CLAIM": "uid",
+    "SIGNING_KEY": SECRET_KEY,
+}
+
+# ── CORS ─────────────────────────────────────────────────────────
+# dev: Vite proxy 로 same-origin 이라 안 탐. Tauri 빌드는 tauri://localhost 에서 오므로 여기 필요.
+CORS_ALLOWED_ORIGINS = [
+    o.strip()
+    for o in _env(
+        "CORS_ORIGINS", "http://localhost:15173,http://tauri.localhost,tauri://localhost"
+    ).split(",")
+    if o.strip()
+]
+CORS_ALLOW_CREDENTIALS = True
+
+# 채팅 이미지 4장 × 5MiB + 여유 (프론트 제한과 동일 계열)
+DATA_UPLOAD_MAX_MEMORY_SIZE = 32 * 1024 * 1024
+
+LANGUAGE_CODE = "ko-kr"
+TIME_ZONE = "Asia/Seoul"
+USE_I18N = True
+USE_TZ = True
+STATIC_URL = "static/"
+STATIC_ROOT = BASE_DIR / "staticfiles"
+
+LOGGING = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {"std": {"format": "%(asctime)s %(levelname)s %(name)s: %(message)s"}},
+    "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "std"}},
+    "root": {"handlers": ["console"], "level": "INFO"},
+    "loggers": {
+        "django.request": {"level": "WARNING"},
+        "httpx": {"level": "WARNING"},
+    },
+}
@@ -0,0 +1,24 @@
+"""최상위 URL — 프론트 VITE_API_BASE_URL(/api/v1) 아래에 매닮."""
+
+from django.conf import settings
+from django.contrib import admin
+from django.http import JsonResponse
+from django.urls import include, path
+
+from apps.accounts.views import MeView
+from common.envelope import envelope
+
+
+def health(_request):
+    return JsonResponse(
+        envelope(success=True, status_code=200, data={"app": settings.APP_NAME, "version": settings.APP_VERSION})
+    )
+
+
+urlpatterns = [
+    path("admin/", admin.site.urls),
+    path("api/v1/health", health),
+    path("api/v1/auth/", include("apps.accounts.urls")),
+    path("api/v1/users/me", MeView.as_view()),
+    path("api/v1/chat/", include("apps.chat.urls")),
+]
@@ -0,0 +1,9 @@
+# docs-lib — 5_django_backend 가 쓰는 라이브러리 참고 문서
+
+구현 전 여기부터 `Grep` 으로 필요한 부분만 꺼내 본다. 기억·추측으로 API 쓰지 말 것.
+
+| 파일 | 뭐냐 | 출처 |
+|---|---|---|
+| `opencode-sdk-1.18.6-types.d.ts` | OpenCode 서버(`opencode serve`) HTTP API 전체 타입 — 세션·메시지·파트·이벤트·엔드포인트 url. `url: "/session…"` 로 grep 하면 엔드포인트, `export type Event…` 로 grep 하면 SSE 이벤트 | `@opencode-ai/sdk@1.18.6` `dist/gen/types.gen.d.ts` (ABAP_OPENCODE `.opencode/node_modules`) |
+
+갱신: OpenCode 버전 올리면 같은 경로에서 다시 복사하고 파일명 버전 바꿈.
@@ -0,0 +1,16 @@
+#!/usr/bin/env python
+"""Django 명령 진입점 — `python manage.py migrate` 등."""
+
+import os
+import sys
+
+
+def main() -> None:
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
+    from django.core.management import execute_from_command_line
+
+    execute_from_command_line(sys.argv)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,12 @@
+---
+description: CodeAssist 즉답형 ABAP 코드 도우미 — 질문에 바로 코드·답을 준다
+mode: primary
+temperature: 0.2
+tools:
+  write: false
+  edit: false
+  bash: false
+---
+
+너는 CodeAssist 의 답변 에이전트다. AGENTS.md 의 "답하는 방식"을 그대로 따른다.
+절차 질문 없이 바로 답하고, 코드는 완결된 조각으로 코드펜스에 넣는다.
@@ -0,0 +1,19 @@
+# CodeAssist — OpenCode 작업 지침
+
+너는 SAP ABAP 개발자가 SAP GUI·Eclipse(ADT) 옆에 띄워놓고 쓰는 **즉답형 코드 도우미**다.
+사용자는 질문하고 바로 코드나 답을 받아서 원래 창에 붙여넣는다. 흐름을 끊지 마라.
+
+## 답하는 방식
+
+- 한국어. SAP 기술 용어(테이블명·필드명·T-code·키워드)는 원문 그대로.
+- **바로 답한다.** 요구사항 정리 질문, 단계별 확인, "~해도 될까요" 금지. 정보가 부족하면 가장 흔한 경우로 가정하고 답한 뒤 가정을 한 줄 덧붙인다.
+- 코드는 항상 코드펜스 + 언어 태그(```abap, ```sql, ```cds). 붙여넣어 바로 되는 완결된 조각으로.
+- 기본은 짧게 — 코드 + 필요한 설명 두세 줄. 사용자 메시지에 `[설명 모드]` 표시가 있으면 배경·원리·대안까지 자세히.
+- 이미지(화면 캡처)가 오면 그 안의 코드·에러·화면을 읽고 답한다. 이미지에 대해 묘사하지 말고 바로 문제 해결로.
+- 모르는 SAP 오브젝트는 지어내지 않는다. 도구(sap-icf MCP)가 연결돼 있으면 조회하고, 없으면 모른다고 하고 확인 방법(T-code, 테이블)을 알려준다.
+
+## 하지 않는 것
+
+- 파일을 만들거나 수정하지 않는다. 이 workspace 는 도구 실행용이 아니라 대화용이다.
+- 설계서·문서 작성 절차를 시작하지 않는다 (그건 다른 제품이 한다).
+- 지침 문서·도구 이름을 사용자에게 언급하지 않는다.
@@ -0,0 +1,25 @@
+# CodeAssist 전용 OpenCode workspace
+
+Django(`5_django_backend`)가 부르는 AI 엔진. ABAP_OPENCODE 의 OpenCode 와는 **별도 인스턴스** — 거긴 AGENTS.md 가 "설계서 브레인스토밍" 규칙이라 즉답 도우미랑 충돌해서 따로 띄움.
+
+## 띄우기 (개발 PC)
+
+```bash
+cd 5_django_backend/opencode
+export OPENROUTER_API_KEY=sk-or-...
+npx -y opencode-ai@1.18.6 serve --hostname 127.0.0.1 --port 4096
+```
+
+- 버전은 1.18.6 고정 — `docs-lib/opencode-sdk-1.18.6-types.d.ts` 가 이 버전 API 기준.
+- **gotcha 1 — `OPENCODE_CONFIG_DIR`:** 이 환경변수가 잡혀 있으면(예: orca 같은 툴이 셸에 박아둠) workspace 의 `opencode.json`·`.opencode/agent` 를 무시하고 그쪽 걸 읽어서 `Agent not found: codeassist` 가 남. 띄우기 전에 `env -u OPENCODE_CONFIG_DIR -u ORCA_OPENCODE_CONFIG_DIR opencode serve …` 로 풀거나 `.zshrc` 에서 빼.
+- **gotcha 2 — 프로젝트(directory):** 1.18 서버는 한 프로세스가 여러 프로젝트를 다루고 요청마다 `?directory=` 로 고름. Django 가 `OPENCODE_DIRECTORY`(기본: 이 폴더 절대경로) 를 모든 요청에 붙이니, 서버를 다른 경로에서 띄우면 `.env` 의 `OPENCODE_DIRECTORY` 를 그 경로로 맞춰. 안 맞으면 세션이 딴 프로젝트에 생기고 prompt 가 로그에만 실패를 남기고 조용히 죽음(60초 뒤 error 로 끝냄).
+- 확인: `curl http://127.0.0.1:4096/session` → `[]`.
+- SAP 조회(MCP)는 `opencode.json` 의 `sap-icf.enabled` 를 true 로 바꾸고 `SAP_MCP_URL`/`SAP_MCP_KEY` 환경변수 주면 붙음 (ABAP_OPENCODE 의 :3200 서버 재사용).
+
+## 파일
+
+| 파일 | 뭐냐 |
+|---|---|
+| `AGENTS.md` | 모든 세션에 붙는 지침 — 즉답, 코드펜스, 절차 질문 금지 |
+| `opencode.json` | 모델(OpenRouter GLM 5.2)·`question` 도구 끔·MCP |
+| `.opencode/agent/codeassist.md` | Django 가 `agent: "codeassist"` 로 지정하는 에이전트. 파일 쓰기·bash 끔 |
@@ -0,0 +1,21 @@
+{
+  "$schema": "https://opencode.ai/config.json",
+  "provider": {
+    "openrouter": {
+      "options": { "apiKey": "{env:OPENROUTER_API_KEY}" },
+      "models": {
+        "z-ai/glm-5.2": { "name": "GLM 5.2 (OpenRouter)" }
+      }
+    }
+  },
+  "model": "openrouter/z-ai/glm-5.2",
+  "tools": { "question": false },
+  "mcp": {
+    "sap-icf": {
+      "type": "remote",
+      "url": "{env:SAP_MCP_URL}",
+      "headers": { "Authorization": "Bearer {env:SAP_MCP_KEY}" },
+      "enabled": false
+    }
+  }
+}
@@ -0,0 +1,5 @@
+[pytest]
+DJANGO_SETTINGS_MODULE = config.settings
+python_files = test_*.py
+testpaths = tests
+asyncio_mode = auto
@@ -0,0 +1,12 @@
+django==5.1.*
+djangorestframework==3.17.*
+djangorestframework-simplejwt==5.5.*
+django-cors-headers==4.9.*
+uvicorn[standard]==0.52.*
+httpx==0.28.1
+psycopg2-binary==2.9.*
+
+# 테스트
+pytest
+pytest-django
+pytest-asyncio
@@ -0,0 +1,26 @@
+import pytest
+from rest_framework.test import APIClient
+
+
+@pytest.fixture
+def api():
+    return APIClient()
+
+
+@pytest.fixture
+def user(db):
+    from apps.accounts.models import User
+
+    return User.objects.create_user(
+        username="u@x.com", email="u@x.com", password="pw1234", user_name="유저"
+    )
+
+
+@pytest.fixture
+def auth_api(api, user):
+    """로그인된 클라이언트 + 토큰 응답."""
+    res = api.post("/api/v1/auth/login", {"email": "u@x.com", "password": "pw1234"}, format="json")
+    tokens = res.json()["data"]
+    api.credentials(HTTP_AUTHORIZATION=f"Bearer {tokens['token']}")
+    api.tokens = tokens
+    return api
@@ -0,0 +1,87 @@
+"""인증 API — 프론트 auth.api.ts / types/api.ts 계약."""
+
+TOKEN_KEYS = {
+    "token",
+    "tokenExpirationTime",
+    "refreshToken",
+    "refreshTokenExpirationTime",
+    "tokenType",
+    "user",
+}
+USER_KEYS = {"id", "email", "userName", "role", "employeeId", "department", "authProvider"}
+
+
+def test_login_returns_token_response(api, user):
+    res = api.post("/api/v1/auth/login", {"email": "U@X.com", "password": "pw1234"}, format="json")
+    assert res.status_code == 200
+    data = res.json()["data"]
+    assert set(data) == TOKEN_KEYS
+    assert data["tokenType"] == "bearer"
+    assert set(data["user"]) == USER_KEYS
+    assert data["user"]["email"] == "u@x.com"
+    assert data["user"]["role"] == "USER"
+    assert data["user"]["authProvider"] == "local"
+    assert data["tokenExpirationTime"] < data["refreshTokenExpirationTime"]
+
+
+def test_login_wrong_password_401_with_code(api, user):
+    res = api.post("/api/v1/auth/login", {"email": "u@x.com", "password": "nope"}, format="json")
+    assert res.status_code == 401
+    assert res.json()["code"] == "INVALID_CREDENTIALS"
+
+
+def test_me_with_bearer(auth_api):
+    res = auth_api.get("/api/v1/users/me")
+    assert res.status_code == 200
+    data = res.json()["data"]
+    assert data["email"] == "u@x.com"
+    assert data["isActive"] is True
+    assert "createdAt" in data
+
+
+def test_me_with_query_token(api, auth_api):
+    # EventSource 등 헤더 못 붙이는 곳용
+    plain = api.__class__()
+    res = plain.get(f"/api/v1/users/me?token={auth_api.tokens['token']}")
+    assert res.status_code == 200
+
+
+def test_refresh_issues_new_access(auth_api):
+    old = auth_api.tokens
+    res = auth_api.post("/api/v1/auth/refresh", {"refreshToken": old["refreshToken"]}, format="json")
+    assert res.status_code == 200
+    data = res.json()["data"]
+    assert set(data) == TOKEN_KEYS
+    assert data["refreshToken"] == old["refreshToken"]  # 회전 안 함
+    assert data["user"]["email"] == "u@x.com"
+
+
+def test_refresh_missing_token(api, db):
+    res = api.post("/api/v1/auth/refresh", {}, format="json")
+    assert res.status_code == 401
+    assert res.json()["code"] == "REFRESH_TOKEN_MISSING"
+
+
+def test_refresh_bad_token(api, db):
+    res = api.post("/api/v1/auth/refresh", {"refreshToken": "garbage"}, format="json")
+    assert res.status_code == 401
+    assert res.json()["code"] == "REFRESH_TOKEN_INVALID"
+
+
+def test_logout_is_public_and_null(api, db):
+    res = api.post("/api/v1/auth/logout", format="json")
+    assert res.status_code == 200
+    assert res.json()["data"] is None
+
+
+def test_entra_config_is_501(api, db):
+    res = api.get("/api/v1/auth/entra/config")
+    assert res.status_code == 501
+    assert res.json()["code"] == "NOT_IMPLEMENTED"
+
+
+def test_seed_users_exist_after_migrate(db):
+    from apps.accounts.models import User
+
+    assert User.objects.filter(email="admin@codeassist.local", is_superuser=True).exists()
+    assert User.objects.filter(email="guest@codeassist.local").exists()
@@ -0,0 +1,29 @@
+"""응답 envelope 가 프론트 CommonResponse 모양인지."""
+
+ENVELOPE_KEYS = {"success", "statusCode", "code", "message", "data", "counts", "errors", "timestamp", "meta"}
+
+
+def test_health_is_wrapped(api, db):
+    body = api.get("/api/v1/health").json()
+    assert set(body) == ENVELOPE_KEYS
+    assert body["success"] is True and body["statusCode"] == 200
+    assert body["data"]["app"]
+
+
+def test_error_is_wrapped_with_code(api, db):
+    res = api.get("/api/v1/users/me")
+    body = res.json()
+    assert res.status_code == 401
+    assert set(body) == ENVELOPE_KEYS
+    assert body["success"] is False
+    assert body["code"] == "UNAUTHORIZED"
+    assert body["data"] is None
+    assert body["errors"] and body["message"]
+
+
+def test_validation_error_is_400(api, db):
+    res = api.post("/api/v1/auth/login", {"email": "not-an-email"}, format="json")
+    body = res.json()
+    assert res.status_code == 400
+    assert body["code"] == "VALIDATION_ERROR"
+    assert "password" in body["message"]
@@ -0,0 +1,123 @@
+"""세션 미러 API — snap.api.ts 계약. OpenCode 는 가짜로."""
+
+import httpx
+import pytest
+
+from apps.chat.models import ChatMessage, ChatSession
+
+SESSION_KEYS = {"id", "title", "titleLlm", "isGenerating", "createdAt", "updatedAt"}
+MESSAGE_KEYS = {"sessionId", "role", "content", "createdAt", "inputTokens", "outputTokens", "costUsd", "elapsedMs"}
+
+
+class FakeOpencode:
+    def __init__(self):
+        self.n = 0
+        self.aborted = []
+        self.down = False
+
+    def create_session(self):
+        if self.down:
+            raise httpx.ConnectError("refused")
+        self.n += 1
+        return {"id": f"ses_{self.n}", "title": ""}
+
+    def abort_session(self, sid):
+        self.aborted.append(sid)
+        return True
+
+
+@pytest.fixture
+def oc(monkeypatch):
+    fake = FakeOpencode()
+    monkeypatch.setattr("apps.chat.views.opencode_service", fake)
+    return fake
+
+
+def _mk(user, sid="ses_a", **kw):
+    return ChatSession.objects.create(id=sid, user=user, **kw)
+
+
+def test_create_session_mirrors_opencode(auth_api, oc):
+    res = auth_api.post("/api/v1/chat/sessions", {}, format="json")
+    assert res.status_code == 201
+    data = res.json()["data"]
+    assert set(data) == SESSION_KEYS
+    assert data["id"] == "ses_1" and data["isGenerating"] is False and data["titleLlm"] is None
+    assert ChatSession.objects.get(id="ses_1").user.email == "u@x.com"
+
+
+def test_create_session_when_opencode_down_503(auth_api, oc):
+    oc.down = True
+    res = auth_api.post("/api/v1/chat/sessions", {}, format="json")
+    assert res.status_code == 503
+    assert res.json()["code"] == "UPSTREAM_UNAVAILABLE"
+
+
+def test_list_paginated_newest_first_own_only(auth_api, user, django_user_model):
+    other = django_user_model.objects.create_user(username="o@x.com", email="o@x.com", password="x")
+    for i in range(5):
+        _mk(user, f"ses_{i}")
+    _mk(other, "ses_other")
+    res = auth_api.get("/api/v1/chat/sessions?page=1&limit=3")
+    body = res.json()
+    assert [s["id"] for s in body["data"]] == ["ses_4", "ses_3", "ses_2"]
+    assert body["counts"] == 5
+    assert body["meta"] == {
+        "currentPage": 1,
+        "pageSize": 3,
+        "totalItems": 5,
+        "totalPages": 2,
+        "hasNextPage": True,
+        "hasPreviousPage": False,
+    }
+    res2 = auth_api.get("/api/v1/chat/sessions?page=2&limit=3")
+    assert [s["id"] for s in res2.json()["data"]] == ["ses_1", "ses_0"]
+
+
+def test_messages_detail(auth_api, user):
+    s = _mk(user, title_llm="첫 질문")
+    ChatMessage.objects.create(session=s, role="user", content="hi")
+    ChatMessage.objects.create(
+        session=s, role="assistant", content="hello", input_tokens=10, output_tokens=5, cost_usd=0.001, elapsed_ms=1200
+    )
+    res = auth_api.get("/api/v1/chat/sessions/ses_a/messages")
+    data = res.json()["data"]
+    assert set(data) == SESSION_KEYS | {"messages"}
+    assert data["titleLlm"] == "첫 질문"
+    assert [m["role"] for m in data["messages"]] == ["user", "assistant"]
+    assert set(data["messages"][1]) == MESSAGE_KEYS
+    assert data["messages"][1]["inputTokens"] == 10
+    assert data["messages"][0]["inputTokens"] is None
+
+
+def test_messages_of_other_user_404(auth_api, django_user_model):
+    other = django_user_model.objects.create_user(username="o@x.com", email="o@x.com", password="x")
+    _mk(other, "ses_other")
+    assert auth_api.get("/api/v1/chat/sessions/ses_other/messages").status_code == 404
+
+
+def test_search_messages_icontains_own_only(auth_api, user, django_user_model):
+    other = django_user_model.objects.create_user(username="o@x.com", email="o@x.com", password="x")
+    mine = _mk(user, "ses_m")
+    theirs = _mk(other, "ses_t")
+    ChatMessage.objects.create(session=mine, role="user", content="SELECT * FROM mara")
+    ChatMessage.objects.create(session=mine, role="assistant", content="MARA 는 자재 마스터")
+    ChatMessage.objects.create(session=mine, role="user", content="다른 얘기")
+    ChatMessage.objects.create(session=theirs, role="user", content="mara 남의 것")
+    res = auth_api.get("/api/v1/chat/sessions/search?query=mara&page=1&limit=20")
+    body = res.json()
+    assert body["counts"] == 2
+    assert {m["content"] for m in body["data"]} == {"SELECT * FROM mara", "MARA 는 자재 마스터"}
+    assert all(m["sessionId"] == "ses_m" for m in body["data"])
+
+
+def test_search_empty_query_400(auth_api):
+    assert auth_api.get("/api/v1/chat/sessions/search?query=%20").status_code == 400
+
+
+def test_cancel_aborts_and_clears_generating(auth_api, user, oc):
+    _mk(user, is_generating=True)
+    res = auth_api.post("/api/v1/chat/sessions/ses_a/cancel", {}, format="json")
+    assert res.status_code == 200 and res.json()["data"] is None
+    assert oc.aborted == ["ses_a"]
+    assert ChatSession.objects.get(id="ses_a").is_generating is False
@@ -0,0 +1,359 @@
+"""POST /chat/stream — OpenCode 이벤트 → 프론트 SSE 계약(token/title/usage/done/error)."""
+
+import asyncio
+import json
+
+import pytest
+from django.test import AsyncClient
+
+from apps.chat import events as events_mod
+from apps.chat import stream as stream_mod
+from apps.chat.models import ChatMessage, ChatSession
+from apps.chat.stream import TurnState
+
+
+# ── TurnState 단위 ───────────────────────────────────────────────
+def _part_updated(mid, pid, text, delta=None, ptype="text", sid="s1"):
+    props = {"part": {"id": pid, "messageID": mid, "sessionID": sid, "type": ptype, "text": text}}
+    if delta is not None:
+        props["delta"] = delta
+    return {"type": "message.part.updated", "properties": props}
+
+
+def _msg_updated(mid, role, sid="s1", error=None):
+    info = {"id": mid, "sessionID": sid, "role": role}
+    if error:
+        info["error"] = error
+    return {"type": "message.updated", "properties": {"info": info}}
+
+
+def test_turnstate_delta_path():
+    st = TurnState("s1", "q")
+    st.handle(_msg_updated("mu", "user"))
+    st.handle(_part_updated("mu", "pu", "q"))  # 사용자 echo → 무시
+    st.handle(_msg_updated("ma", "assistant"))
+    assert st.handle(_part_updated("ma", "pa", "안", delta="안")) == [("token", {"delta": "안"})]
+    assert st.handle(_part_updated("ma", "pa", "안녕", delta="녕")) == [("token", {"delta": "녕"})]
+    assert st.accumulated() == "안녕"
+
+
+def test_turnstate_snapshot_path_without_delta():
+    st = TurnState("s1", "q")
+    st.handle(_msg_updated("ma", "assistant"))
+    assert st.handle(_part_updated("ma", "pa", "ab")) == [("token", {"delta": "ab"})]
+    assert st.handle(_part_updated("ma", "pa", "abcd")) == [("token", {"delta": "cd"})]
+    assert st.handle(_part_updated("ma", "pa", "abcd")) == []  # 같은 스냅샷 재전송
+    assert st.accumulated() == "abcd"
+
+
+def test_turnstate_ignores_reasoning_and_emits_title():
+    st = TurnState("s1", "q")
+    st.handle(_msg_updated("ma", "assistant"))
+    assert st.handle(_part_updated("ma", "pr", "생각중", ptype="reasoning")) == []
+    ev = {"type": "session.updated", "properties": {"info": {"id": "s1", "title": "MARA 조회"}}}
+    assert st.handle(ev) == [("title", {"title": "MARA 조회"})]
+    assert st.handle(ev) == []  # 같은 제목 반복 안 보냄
+    placeholder = {"type": "session.updated", "properties": {"info": {"id": "s1", "title": "New session - 2026-09-16T08:54:31.272Z"}}}
+    assert st.handle(placeholder) == []  # OpenCode 기본 제목은 무시
+
+
+def _delta(mid, pid, delta, sid="s1", field="text"):
+    return {"type": "message.part.delta", "properties": {"sessionID": sid, "messageID": mid, "partID": pid, "field": field, "delta": delta}}
+
+
+def test_turnstate_real_server_shape_delta_events():
+    """실서버 1.18.6 순서: part.updated(타입, 빈 텍스트) → message.part.delta ×N → part.updated(최종 스냅샷)."""
+    st = TurnState("s1", "q")
+    st.handle(_msg_updated("mu", "user"))
+    st.handle(_part_updated("mu", "pu", "q"))
+    st.handle(_msg_updated("ma", "assistant"))
+    assert st.handle(_part_updated("ma", "pr", "", ptype="reasoning")) == []
+    assert st.handle(_delta("ma", "pr", "생각")) == []  # reasoning delta 는 안 보냄
+    assert st.handle(_part_updated("ma", "pt", "")) == []  # 빈 스냅샷
+    assert st.handle(_delta("ma", "pt", "SELECT")) == [("token", {"delta": "SELECT"})]
+    assert st.handle(_delta("ma", "pt", " SINGLE")) == [("token", {"delta": " SINGLE"})]
+    assert st.handle(_part_updated("ma", "pt", "SELECT SINGLE")) == []  # 최종 스냅샷 = 누적 → 중복 없음
+    assert st.handle(_part_updated("ma", "pt", "SELECT SINGLE mtart")) == [("token", {"delta": " mtart"})]  # 못 받은 꼬리 보충
+    assert st.handle(_delta("ma", "unknown", "x")) == []  # 타입 모르는 파트
+    assert st.accumulated() == "SELECT SINGLE mtart"
+
+
+# ── 통합: 뷰 + 이벤트 버스 + 가짜 OpenCode ───────────────────────
+class FakeOpencode:
+    def __init__(self, script):
+        self.script = script  # prompt 뒤에 흘려보낼 이벤트 dict 목록
+        self.prompted = asyncio.Event()
+        self.payloads = []
+        self.aborted = []
+        self.fail_prompt = False
+        self.messages = []
+        self.session_info = {"id": "s1", "title": ""}
+
+    async def prompt_async(self, sid, payload):
+        self.payloads.append((sid, payload))
+        if self.fail_prompt:
+            raise RuntimeError("down")
+        self.prompted.set()
+
+    async def list_messages_a(self, sid):
+        return self.messages
+
+    async def get_session_a(self, sid):
+        return self.session_info
+
+    async def abort_session_a(self, sid):
+        self.aborted.append(sid)
+
+    async def events(self):
+        await self.prompted.wait()
+        for ev in self.script:
+            yield json.dumps(ev)
+        await asyncio.sleep(3600)  # 진짜처럼 계속 열려 있음
+
+
+@pytest.fixture
+def fake(monkeypatch):
+    def _install(script):
+        oc = FakeOpencode(script)
+        monkeypatch.setattr(stream_mod, "opencode_service", oc)
+        monkeypatch.setattr(events_mod, "event_bus", events_mod.EventBus(source=oc.events))
+        monkeypatch.setattr(stream_mod, "event_bus", events_mod.event_bus)
+        return oc
+
+    return _install
+
+
+async def _login(client: AsyncClient, email, pw):
+    res = await client.post("/api/v1/auth/login", {"email": email, "password": pw}, content_type="application/json")
+    return res.json()["data"]["token"]
+
+
+async def _collect(resp):
+    """SSE 프레임 → [(event, data)]"""
+    raw = b""
+    async for chunk in resp.streaming_content:
+        raw += chunk
+    out = []
+    for frame in raw.decode().split("\n\n"):
+        if not frame.startswith("event:"):
+            continue
+        lines = frame.split("\n")
+        ev = lines[0][7:]
+        data = json.loads(lines[1][6:])
+        out.append((ev, data))
+    return out
+
+
+ASSISTANT_DONE = [
+    _msg_updated("mu", "user"),
+    _part_updated("mu", "pu", "MARA 뭐야"),
+    _msg_updated("ma", "assistant"),
+    _part_updated("ma", "pa", "자재", delta="자재"),
+    _part_updated("ma", "pa", "자재 마스터", delta=" 마스터"),
+    {"type": "session.updated", "properties": {"info": {"id": "s1", "title": "MARA 설명"}}},
+    {"type": "session.idle", "properties": {"sessionID": "s1"}},
+]
+
+FINAL_MESSAGES = [
+    {"info": {"id": "mu", "role": "user"}, "parts": [{"type": "text", "text": "MARA 뭐야"}]},
+    {
+        "info": {
+            "id": "ma",
+            "role": "assistant",
+            "tokens": {"input": 120, "output": 30, "reasoning": 0, "cache": {"read": 0, "write": 0}},
+            "cost": 0.0012,
+            "time": {"created": 1000, "completed": 3500},
+        },
+        "parts": [{"type": "text", "text": "자재 마스터"}],
+    },
+]
+
+
+@pytest.mark.django_db(transaction=True)
+async def test_stream_happy_path(fake, django_user_model):
+    user = await django_user_model.objects.acreate(username="u@x.com", email="u@x.com", password="x")
+    user.set_password("pw1234")
+    await user.asave()
+    await ChatSession.objects.acreate(id="s1", user=user)
+    oc = fake(ASSISTANT_DONE)
+    oc.messages = FINAL_MESSAGES
+
+    client = AsyncClient()
+    token = await _login(client, "u@x.com", "pw1234")
+    resp = await client.post(
+        "/api/v1/chat/stream",
+        {"sessionId": "s1", "content": "MARA 뭐야", "explain": True},
+        content_type="application/json",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert resp.status_code == 200
+    assert resp["Content-Type"].startswith("text/event-stream")
+    frames = await _collect(resp)
+
+    assert [f[0] for f in frames] == ["token", "token", "title", "usage", "done"]
+    assert "".join(d["delta"] for e, d in frames if e == "token") == "자재 마스터"
+    assert frames[2][1] == {"title": "MARA 설명"}
+    assert frames[3][1] == {"used": 150, "limit": 128000, "ratio": round(150 / 128000, 4), "elapsed_ms": 2500}
+
+    # OpenCode 에 보낸 payload — agent + explain 표시
+    sid, payload = oc.payloads[0]
+    assert sid == "s1" and payload["agent"] == "codeassist"
+    assert payload["parts"] == [{"type": "text", "text": "[설명 모드] MARA 뭐야"}]
+
+    # 미러 DB
+    session = await ChatSession.objects.aget(id="s1")
+    assert session.is_generating is False and session.title_llm == "MARA 설명"
+    msgs = [m async for m in ChatMessage.objects.filter(session_id="s1").order_by("id")]
+    assert [(m.role, m.content) for m in msgs] == [("user", "MARA 뭐야"), ("assistant", "자재 마스터")]
+    assert (msgs[1].input_tokens, msgs[1].output_tokens, msgs[1].cost_usd, msgs[1].elapsed_ms) == (120, 30, 0.0012, 2500)
+
+
+@pytest.mark.django_db(transaction=True)
+async def test_stream_images_become_file_parts(fake, django_user_model):
+    user = await django_user_model.objects.acreate(username="u@x.com", email="u@x.com", password="x")
+    user.set_password("pw1234")
+    await user.asave()
+    await ChatSession.objects.acreate(id="s1", user=user)
+    oc = fake([_msg_updated("ma", "assistant"), {"type": "session.idle", "properties": {"sessionID": "s1"}}])
+
+    client = AsyncClient()
+    token = await _login(client, "u@x.com", "pw1234")
+    resp = await client.post(
+        "/api/v1/chat/stream",
+        {"sessionId": "s1", "content": "", "images": [{"mediaType": "image/png", "data": "data:image/png;base64,AA=="}]},
+        content_type="application/json",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    frames = await _collect(resp)
+    assert frames[-1] == ("done", {})
+    _, payload = oc.payloads[0]
+    assert payload["parts"] == [
+        {"type": "file", "mime": "image/png", "filename": "image-1.png", "url": "data:image/png;base64,AA=="}
+    ]
+
+
+@pytest.mark.django_db(transaction=True)
+async def test_stream_session_error_emits_error_and_clears_generating(fake, django_user_model):
+    user = await django_user_model.objects.acreate(username="u@x.com", email="u@x.com", password="x")
+    user.set_password("pw1234")
+    await user.asave()
+    await ChatSession.objects.acreate(id="s1", user=user)
+    fake(
+        [
+            _msg_updated("ma", "assistant"),
+            _part_updated("ma", "pa", "절반", delta="절반"),
+            {"type": "session.error", "properties": {"sessionID": "s1", "error": {"name": "ApiError", "data": {"message": "rate limited"}}}},
+        ]
+    )
+    client = AsyncClient()
+    token = await _login(client, "u@x.com", "pw1234")
+    resp = await client.post(
+        "/api/v1/chat/stream",
+        {"sessionId": "s1", "content": "q"},
+        content_type="application/json",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    frames = await _collect(resp)
+    assert frames == [("token", {"delta": "절반"}), ("error", {"message": "rate limited", "code": "LLM_ERROR"})]
+    session = await ChatSession.objects.aget(id="s1")
+    assert session.is_generating is False
+    last = await ChatMessage.objects.filter(session_id="s1", role="assistant").alast()
+    assert last.content == "절반"  # 부분 답변 보존
+
+
+@pytest.mark.django_db(transaction=True)
+async def test_stream_409_when_generating_and_401_without_token(fake, django_user_model):
+    user = await django_user_model.objects.acreate(username="u@x.com", email="u@x.com", password="x")
+    user.set_password("pw1234")
+    await user.asave()
+    await ChatSession.objects.acreate(id="s1", user=user, is_generating=True)
+    fake([])
+    client = AsyncClient()
+    res = await client.post("/api/v1/chat/stream", {"sessionId": "s1", "content": "q"}, content_type="application/json")
+    assert res.status_code == 401 and res.json()["code"] == "UNAUTHORIZED"
+    token = await _login(client, "u@x.com", "pw1234")
+    res = await client.post(
+        "/api/v1/chat/stream",
+        {"sessionId": "s1", "content": "q"},
+        content_type="application/json",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert res.status_code == 409 and res.json()["code"] == "CHAT_GENERATION_IN_PROGRESS"
+
+
+@pytest.mark.django_db(transaction=True)
+async def test_stream_validation(fake, django_user_model):
+    user = await django_user_model.objects.acreate(username="u@x.com", email="u@x.com", password="x")
+    user.set_password("pw1234")
+    await user.asave()
+    await ChatSession.objects.acreate(id="s1", user=user)
+    fake([])
+    client = AsyncClient()
+    token = await _login(client, "u@x.com", "pw1234")
+    hdr = {"Authorization": f"Bearer {token}"}
+    res = await client.post("/api/v1/chat/stream", {"sessionId": "s1", "content": "  "}, content_type="application/json", headers=hdr)
+    assert res.status_code == 400
+    res = await client.post(
+        "/api/v1/chat/stream",
+        {"sessionId": "s1", "content": "x", "images": [{"mediaType": "image/gif", "data": "d"}]},
+        content_type="application/json",
+        headers=hdr,
+    )
+    assert res.status_code == 400
+    res = await client.post("/api/v1/chat/stream", {"sessionId": "nope", "content": "x"}, content_type="application/json", headers=hdr)
+    assert res.status_code == 404
+
+
+@pytest.mark.django_db(transaction=True)
+async def test_stream_no_events_at_all_fails_fast(fake, django_user_model, monkeypatch):
+    """OpenCode 가 prompt 는 받았는데(204) 이벤트가 하나도 안 오면 → 첫 이벤트 타임아웃으로 error."""
+    monkeypatch.setattr(stream_mod, "FIRST_EVENT_TIMEOUT_S", 0.2)
+    monkeypatch.setattr(stream_mod, "KEEPALIVE_S", 0.05)
+    user = await django_user_model.objects.acreate(username="u@x.com", email="u@x.com", password="x")
+    user.set_password("pw1234")
+    await user.asave()
+    await ChatSession.objects.acreate(id="s1", user=user)
+    oc = fake([])  # 이벤트 없음
+    client = AsyncClient()
+    token = await _login(client, "u@x.com", "pw1234")
+    resp = await client.post(
+        "/api/v1/chat/stream",
+        {"sessionId": "s1", "content": "q"},
+        content_type="application/json",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    frames = await _collect(resp)
+    assert frames[-1][0] == "error" and frames[-1][1]["code"] == "LLM_ERROR"
+    assert oc.aborted == ["s1"]
+    session = await ChatSession.objects.aget(id="s1")
+    assert session.is_generating is False
+
+
+@pytest.mark.django_db(transaction=True)
+async def test_stream_title_arriving_after_idle_is_mirrored(fake, django_user_model, monkeypatch):
+    monkeypatch.setattr(stream_mod, "TITLE_WAIT_S", 2)
+    user = await django_user_model.objects.acreate(username="u@x.com", email="u@x.com", password="x")
+    user.set_password("pw1234")
+    await user.asave()
+    await ChatSession.objects.acreate(id="s1", user=user)
+    fake(
+        [
+            _msg_updated("ma", "assistant"),
+            {"type": "session.updated", "properties": {"info": {"id": "s1", "title": "New session - 2026"}}},
+            {"type": "session.idle", "properties": {"sessionID": "s1"}},
+            {"type": "session.updated", "properties": {"info": {"id": "s1", "title": "늦게 온 제목"}}},
+        ]
+    )
+    client = AsyncClient()
+    token = await _login(client, "u@x.com", "pw1234")
+    resp = await client.post(
+        "/api/v1/chat/stream",
+        {"sessionId": "s1", "content": "q"},
+        content_type="application/json",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    frames = await _collect(resp)
+    assert frames[-1] == ("done", {})  # 제목 기다리느라 done 이 늦어지면 안 됨
+    await asyncio.sleep(0.3)
+    session = await ChatSession.objects.aget(id="s1")
+    assert session.title_llm == "늦게 온 제목"