feat(project): 3단계 구조 백엔드 기반 — 세션 삭제자·하위업무·멀티호밍

프로젝트→세션→업무 3단계(아사나식) 전환의 1단계(백엔드). "세션"은 새 개념이
아니라 기존 "단계"(project_stages, 칸반 컬럼)를 그대로 재사용한다.

- project_stages 에 created_by 추가 — 세션 삭제도 "만든 사람만"(_can_delete
  재사용) 규칙 적용. api_delete_stage 를 _require_manage 단독 체크에서
  생성자 체크로 변경.
- 프로젝트 생성을 관리자 전용 → 로그인 사용자 전체로 개방
  (api_create_project: _require_admin → _require_user). 홈 화면의 "새
  프로젝트"/수정 모달·버튼도 is_admin 게이트를 걷어냄(생성자가 자동으로
  owner 가 되어 _can_manage 통과 — 기존 로직 재사용). 멤버 배정 API는 계속
  관리자 전용이라, 비관리자가 새 프로젝트 모달을 열면 멤버 선택란은 안내
  문구만 보여주고 시도하지 않는다(불필요한 403 방지).
- 하위 업무 — 새 테이블 없이 tasks.parent_task_id 재사용(부모 삭제 시
  CASCADE). list_tasks 는 기본적으로 최상위만 반환(보드/캘린더/리스트에
  하위업무가 카드로 안 뜨게) + comment_count 와 같은 방식으로 subtask_total/
  subtask_done 집계 추가. create_task 에 parent_task_id 지원(하위업무는
  부모와 같은 프로젝트, 세션 없음).
- 멀티호밍(task_project_links, 아사나 기능) — 업무 하나가 여러 프로젝트/세션에
  동시에 속할 수 있다. list_tasks(project_id=X, 비재귀)가 자동으로 그
  프로젝트에 연결된 업무까지 포함하도록 확장(다른 호출 경로는 안 건드림 —
  홈/내업무 전체보기에서 중복 노출 방지). 신규 API: GET /api/tasks/{id}(하위
  업무·연결 포함 상세), POST/DELETE /api/tasks/{id}/links.
- 설명란 이미지 삽입 준비: store.sanitize_description_html — 허용 태그만
  남기고 스크립트/이벤트속성/위험 URL 스킴 제거(표준 html.parser, 외부
  라이브러리 없음). api_create_task/api_update_task 저장 직전 통과.

⚠️ 배포 전 필수: scripts/sql/project_db_003_sections_subtasks_links.sql 를
project_db 에 적용해야 한다(project_stages.created_by, tasks.parent_task_id,
task_project_links 없이는 이 커밋의 쿼리가 전부 실패한다). 그 안의 기존
설명 텍스트→HTML 이스케이프 UPDATE 문은 1회만 실행할 것.

프론트엔드(세션 드래그 재정렬, 사이드바 프로젝트별 보기, 하위업무/멀티호밍
UI, 설명란 이미지 붙여넣기 편집기)는 다음 커밋에서 이어간다.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

app/modules/project/db.py

@@ -125,9 +125,9 @@ class ProjectStore:
                     for i, (stage_name, is_done) in enumerate(store.DEFAULT_STAGES):
                         conn.execute(
                             "INSERT INTO project_stages "
-                            "(project_id, name, sort_order, is_done_stage) "
+                            "(project_id, name, sort_order, is_done_stage, created_by) "
-                            "VALUES (%s,%s,%s,%s)",
+                            "VALUES (%s,%s,%s,%s,%s)",
-                            (pid, stage_name, i, is_done),
+                            (pid, stage_name, i, is_done, store.norm_str(created_by, lower=True)),
                         )
                 self._log(conn, project_id=pid, actor=created_by,
                           action=store.ACTION_CREATED, detail=nm)
@@ -238,11 +238,11 @@ class ProjectStore:
         return [self._serialize(r) for r in rows]
 
     def add_stage(
-        self, *, project_id: int, name: str, is_done_stage: bool = False
+        self, *, project_id: int, name: str, is_done_stage: bool = False, created_by: str = ""
     ) -> dict[str, Any]:
         nm = store.norm_str(name)
         if not nm:
-            raise ValueError("단계 이름은 필수입니다.")
+            raise ValueError("세션 이름은 필수입니다.")
         with self._pool.connection() as conn:
             nxt = conn.execute(
                 "SELECT COALESCE(MAX(sort_order), -1) + 1 AS n "
@@ -250,12 +250,20 @@ class ProjectStore:
                 (project_id,),
             ).fetchone()
             row = conn.execute(
-                "INSERT INTO project_stages (project_id, name, sort_order, is_done_stage) "
+                "INSERT INTO project_stages (project_id, name, sort_order, is_done_stage, created_by) "
-                "VALUES (%s,%s,%s,%s) RETURNING *",
+                "VALUES (%s,%s,%s,%s,%s) RETURNING *",
-                (project_id, nm, int(nxt["n"]), bool(is_done_stage)),
+                (project_id, nm, int(nxt["n"]), bool(is_done_stage),
+                 store.norm_str(created_by, lower=True)),
             ).fetchone()
         return self._serialize(row)
 
