코드와 CSV Artifact의 원본 다운로드를 지원함

.github/workflows/ci.yml

@@ -74,6 +74,9 @@ jobs:
       - name: Test search masking feedback
         run: npx playwright test --config playwright.search-masking.config.ts
 
+      - name: Test artifact source downloads
+        run: npx playwright test --config playwright.artifact-source.config.ts
+
   api:
     name: API · lint, tests
     runs-on: ubuntu-latest

apps/api/app/routers/workspace.py

@@ -2086,6 +2086,35 @@ def _attachment(body: bytes, media: str, stem: str, suffix: str) -> Response:
     )
 
 
+_SOURCE_EXTENSIONS = {
+    "csv": "csv", "tsv": "tsv", "json": "json", "yaml": "yaml", "yml": "yml",
+    "python": "py", "py": "py", "javascript": "js", "js": "js", "jsx": "jsx",
+    "typescript": "ts", "ts": "ts", "tsx": "tsx", "bash": "sh", "shell": "sh",
+    "sh": "sh", "zsh": "zsh", "sql": "sql", "css": "css", "xml": "xml",
+    "markdown": "md", "md": "md", "text": "txt", "txt": "txt", "plain": "txt",
+}
+_SOURCE_MEDIA = {
+    "csv": "text/csv", "tsv": "text/tab-separated-values", "json": "application/json",
+}
+
+
+def _export_code_source(artifact: Artifact) -> Response:
+    """Download stored source, with no execution, conversion or language inference."""
+    data = artifact.data or {}
+    content = data.get("content")
+    if not isinstance(content, str):
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="missing_source")
+    language = str(data.get("language") or "").strip().lower()
+    suffix = _SOURCE_EXTENSIONS.get(language, "txt")
+    media = _SOURCE_MEDIA.get(suffix, "text/plain")
+    stem = re.sub(r'[\\/:*?"<>|\x00-\x1f\x7f]+', "_", artifact.title or "").strip(" .")
+    if stem.lower().endswith("." + suffix):
+        stem = stem[: -len(suffix) - 1].rstrip(" .")
+    response = _attachment(content.encode("utf-8"), media, stem[:60] or "code", suffix)
+    response.headers["Access-Control-Expose-Headers"] = "Content-Disposition"
+    return response
+
+
 def _export_deck(artifact: Artifact, format: str) -> Response:
     """A deck as `.pptx`, `.pdf` or Markdown."""
     slides = list((artifact.data or {}).get("slides") or [])
@@ -2216,12 +2245,14 @@ async def _export_page(artifact: Artifact, format: str) -> Response:
 
 @router.get("/artifacts/{artifact_id}/export")
 async def export_artifact(artifact_id: str, user: CurrentUser, db: DbSession, format: str = "docx"):
-    """A report, deck, or HTML artifact as a file.
+    """An owned artifact as a file.
 
     Reports take `docx`, `pdf`, `hwpx` or `md`; decks take `pptx`, `pdf` or `md`;