+    def get_stage(self, *, stage_id: int) -> dict[str, Any] | None:
+        with self._pool.connection() as conn:
+            row = conn.execute(
+                "SELECT * FROM project_stages WHERE id = %s", (stage_id,)
+            ).fetchone()
+        return self._serialize(row) if row else None
+
     def delete_stage(self, *, stage_id: int) -> None:
         with self._pool.connection() as conn:
             cur = conn.execute("DELETE FROM project_stages WHERE id = %s", (stage_id,))
@@ -265,14 +273,47 @@ class ProjectStore:
     # ════════════════════════════════════════════════════════════
     # 업무 (tasks)
     # ════════════════════════════════════════════════════════════
+    # 업무 목록에 함께 붙이는 하위업무 집계(댓글수와 같은 방식). 여러 쿼리에서
+    # 재사용하므로 상수로 뽑아둔다.
+    _SUBTASK_COUNTS_SQL = (
+        "(SELECT COUNT(*) FROM tasks st WHERE st.parent_task_id = t.id) AS subtask_total, "
+        "(SELECT COUNT(*) FROM tasks st WHERE st.parent_task_id = t.id "
+        "        AND st.completed_at IS NOT NULL) AS subtask_done"
+    )
+
     def list_tasks(
         self,
         *,
         project_id: int | None = None,
         assignee_email: str | None = None,
         include_subprojects: bool = False,
+        parent_task_id: int | None = None,
+        top_level_only: bool = True,
+        include_links: bool = True,
         limit: int = 1000,
     ) -> list[dict[str, Any]]:
+        """업무 목록.
+
+        기본은 최상위 업무만(`parent_task_id IS NULL`) — 하위 업무는 부모 안에서만
+        보인다(아사나와 동일, 보드/캘린더/리스트 카드로 따로 뜨지 않는다).
+        `parent_task_id` 를 주면 그 부모의 하위 업무만 받는다.
+
+        `project_id` 하나만 지정하고(`include_subprojects=False`, 최상위 조회일
+        때) `include_links=True`(기본값)이면, 그 프로젝트에 멀티호밍
+        (`task_project_links`)으로 연결된 다른 프로젝트의 업무도 함께 반환한다.
+        프로젝트 상세 화면이 '이 프로젝트만' 보여줄 때 쓰는 경로가 이 조합이다.
+        다른 조합(홈 전체보기·내 업무 등, `project_id=None`)은 연결 업무를 섞지
+        않는다 — 같은 업무가 여러 곳에서 중복으로 집계되는 것을 막기 위함이다.
+        """
+        link_mode = (
+            include_links and project_id is not None
+            and not include_subprojects and parent_task_id is None
+        )
+        if link_mode:
+            return self._list_tasks_for_project_with_links(
+                project_id=project_id, assignee_email=assignee_email, limit=limit,
+            )
+
         clauses: list[str] = []
         params: list[Any] = []
         if project_id is not None:
@@ -292,13 +333,19 @@ class ProjectStore:
         if assignee_email:
             clauses.append("t.assignee_email = %s")
             params.append(store.norm_str(assignee_email, lower=True))
+        if parent_task_id is not None:
+            clauses.append("t.parent_task_id = %s")
+            params.append(parent_task_id)
+        elif top_level_only:
+            clauses.append("t.parent_task_id IS NULL")
         where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
         params.append(int(limit))
         with self._pool.connection() as conn:
             rows = conn.execute(
                 f"SELECT t.*, p.name AS project_name, p.color AS project_color, "
                 f"       s.name AS stage_name, s.is_done_stage, "
-                f"       (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count "
+                f"       (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count, "
+                f"       {self._SUBTASK_COUNTS_SQL} "
                 f"FROM tasks t "
                 f"JOIN projects p ON p.id = t.project_id "
                 f"LEFT JOIN project_stages s ON s.id = t.stage_id "
@@ -308,15 +355,61 @@ class ProjectStore:
             ).fetchall()
         return [self._serialize(r) for r in rows]
 
+    def _list_tasks_for_project_with_links(
+        self, *, project_id: int, assignee_email: str | None, limit: int,
+    ) -> list[dict[str, Any]]:
+        """`list_tasks` 의 "이 프로젝트 + 여기 연결된 업무" 경로.
+
+        `t.*` 뒤에 같은 이름(`stage_id`)을 다시 셀렉트해 덮어쓴다 — psycopg 의
+        dict_row 는 중복 컬럼명이 있으면 마지막 값으로 덮어쓰므로, "이 프로젝트
+        맥락에서의 세션"(연결이면 link.stage_id, 기본홈이면 t.stage_id)이
+        최종값이 된다. 원래 소속 프로젝트 배지(`project_name`/`project_color`)는
+        `t.project_id` 기준 그대로 둬 "다른 프로젝트에서 온 업무"임을 알 수 있게
+        한다.
+        """
+        clauses = [
+            "t.parent_task_id IS NULL",
+            "(t.project_id = %(pid)s OR link.project_id = %(pid)s)",
+        ]
+        params: dict[str, Any] = {"pid": project_id, "limit": int(limit)}
+        if assignee_email:
+            clauses.append("t.assignee_email = %(assignee)s")
+            params["assignee"] = store.norm_str(assignee_email, lower=True)
+        where = "WHERE " + " AND ".join(clauses)
+        with self._pool.connection() as conn:
+            rows = conn.execute(
+                f"""
+                SELECT t.*, p.name AS project_name, p.color AS project_color,
+                       (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count,
+                       {self._SUBTASK_COUNTS_SQL},
+                       (link.id IS NOT NULL) AS is_linked,
+                       COALESCE(es.name, s.name) AS stage_name,
+                       COALESCE(es.is_done_stage, s.is_done_stage) AS is_done_stage,
+                       COALESCE(link.stage_id, t.stage_id) AS stage_id
+                FROM tasks t
+                JOIN projects p ON p.id = t.project_id
+                LEFT JOIN project_stages s ON s.id = t.stage_id
+                LEFT JOIN task_project_links link
+                       ON link.task_id = t.id AND link.project_id = %(pid)s
+                      AND t.project_id <> %(pid)s
+                LEFT JOIN project_stages es ON es.id = link.stage_id
+                {where}
+                ORDER BY t.sort_order ASC, t.id ASC LIMIT %(limit)s
+                """,
+                params,
+            ).fetchall()
+        return [self._serialize(r) for r in rows]
+
     def get_task(self, *, task_id: int) -> dict[str, Any] | None:
         with self._pool.connection() as conn:
             row = conn.execute(
-                "SELECT t.*, p.name AS project_name, p.color AS project_color, "
+                f"SELECT t.*, p.name AS project_name, p.color AS project_color, "
-                "       s.name AS stage_name, s.is_done_stage, "
+                f"       s.name AS stage_name, s.is_done_stage, "
-                "       (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count "
+                f"       (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count, "
-                "FROM tasks t JOIN projects p ON p.id = t.project_id "
+                f"       {self._SUBTASK_COUNTS_SQL} "
-                "LEFT JOIN project_stages s ON s.id = t.stage_id "
+                f"FROM tasks t JOIN projects p ON p.id = t.project_id "
-                "WHERE t.id = %s",
+                f"LEFT JOIN project_stages s ON s.id = t.stage_id "
+                f"WHERE t.id = %s",
                 (task_id,),
             ).fetchone()
         return self._serialize(row) if row else None
@@ -336,6 +429,7 @@ class ProjectStore:
         start_time: str | None = None,
         due_time: str | None = None,
         created_by: str = "",
+        parent_task_id: int | None = None,
     ) -> dict[str, Any]:
         ttl = store.validate_task_title(title)
         pr = store.validate_priority(priority)
@@ -346,8 +440,18 @@ class ProjectStore:
         assignee = store.norm_str(assignee_email, lower=True)
         with self._pool.connection() as conn:
             with conn.transaction():