-    HTML artifacts take `html` plus the set matching their template.
+    HTML artifacts take `html` plus the set matching their template; code takes `source`.
     """
     artifact = await _own(db, Artifact, "user_id", user, artifact_id)
+    if artifact.kind is ArtifactKind.code and format == "source":
+        return _export_code_source(artifact)
     if artifact.kind not in (ArtifactKind.report, ArtifactKind.deck, ArtifactKind.html):
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="not_exportable")
 

apps/api/app/services/tools/builtin.py

@@ -1051,7 +1051,11 @@ async def create_artifact(args: dict[str, Any], ctx: ToolContext) -> ToolResult:
             },
             "language": {
                 "type": "string",
-                "description": "kind 가 code 일 때의 언어 (python, bash, yaml 등).",
+                "description": (
+                    "kind 가 code 일 때 원본 다운로드의 확장자를 결정하는 언어. "
+                    "CSV는 csv, JSON은 json, YAML은 yaml, Python은 python으로 지정하세요. "
+                    "생략하거나 지원하지 않는 언어면 text로 취급해 .txt로 다운로드합니다."
+                ),
             },
             "userRequested": {
                 "type": "boolean",

apps/api/tests/test_code_artifact_source_export.py

@@ -0,0 +1,195 @@
+"""Stored code can be downloaded without re-generation, execution or format guessing."""
+
+from copy import deepcopy
+from urllib.parse import unquote
+
+import httpx
+import pytest
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.core import deps
+from app.core.db import get_session
+from app.models.user import User
+from app.models.workspace import Artifact, ArtifactKind
+from app.routers import workspace
+from app.services.tools.builtin import CREATE_ARTIFACT
+
+
+class _ReadOnlyDb:
+    def __init__(self, artifact):
+        self.artifact = artifact
+        self.reads = []
+
+    async def get(self, model, item_id):
+        self.reads.append((model, item_id))
+        return self.artifact
+
+    def add(self, _row):
+        pytest.fail("Downloading stored code must not write to the database")
+
+    async def commit(self):
+        pytest.fail("Downloading stored code must not commit")
+
+
+def _artifact(language="csv", *, title="합성 자료", content="label,value\r\n가,1\r\n"):
+    data = {"content": content}
+    if language is not None:
+        data["language"] = language
+    return Artifact(
+        id="artifact-1", user_id="owner", kind=ArtifactKind.code, title=title, data=data
+    )
+
+
+def _user(user_id="owner"):
+    return User(
+        id=user_id, email="synthetic@example.test", password_hash="unused", name="Synthetic"
+    )
+
+
+def _filename(response):
+    disposition = response.headers["content-disposition"]
+    assert disposition.startswith("attachment; filename*=UTF-8''")
+    return unquote(disposition.split("''", 1)[1])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("language,suffix,media", [
+    ("csv", "csv", "text/csv"),
+    ("tsv", "tsv", "text/tab-separated-values"),
+    ("json", "json", "application/json"),
+    ("python", "py", "text/plain"),
+    ("py", "py", "text/plain"),
+    ("javascript", "js", "text/plain"),
+    ("typescript", "ts", "text/plain"),
+    ("yaml", "yaml", "text/plain"),
+    ("yml", "yml", "text/plain"),
+    ("bash", "sh", "text/plain"),
+    ("sql", "sql", "text/plain"),
+    ("markdown", "md", "text/plain"),
+    ("text", "txt", "text/plain"),
+    (None, "txt", "text/plain"),
+    ("", "txt", "text/plain"),
+    ("unknown-format", "txt", "text/plain"),
+    ("html", "txt", "text/plain"),
+])
+async def test_source_download_keeps_exact_saved_bytes_and_explicit_format(language, suffix, media):
+    artifact = _artifact(language)
+    original = deepcopy(artifact.data)
+    db = _ReadOnlyDb(artifact)
+    response = await workspace.export_artifact(artifact.id, _user(), db, format="source")
+    assert response.status_code == 200
+    assert response.body == artifact.data["content"].encode("utf-8")
+    assert response.headers["content-type"].split(";", 1)[0] == media
+    assert response.headers["x-content-type-options"] == "nosniff"
+    assert _filename(response) == f"합성 자료.{suffix}"
+    assert artifact.data == original
+    assert db.reads == [(Artifact, artifact.id)]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("title,language,wanted", [
+    ("data.csv", "csv", "data.csv"),
+    ("data.CSV", " CSV ", "data.csv"),
+    ("data.csv", None, "data.csv.txt"),
+    ("../../자료\r\n.csv", "csv", "_.._자료_.csv"),
+    ("", None, "code.txt"),
+])
+async def test_filename_uses_safe_title_without_guessing_language(title, language, wanted):
+    artifact = _artifact(language, title=title)
+    response = await workspace.export_artifact(
+        artifact.id, _user(), _ReadOnlyDb(artifact), "source"
+    )
+    assert _filename(response) == wanted
+    assert "\r" not in response.headers["content-disposition"]
+    assert "\n" not in response.headers["content-disposition"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("artifact,user", [(_artifact(), _user("other")), (None, _user())])
+async def test_other_owner_and_absent_ids_are_indistinguishable(artifact, user):
+    with pytest.raises(HTTPException) as error:
+        await workspace.export_artifact("requested-id", user, _ReadOnlyDb(artifact), "source")
+    assert error.value.status_code == 404
+    assert error.value.detail == "not_found"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", [kind for kind in ArtifactKind if kind is not ArtifactKind.code])
+async def test_source_export_does_not_admit_other_artifact_kinds(kind):
+    artifact = _artifact()
+    artifact.kind = kind
+    with pytest.raises(HTTPException) as error:
+        await workspace.export_artifact(artifact.id, _user(), _ReadOnlyDb(artifact), "source")
+    assert error.value.status_code == 400
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("format", ["docx", "pdf", "csv", "html"])
+async def test_code_export_does_not_add_conversions(format):
+    artifact = _artifact()
+    with pytest.raises(HTTPException) as error:
+        await workspace.export_artifact(artifact.id, _user(), _ReadOnlyDb(artifact), format)
+    assert error.value.status_code == 400
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("content", [None, 17, {"guess": "not source"}])
+async def test_non_text_source_is_not_stringified(content):
+    artifact = _artifact(content=content)
+    with pytest.raises(HTTPException) as error:
+        await workspace.export_artifact(artifact.id, _user(), _ReadOnlyDb(artifact), "source")
+    assert error.value.status_code == 400
+
+
+def test_tool_description_explains_the_optional_language_download_contract():
+    language = CREATE_ARTIFACT.parameters["properties"]["language"]["description"]
+    assert "확장자" in language
+    assert "csv" in language and "json" in language and ".txt" in language
+    assert "language" not in CREATE_ARTIFACT.parameters["required"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("caller,status", [("owner", 200), ("other", 404)])
+@pytest.mark.parametrize("origin,allowed", [
+    ("https://app.example.test", True), ("https://other.example.test", False),
+])
+async def test_http_source_endpoint_preserves_bytes_and_owner_boundary(
+    caller, status, origin, allowed
+):
+    artifact = _artifact(content='"label","value"\r\n"가,나","=1+1"\r\n')
+    db = _ReadOnlyDb(artifact)
+    app = FastAPI()
+    app.add_middleware(CORSMiddleware, allow_origins=["https://app.example.test"])
+    app.add_api_route("/artifacts/{artifact_id}/export", workspace.export_artifact, methods=["GET"])
+    app.dependency_overrides[deps.current_user] = lambda: _user(caller)
+    app.dependency_overrides[get_session] = lambda: db
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="http://test"
+    ) as client:
+        response = await client.get(
+            f"/artifacts/{artifact.id}/export?format=source",
+            headers={"Origin": origin},
+        )
+    assert response.status_code == status
+    if status == 200:
+        assert response.content == artifact.data["content"].encode("utf-8")
+        assert response.headers["content-type"] == "text/csv; charset=utf-8"
+        assert _filename(response) == "합성 자료.csv"
+        assert response.headers.get("access-control-allow-origin") == (origin if allowed else None)
+        assert response.headers.get("access-control-expose-headers") == "Content-Disposition"
+    else:
+        assert response.json() == {"detail": "not_found"}
+
+
+@pytest.mark.asyncio
+async def test_code_with_markup_is_downloaded_as_inert_text_without_rendering():
+    source = '<script>throw new Error("must not execute")</script>'
+    artifact = _artifact("html", title="page.html", content=source)
+    response = await workspace.export_artifact(
+        artifact.id, _user(), _ReadOnlyDb(artifact), "source"
+    )
+    assert response.body == source.encode()
+    assert response.headers["content-type"] == "text/plain; charset=utf-8"
+    assert response.headers["x-content-type-options"] == "nosniff"
+    assert _filename(response) == "page.html.txt"