-                # stage 미지정 시 프로젝트의 첫 단계로
+                if parent_task_id is not None:
-                if stage_id is None:
+                    # 하위 업무는 부모와 같은 프로젝트에 속하고(담당자 선택지 등이
+                    # 맞아떨어지게), 보드에는 안 보이므로 세션을 갖지 않는다.
+                    parent = conn.execute(
+                        "SELECT project_id FROM tasks WHERE id = %s", (parent_task_id,)
+                    ).fetchone()
+                    if not parent:
+                        raise KeyError(f"상위 업무를 찾을 수 없습니다: {parent_task_id}")
+                    project_id = parent["project_id"]
+                    stage_id = None
+                elif stage_id is None:
+                    # stage 미지정 시 프로젝트의 첫 세션으로
                     st = conn.execute(
                         "SELECT id FROM project_stages WHERE project_id = %s "
                         "ORDER BY sort_order ASC, id ASC LIMIT 1",
@@ -362,14 +466,14 @@ class ProjectStore:
                 row = conn.execute(
                     """
                     INSERT INTO tasks
-                        (project_id, stage_id, title, description, assignee_email,
+                        (project_id, stage_id, parent_task_id, title, description,
-                         assignee_name, priority, start_date, due_date,
+                         assignee_email, assignee_name, priority, start_date, due_date,
                          start_time, due_time, sort_order, created_by)
-                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                     RETURNING *
                     """,
                     (
-                        project_id, stage_id, ttl, store.norm_str(description),
+                        project_id, stage_id, parent_task_id, ttl, store.norm_str(description),
                         assignee, store.norm_str(assignee_name), pr, sd, dd,
                         sti, dti, int(nxt["n"]), store.norm_str(created_by, lower=True),
                     ),
@@ -482,6 +586,51 @@ class ProjectStore:
             ).fetchone()
         return bool(row and row["is_done_stage"])
 
+    # ════════════════════════════════════════════════════════════
+    # 멀티호밍 (task_project_links) — 업무 1개가 여러 프로젝트/세션에 동시 소속.
+    # 원래 소속(tasks.project_id/stage_id)은 여기서 건드리지 않는다("기본 홈"은
+    # 업무 삭제로만 없어진다). 같은 프로젝트에 중복 연결하면 세션만 갱신한다.
+    # ════════════════════════════════════════════════════════════
+    def add_task_link(
+        self, *, task_id: int, project_id: int, stage_id: int | None = None,
+        created_by: str = "",
+    ) -> dict[str, Any]:
+        with self._pool.connection() as conn:
+            row = conn.execute(
+                """
+                INSERT INTO task_project_links (task_id, project_id, stage_id, created_by)
+                VALUES (%s,%s,%s,%s)
+                ON CONFLICT (task_id, project_id) DO UPDATE
+                   SET stage_id = EXCLUDED.stage_id
+                RETURNING *
+                """,
+                (task_id, project_id, stage_id, store.norm_str(created_by, lower=True)),
+            ).fetchone()
+        return self._serialize(row)
+
+    def remove_task_link(self, *, task_id: int, project_id: int) -> None:
+        with self._pool.connection() as conn:
+            conn.execute(
+                "DELETE FROM task_project_links WHERE task_id = %s AND project_id = %s",
+                (task_id, project_id),
+            )
+
+    def list_task_links(self, *, task_id: int) -> list[dict[str, Any]]:
+        with self._pool.connection() as conn:
+            rows = conn.execute(
+                """
+                SELECT l.*, p.name AS project_name, p.color AS project_color,
+                       s.name AS stage_name
+                FROM task_project_links l
+                JOIN projects p ON p.id = l.project_id
+                LEFT JOIN project_stages s ON s.id = l.stage_id
+                WHERE l.task_id = %s
+                ORDER BY l.created_at ASC
+                """,
+                (task_id,),
+            ).fetchall()
+        return [self._serialize(r) for r in rows]
+
     # ════════════════════════════════════════════════════════════
     # 활동 이력
     # ════════════════════════════════════════════════════════════
@@ -748,7 +897,8 @@ class ProjectStore:
                 out[k] = v.strftime("%H:%M")
             elif isinstance(v, date):
                 out[k] = v.isoformat()
-        for k in ("id", "parent_id", "project_id", "stage_id", "task_id", "sort_order"):
+        for k in ("id", "parent_id", "project_id", "stage_id", "task_id", "sort_order",
+                  "parent_task_id"):
             if k in out and out[k] is not None:
                 try:
                     out[k] = int(out[k])

app/modules/project/router.py

@@ -570,6 +570,7 @@ async def project_page(request: Request, project_id: int) -> HTMLResponse:
             "page_title": project["name"],
             "page_subtitle": project.get("description") or "프로젝트 보드",
             "project": project,
+            "current_project_id": project_id,
             "nav_projects": nav_projects,
             "stages": stages,
             "tasks": all_tasks,
@@ -619,8 +620,8 @@ async def api_list_projects(request: Request) -> JSONResponse:
 async def api_create_project(
     request: Request, payload: dict[str, Any] = Body(...)
 ) -> JSONResponse:
-    """최상위 프로젝트 생성 — 관리자 전용."""
+    """최상위 프로젝트 생성 — 로그인해 프로젝트 모듈에 접근할 수 있는 사용자 누구나."""
-    user = _require_admin(request)
+    user = _require_user(request)
     st = _db_or_503(request)
     try:
         project = st.create_project(
@@ -778,6 +779,7 @@ async def api_add_stage(
             project_id=project_id,
             name=payload.get("name", ""),
             is_done_stage=bool(payload.get("is_done_stage", False)),
+            created_by=user["email"],
         )
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc))
@@ -788,13 +790,20 @@ async def api_add_stage(
 async def api_delete_stage(
     request: Request, project_id: int, stage_id: int
 ) -> JSONResponse:
+    """세션(단계) 삭제 — 만든 사람만(슈퍼관리자 예외). 생성자 정보가 없는 과거
+    데이터는 관리자에게 허용(잠김 방지) — `_can_delete` 규칙 그대로 재사용."""
     user = _require_user(request)
     st = _db_or_503(request)
     _require_manage(request, st, user, project_id)
+    stage = st.get_stage(stage_id=stage_id)
+    if stage is None:
+        raise HTTPException(status_code=404, detail="세션을 찾을 수 없습니다.")
+    if not _can_delete(user, stage.get("created_by")):
+        raise HTTPException(status_code=403, detail=_DELETE_DENIED)
     try:
         st.delete_stage(stage_id=stage_id)
     except KeyError:
-        raise HTTPException(status_code=404, detail="단계를 찾을 수 없습니다.")
+        raise HTTPException(status_code=404, detail="세션을 찾을 수 없습니다.")
     return JSONResponse({"ok": True})
 
 
@@ -834,7 +843,7 @@ async def api_create_task(
             project_id=project_id,
             title=payload.get("title", ""),
             stage_id=payload.get("stage_id"),
-            description=payload.get("description", ""),
+            description=store.sanitize_description_html(payload.get("description", "")),
             assignee_email=payload.get("assignee_email", ""),
             assignee_name=payload.get("assignee_name", ""),
             priority=payload.get("priority", "normal"),
@@ -843,9 +852,12 @@ async def api_create_task(
             start_time=payload.get("start_time"),
             due_time=payload.get("due_time"),
             created_by=user["email"],
+            parent_task_id=payload.get("parent_task_id"),
         )
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc))
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail=str(exc))
     # 생성과 동시에 담당자 지정 시 관리자 메일 + 담당자 인앱 알림
     if task.get("assignee_email"):
         _notify(
@@ -886,6 +898,8 @@ async def api_update_task(
     if existing is None:
         raise HTTPException(status_code=404, detail="업무를 찾을 수 없습니다.")
     _require_manage(request, st, user, existing["project_id"])
+    if "description" in payload:
+        payload["description"] = store.sanitize_description_html(payload["description"])
     try:
         task = st.update_task(task_id=task_id, fields=payload, actor=user["email"])
     except ValueError as exc:
@@ -952,6 +966,65 @@ async def api_delete_task(request: Request, task_id: int) -> JSONResponse:
     return JSONResponse({"ok": True})
 
 
+@router.get("/api/tasks/{task_id}")
+async def api_get_task(request: Request, task_id: int) -> JSONResponse:
+    """업무 상세 — 하위 업무·연결된 프로젝트 포함. 업무 편집 팝업을 열 때 쓴다."""
+    user = _require_user(request)
+    st = _db_or_503(request)
+    task = st.get_task(task_id=task_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="업무를 찾을 수 없습니다.")
+    _require_manage(request, st, user, task["project_id"])
+    task["subtasks"] = st.list_tasks(parent_task_id=task_id)
+    task["links"] = st.list_task_links(task_id=task_id)
+    return JSONResponse({"task": task})
+
+
+# ────────────────────────────────────────────────────────────
+# JSON API — 업무 멀티호밍 (다른 프로젝트/세션에 연결)
+# ────────────────────────────────────────────────────────────
+@router.post("/api/tasks/{task_id}/links")
+async def api_add_task_link(
+    request: Request, task_id: int, payload: dict[str, Any] = Body(...)
+) -> JSONResponse:
+    """업무를 다른 프로젝트(의 한 세션)에도 연결한다(아사나의 멀티호밍).
+
+    원래 소속(프로젝트)과 연결할 대상 프로젝트 양쪽에 대한 관리 권한이 필요하다.
+    """
+    user = _require_user(request)
+    st = _db_or_503(request)
+    task = st.get_task(task_id=task_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="업무를 찾을 수 없습니다.")
+    _require_manage(request, st, user, task["project_id"])
+    try:
+        target_project_id = int(payload.get("project_id"))
+    except (TypeError, ValueError):
+        raise HTTPException(status_code=400, detail="project_id 가 필요합니다.")
+    if target_project_id == task["project_id"]:
+        raise HTTPException(status_code=400, detail="이미 이 업무의 기본 프로젝트입니다.")
+    _require_manage(request, st, user, target_project_id)
+    stage_id = payload.get("stage_id")
+    link = st.add_task_link(
+        task_id=task_id, project_id=target_project_id,
+        stage_id=int(stage_id) if stage_id else None, created_by=user["email"],
+    )
+    return JSONResponse({"link": link}, status_code=201)
+
+
+@router.delete("/api/tasks/{task_id}/links/{project_id}")
+async def api_remove_task_link(request: Request, task_id: int, project_id: int) -> JSONResponse:
+    user = _require_user(request)
+    st = _db_or_503(request)
+    task = st.get_task(task_id=task_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="업무를 찾을 수 없습니다.")
+    _require_manage(request, st, user, task["project_id"])
+    _require_manage(request, st, user, project_id)
+    st.remove_task_link(task_id=task_id, project_id=project_id)
+    return JSONResponse({"ok": True})
+
+
 @router.get("/api/projects/{project_id}/activity")
 async def api_activity(request: Request, project_id: int) -> JSONResponse:
     _require_user(request)

app/modules/project/store.py

@@ -6,6 +6,7 @@ DB 없이 단위 테스트 가능한 순수 함수만 둔다(malaysia.store 패
 
 from __future__ import annotations
 
+from html.parser import HTMLParser
 from typing import Any
 
 # 프로젝트 생성 시 자동으로 깔리는 기본 진행 단계(칸반 컬럼).
@@ -134,3 +135,119 @@ def build_subject_completed(*, project_name: str, task_title: str) -> str:
 
 def build_subject_project_done(*, project_name: str) -> str:
     return f"[DBX 프로젝트] 프로젝트 완료: {project_name}"
+
+
+# ════════════════════════════════════════════════════════════
+# 업무 설명 HTML 정리 (이미지 붙여넣기/업로드 삽입 지원)
+#
+# 업무 편집창의 설명란은 contenteditable 이라 브라우저가 임의의 HTML을
+# 만들어 보낼 수 있다. 여기서 허용 태그만 남기고 스크립트/이벤트 속성/위험
+# URL 스킴을 제거한다 — 외부 라이브러리(bleach 등) 없이 표준 라이브러리
+# html.parser 만으로 구현한다(카페24 모듈 자체호스팅 원칙과 동일).
+#
+# 기존에 일반 텍스트로 저장돼 있던 설명은 마이그레이션(project_db_003)에서
+# 한 번에 안전한 HTML로 변환해뒀다 — 여기서는 "이게 HTML이냐 텍스트냐"를
+# 런타임에 추측하지 않는다(추측은 오탐이 난다). description 컬럼은 이제
+# 항상 HTML이라고 가정한다.
+# ════════════════════════════════════════════════════════════
+_DESC_ALLOWED_TAGS: frozenset[str] = frozenset({
+    "p", "br", "b", "strong", "i", "em", "u", "a", "img",
+    "ul", "ol", "li", "blockquote", "code", "pre", "span", "div",
+})
+_DESC_VOID_TAGS: frozenset[str] = frozenset({"br", "img"})
+_DESC_ALLOWED_ATTRS: dict[str, frozenset[str]] = {
+    "a": frozenset({"href"}),
+    "img": frozenset({"src", "alt", "width", "height"}),
+}
+_DESC_DANGEROUS_URL_PREFIXES: tuple[str, ...] = ("javascript:", "vbscript:", "data:text/html")
+# script/style 은 태그 자체는 물론 그 안의 텍스트(코드)도 통째로 버린다 —
+# 허용 태그 목록에 없어 태그는 이미 안 남지만, HTMLParser는 이 둘의 내용을
+# 별도 데이터로 넘기므로 handle_data 에서 따로 걸러야 한다.
+_DESC_RAW_TEXT_TAGS: frozenset[str] = frozenset({"script", "style"})
+
+
+def _desc_escape_text(text: str) -> str:
+    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+def _desc_escape_attr(value: str) -> str:
+    return _desc_escape_text(value).replace('"', "&quot;")
+
+
+def _desc_url_is_safe(value: str) -> bool:
+    v = value.strip().lower()
+    return not any(v.startswith(prefix) for prefix in _DESC_DANGEROUS_URL_PREFIXES)
+
+
+class _DescriptionSanitizer(HTMLParser):
+    """허용 태그/속성만 통과시키는 최소 HTML 정리기."""
+
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self.out: list[str] = []
+        self._open: list[str] = []
+        self._raw = False  # <script>/<style> 내부(태그 자체와 별개로 텍스트도 버림)
+
+    def _emit_start(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        if tag in _DESC_RAW_TEXT_TAGS:
+            self._raw = True
+            return
+        if tag not in _DESC_ALLOWED_TAGS:
+            return
+        allowed = _DESC_ALLOWED_ATTRS.get(tag, frozenset())
+        kept: list[str] = []
+        for name, value in attrs:
+            if name not in allowed:
+                continue
+            value = value or ""
+            if name in ("href", "src") and not _desc_url_is_safe(value):
+                continue
+            kept.append(f'{name}="{_desc_escape_attr(value)}"')
+        attr_text = (" " + " ".join(kept)) if kept else ""
+        self.out.append(f"<{tag}{attr_text}>")
+        if tag not in _DESC_VOID_TAGS:
+            self._open.append(tag)
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        self._emit_start(tag, attrs)
+
+    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        self._emit_start(tag, attrs)
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag in _DESC_RAW_TEXT_TAGS:
+            self._raw = False
+            return
+        if tag not in _DESC_ALLOWED_TAGS or tag in _DESC_VOID_TAGS:
+            return
+        if tag not in self._open:
+            return
+        # 짝이 안 맞는 깨진 HTML 방어 — 스택에서 같은 태그가 나올 때까지 닫는다.
+        while self._open:
+            top = self._open.pop()
+            self.out.append(f"</{top}>")
+            if top == tag:
+                break
+
+    def handle_data(self, data: str) -> None:
+        if self._raw:
+            return
+        self.out.append(_desc_escape_text(data))
+
+    def close(self) -> None:
+        super().close()
+        while self._open:
+            self.out.append(f"</{self._open.pop()}>")
+
+
+def sanitize_description_html(html: str) -> str:
+    """업무 설명 HTML을 허용 태그만 남기고 정리한다.
+
+    스크립트·이벤트 속성(onerror 등)·위험 URL 스킴(javascript: 등)을 제거한다.
+    저장 직전 항상 이 함수를 거친다 — 화면에서 붙여넣은 이미지/서식은 거의
+    그대로 남고, 위험한 부분만 사라진다.
+    """
+    parser = _DescriptionSanitizer()
+    parser.feed(html or "")
+    parser.close()
+    return "".join(parser.out).strip()

app/modules/project/templates/project/index.html

@@ -56,7 +56,7 @@
           <span class="material-symbols-outlined">add_task</span>
         </button>
         {% endif %}
-        {% if is_admin %}
+        {% if p.can_add_task %}
         <button type="button" class="pj-icon-btn pj-card-edit" title="프로젝트 수정">
           <span class="material-symbols-outlined">edit</span>
         </button>
@@ -111,19 +111,15 @@
             <span class="material-symbols-outlined">notifications</span>
             <span class="pj-bell-badge" id="pj-bell-badge" hidden>0</span>
           </a>
-          {% if is_admin %}
           <button type="button" class="pj-btn pj-btn-primary" id="pj-new-project">
             + 새 프로젝트
           </button>
-          {% endif %}
         </div>
       </div>
 
       {% if not active_projects and not done_projects %}
       <div class="pj-empty">
-        아직 프로젝트가 없습니다.
+        아직 프로젝트가 없습니다. 상단의 <b>새 프로젝트</b> 버튼으로 만들어 보세요.
-        {% if is_admin %}상단의 <b>새 프로젝트</b> 버튼으로 만들어 보세요.
-        {% else %}관리자가 프로젝트에 배정하면 여기에 표시됩니다.{% endif %}
       </div>
       {% elif not active_projects %}
       <div class="pj-empty pj-empty-sm">진행중인 프로젝트가 없습니다.</div>
@@ -192,8 +188,7 @@
   </div>
 </div>
 
-<!-- 새 프로젝트 모달 -->
+<!-- 새 프로젝트 모달 — 생성은 로그인한 사용자 누구나 -->
-{% if is_admin %}
 <div class="pj-modal" id="pj-modal-project" hidden>
   <div class="pj-modal-card">
     <h3>새 프로젝트</h3>
@@ -241,7 +236,6 @@
     </div>
   </div>
 </div>
-{% endif %}
 
 <!-- 업무 편집 모달 — 카드/내 업무에서 클릭 시 페이지 이동 없이 팝업으로 편집 -->
 <div class="pj-modal" id="pj-modal-task" hidden>

app/modules/project/templates/project/project.html

@@ -279,6 +279,7 @@
 <script id="pj-data" type="application/json">
 {
   "project": {{ project | tojson }},
+  "currentProjectId": {{ current_project_id | tojson }},
   "stages": {{ stages | tojson }},
   "tasks": {{ tasks | tojson }},
   "members": {{ members | tojson }},

app/static/project.js

@@ -71,6 +71,7 @@
     const wrap = document.querySelector(".pj-wrap");
     const me = ((wrap && wrap.dataset.me) || "").toLowerCase();
     const isSuper = !!(wrap && wrap.dataset.isSuper === "true");
+    const isAdmin = !!(wrap && wrap.dataset.isAdmin === "true");
     function canDelete(createdBy) {
       if (isSuper) return true;
       const cb = (createdBy || "").toLowerCase();
@@ -108,6 +109,12 @@
       newBtn.addEventListener("click", async function () {
         openModal(modal);
         if (membersBox) {
+          // 멤버 배정은 관리자 전용 API(/assignable-users)라 관리자가 아니면
+          // 호출해도 403 이 난다 — 아예 시도하지 않고 안내만 보여준다.
+          // (프로젝트 생성 자체는 이제 누구나 가능 — 멤버는 나중에 관리자가 배정)
+          if (!isAdmin) {
+            membersBox.innerHTML = "<span class='pj-empty-sm'>멤버 배정은 관리자가 나중에 할 수 있습니다.</span>";
+          } else {
             membersBox.innerHTML = "불러오는 중…";
             try {
               const res = await api("GET", "/project/api/assignable-users");
@@ -120,6 +127,7 @@
               membersBox.innerHTML = "<span class='pj-empty-sm'>불러오기 실패</span>";
             }
           }
+        }
       });
       el("pj-save-project").addEventListener("click", async function () {
         const name = el("pj-f-name").value.trim();

scripts/sql/project_db_003_sections_subtasks_links.sql

@@ -0,0 +1,59 @@
+-- =====================================================================
+-- project_db 003 — 세션(단계) 삭제자 기록 · 하위 업무 · 프로젝트 간 멀티호밍
+-- =====================================================================
+-- 멱등(idempotent): 여러 번 실행해도 안전하다 — 단, 아래 UPDATE(설명 이스케이프)는
+-- 예외다. 그 문장은 반드시 1회만 실행할 것(주석 참고). DROP/TRUNCATE 없음.
+--
+-- 실행:
+--   docker exec -i postgres-db psql -U postgres -d project_db \
+--       < scripts/sql/project_db_003_sections_subtasks_links.sql
+--
+-- 배경:
+--   프로젝트 → 세션(기존 "단계"를 그대로 재사용) → 업무 3단계 구조로 바뀌면서
+--   1) 세션도 "만든 사람만 삭제" 규칙을 적용하려면 만든 사람을 기록해야 한다.
+--   2) 업무는 하위 업무(subtask)를 가질 수 있어야 한다 — tasks 테이블 자체를
+--      재사용한다(parent_task_id 만 다르다. 부모 삭제 시 CASCADE).
+--   3) 업무 하나가 여러 프로젝트/세션에 동시에 속하는 멀티호밍(아사나 기능)을
+--      지원해야 한다 — 원래 소속(tasks.project_id/stage_id)은 "기본 홈"으로 두고,
+--      추가 소속만 담는 새 표를 둔다.
+-- =====================================================================
+
+\set ON_ERROR_STOP on
+
+ALTER TABLE project_stages ADD COLUMN IF NOT EXISTS created_by TEXT NOT NULL DEFAULT '';
+
+ALTER TABLE tasks ADD COLUMN IF NOT EXISTS parent_task_id BIGINT REFERENCES tasks(id) ON DELETE CASCADE;
+CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks (parent_task_id);
+
+-- ⚠️ 1회성 데이터 변환 — 반드시 한 번만 실행할 것(두 번 실행하면 이중 이스케이프됨).
+-- 지금까지 tasks.description 은 일반 텍스트였다. 앞으로 화면은 이 컬럼을 항상
+-- HTML로 취급한다(붙여넣기 이미지 삽입을 지원하기 위해 — app/modules/project/
+-- store.py 의 sanitize_description_html 참고). 런타임에 "이게 HTML이냐 텍스트냐"
+-- 판단하는 방식은 오탐이 나므로 쓰지 않고, 기존 값을 여기서 한 번에 안전한 HTML로
+-- 바꿔둔다(개행→<br>, &/</> 이스케이프 — 순서가 중요하다: &부터 바꿔야 한다).
+UPDATE tasks
+SET description = replace(replace(replace(replace(
+        description, '&', '&amp;'), '<', '&lt;'), '>', '&gt;'), E'\n', '<br>')
+WHERE description IS NOT NULL AND description <> '';
+
+-- ════════════════════════════════════════════════════════════
+-- 멀티호밍 — 업무 1개가 여러 프로젝트(각각 세션 지정)에 동시에 속함.
+-- 원래 소속(tasks.project_id/stage_id)은 그대로 "기본 홈"이고, 이 표는 "추가
+-- 홈"만 담는다. 기본 홈은 이 표로 지우지 못한다(업무 삭제로만 없어진다).
+-- ════════════════════════════════════════════════════════════
+CREATE TABLE IF NOT EXISTS task_project_links (
+    id          BIGSERIAL   PRIMARY KEY,
+    task_id     BIGINT      NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
+    project_id  BIGINT      NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+    stage_id    BIGINT      REFERENCES project_stages(id) ON DELETE SET NULL,
+    created_by  TEXT        NOT NULL DEFAULT '',
+    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (task_id, project_id)
+);
+CREATE INDEX IF NOT EXISTS idx_task_links_task    ON task_project_links (task_id);
+CREATE INDEX IF NOT EXISTS idx_task_links_project ON task_project_links (project_id, stage_id);
+
+GRANT SELECT, INSERT, UPDATE, DELETE ON task_project_links TO project_app;
+GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO project_app;
+
+SELECT 'project_db 003 applied' AS status;