Allow group workspaces to upload Agent-supported formats

backend/app/api/groups.py

@@ -5,13 +5,16 @@
 import uuid
 from datetime import UTC, datetime
 from typing import Annotated, Any, Literal
+from urllib.parse import quote
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
+from fastapi.responses import Response
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from pydantic import BaseModel, ConfigDict, Field
 from sqlalchemy import exists, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.core.security import get_current_user
+from app.core.security import decode_access_token, get_current_user
 from app.database import get_db
 from app.models.agent import Agent
 from app.models.agent_run import AgentRun
@@ -38,6 +41,7 @@
 from app.services.group_message_service import GroupMessageServiceError
 from app.services.group_realtime import publish_group_message_created
 from app.services.participant_identity import get_or_create_user_participant
+from app.services.storage import guess_content_type
 
 
 router = APIRouter(prefix="/api/groups", tags=["groups"])
@@ -188,6 +192,14 @@ class GroupWorkspaceEntryOut(BaseModel):
     version_token: str | None = None
 
 
+class GroupWorkspaceUploadOut(BaseModel):
+    path: str
+    size: int
+    version_token: str
+    modified_at: str | None = None
+    revision_id: uuid.UUID | None = None
+
+
 class GroupSessionSummaryOut(BaseModel):
     version: int
     summary: str
@@ -1403,6 +1415,130 @@ async def put_group_workspace_file(
     return _text_file_out(value)
 
 
+@router.post("/{group_id}/workspace/upload", response_model=GroupWorkspaceUploadOut)
+async def upload_group_workspace_file(
+    group_id: uuid.UUID,
+    path: Annotated[str, Query(min_length=1, max_length=500)],
+    file: UploadFile = File(...),
+    expected_version_token: Annotated[str | None, Query()] = None,
+    require_absent: Annotated[bool, Query()] = False,
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Upload one group workspace file without converting binary bytes to text."""
+    tenant_id = _tenant_id(current_user)
+    participant = await _current_participant(db, current_user)
+    try:
+        value = await group_file_service.write_workspace_binary_file(
+            db,
+            tenant_id=tenant_id,
+            group_id=group_id,
+            actor_participant_id=participant.id,
+            path=path,
+            content=await file.read(),
+            content_type=guess_content_type(path),
+            expected_version_token=expected_version_token,
+            require_absent=require_absent,
+        )
+    except GroupChatServiceError as exc:
+        raise _translate_domain_error(exc) from exc
+    except GroupFileServiceError as exc:
+        raise _translate_file_error(exc) from exc
+    _stage_audit(
+        db,
+        current_user=current_user,
+        action="group:workspace_write",
+        tenant_id=tenant_id,
+        group_id=group_id,
+        details={
+            "path": value.path,
+            "revision_id": str(value.revision_id) if value.revision_id else None,
+        },
+    )
+    return GroupWorkspaceUploadOut(
+        path=value.path,
+        size=len(value.content),
+        version_token=value.version_token,
+        modified_at=value.modified_at,
+        revision_id=value.revision_id,
+    )
+
+
+async def _download_user(
+    *,
+    token: str,
+    credentials: HTTPAuthorizationCredentials | None,
+    db: AsyncSession,
+) -> User:
+    jwt_token = credentials.credentials if credentials is not None else token
+    if not jwt_token:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Authentication required",
+        )
+    payload = decode_access_token(jwt_token)
+    user_id = payload.get("sub")
+    if not user_id:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
+    try:
+        parsed_user_id = uuid.UUID(user_id)
+    except (TypeError, ValueError) as exc:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") from exc
+    result = await db.execute(select(User).where(User.id == parsed_user_id))
+    user = result.scalar_one_or_none()
+    if user is None or not user.is_active:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="User not found or inactive",
+        )
+    return user
+
+
+@router.get("/{group_id}/workspace/download")
+async def download_group_workspace_file(
+    group_id: uuid.UUID,
+    path: Annotated[str, Query(min_length=1, max_length=500)],
+    token: str = "",
+    inline: bool = False,
+    credentials: HTTPAuthorizationCredentials | None = Depends(HTTPBearer(auto_error=False)),
+    db: AsyncSession = Depends(get_db),
+):
+    """Download a group workspace file with membership authorization."""
+    current_user = await _download_user(token=token, credentials=credentials, db=db)
+    tenant_id = _tenant_id(current_user)
+    participant = await _current_participant(db, current_user)
+    try:
+        value = await group_file_service.read_workspace_binary_file(
+            db,
+            tenant_id=tenant_id,
+            group_id=group_id,
+            actor_participant_id=participant.id,
+            path=path,
+        )
+    except GroupChatServiceError as exc:
+        raise _translate_domain_error(exc) from exc
+    except GroupFileServiceError as exc:
+        raise _translate_file_error(exc) from exc
+    filename = value.path.rsplit("/", 1)[-1]
+    media_type = guess_content_type(filename)
+    inline_media_types = {
+        "image/gif",
+        "image/jpeg",
+        "image/png",
+        "image/svg+xml",
+        "image/webp",
+    }
+    allow_inline = inline and media_type in inline_media_types
+    disposition = "inline" if allow_inline else "attachment"
+    headers = {
+        "Content-Disposition": f"{disposition}; filename*=UTF-8''{quote(filename)}",
+        "X-Content-Type-Options": "nosniff",
+    }
+    if allow_inline and media_type == "image/svg+xml":
+        headers["Content-Security-Policy"] = "sandbox; default-src 'none'; style-src 'unsafe-inline'"
+    return Response(content=value.content, media_type=media_type, headers=headers)
+
+
 @router.delete("/{group_id}/workspace/file", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_group_workspace_file(
     group_id: uuid.UUID,

backend/app/services/group_file_service.py

@@ -17,8 +17,15 @@
 from app.models.participant import Participant
 from app.services import group_chat_service
 from app.services.storage import get_storage_backend, normalize_storage_key
-from app.services.storage_runtime.base import StorageEntry, StorageVersion, WriteCondition
+from app.services.storage_runtime.base import (
+    StorageEntry,
+    StorageVersion,
+    WriteCondition,
+    content_hash_bytes,
+)
 from app.services.workspace_collaboration import (
+    BINARY_REVISION_EXTENSIONS,
+    MAX_REVISION_TEXT_BYTES,
     content_hash,
     finalize_group_runtime_revision,
     get_group_runtime_revision,
@@ -48,6 +55,17 @@ class GroupTextFile:
     revision_id: uuid.UUID | None = None
 
 
+@dataclass(frozen=True, slots=True)
+class GroupBinaryFile:
+    """Business-level view of one group workspace binary file."""
+
+    path: str
+    content: bytes
+    version_token: str
+    modified_at: str | None
+    revision_id: uuid.UUID | None = None
+
+
 @dataclass(frozen=True, slots=True)
 class GroupWorkspaceEntry:
     """One immediate child in the group workspace."""
@@ -184,6 +202,23 @@ def _validate_text(content: str) -> str:
     return content
 
 
+async def _revision_text(storage, key: str, business_path: str) -> str | None:
+    """Return exact revision text without decoding binary data into database TEXT."""
+    raw = await storage.read_bytes(key)
+    filename = business_path.rsplit("/", 1)[-1]
+    suffix = "." + filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
+    if (
+        suffix in BINARY_REVISION_EXTENSIONS
+        or len(raw) > MAX_REVISION_TEXT_BYTES
+        or b"\x00" in raw
+    ):
+        return None
+    try:
+        return raw.decode("utf-8")
+    except UnicodeDecodeError:
+        return None
+
+
 def _runtime_revision_path(path: str) -> str:
     return _revision_path("workspace", path)
 
@@ -548,7 +583,7 @@ async def _write_text(
             "Group file already exists at this path",
         )
     before = (
-        await storage.read_text(key, encoding="utf-8", errors="replace")
+        await _revision_text(storage, key, business_path)
         if current.exists and not current.is_dir
         else None
     )
@@ -607,7 +642,7 @@ async def _delete_text(
     current = await storage.get_version(key)
     if not current.exists or current.is_dir:
         raise GroupFileServiceError("group_file_not_found", "Group file not found")
-    before = await storage.read_text(key, encoding="utf-8", errors="replace")
+    before = await _revision_text(storage, key, revision_path)
     result = await storage.delete_if_match(
         key,
         condition=(
@@ -912,6 +947,36 @@ async def read_workspace_file(
     )
 
 
+async def read_workspace_binary_file(
+    db: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    group_id: uuid.UUID,
+    actor_participant_id: uuid.UUID,
+    path: str,
+) -> GroupBinaryFile:
+    """Read one group workspace file as exact bytes."""
+    await _authorize_actor(
+        db,
+        tenant_id=tenant_id,
+        group_id=group_id,
+        actor_participant_id=actor_participant_id,
+    )
+    normalized, key = _workspace_key(group_id, path, allow_empty=False)
+    storage = get_storage_backend()
+    version = await storage.get_version(key)
+    if not version.exists:
+        raise GroupFileServiceError("group_file_not_found", "Group file not found")
+    if version.is_dir:
+        raise GroupFileServiceError("group_file_not_readable", "Path is a directory")
+    return GroupBinaryFile(
+        path=normalized,
+        content=await storage.read_bytes(key),
+        version_token=version.token,
+        modified_at=version.modified_at or None,
+    )
+
+
 async def write_workspace_file(
     db: AsyncSession,
     *,
@@ -946,6 +1011,82 @@ async def write_workspace_file(
     )
 
 
+async def write_workspace_binary_file(
+    db: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    group_id: uuid.UUID,
+    actor_participant_id: uuid.UUID,
+    path: str,
+    content: bytes,
+    content_type: str,
+    expected_version_token: str | None = None,
+    require_absent: bool = False,
+    session_id: uuid.UUID | None = None,
+) -> GroupBinaryFile:
+    """Create or replace one group workspace file without text transcoding."""
+    actor = await _authorize_actor(
+        db,
+        tenant_id=tenant_id,
+        group_id=group_id,
+        actor_participant_id=actor_participant_id,
+    )
+    normalized, key = _workspace_key(group_id, path, allow_empty=False)
+    if require_absent and expected_version_token is not None:
+        raise GroupFileServiceError(
+            "group_file_write_condition_invalid",
+            "A create-only write cannot also provide a version token",
+        )
+    storage = get_storage_backend()
+    current = await storage.get_version(key)
+    if current.is_dir:
+        raise GroupFileServiceError("group_file_conflict", "Group path is a directory")
+    before = (
+        await _revision_text(storage, key, normalized)
+        if current.exists and not current.is_dir
+        else None
+    )
+    result = await storage.write_bytes_if_match(
+        key,
+        content,
+        condition=(
+            WriteCondition(require_absent=True)
+            if require_absent
+            else (
+                WriteCondition(version_token=expected_version_token)
+                if expected_version_token is not None
+                else None
+            )
+        ),
+        content_type=content_type,
+    )
+    if not result.ok:
+        raise GroupFileServiceError(
+            "group_file_conflict",
+            "Group file changed before this write completed",
+        )
+    revision = await record_group_revision(
+        db,
+        group_id=group_id,
+        path=_revision_path("workspace", normalized),
+        operation="write",
+        actor_type=actor.type,
+        actor_id=actor.ref_id,
+        before_content=before,
+        after_content=None,
+        content_hash_override=content_hash_bytes(content),
+        session_id=str(session_id) if session_id is not None else None,
+    )
+    updated = result.current_version or await storage.get_version(key)
+    return GroupBinaryFile(
+        path=normalized,
+        content=content,
+        version_token=updated.token,
+        modified_at=updated.modified_at or None,
+        revision_id=revision.id if revision is not None else None,
+    )
+
+
 async def _delete_empty_workspace_directory(
     db: AsyncSession,
     *,
@@ -1069,6 +1210,7 @@ async def delete_workspace_file(
 
 __all__ = [
     "GroupFileServiceError",
+    "GroupBinaryFile",
     "GroupTextFile",
     "GroupWorkspaceEntry",
     "delete_agent_memory",
@@ -1078,7 +1220,9 @@ async def delete_workspace_file(
     "read_agent_memory",
     "read_announcement",
     "read_workspace_file",
+    "read_workspace_binary_file",
     "write_agent_memory",
     "write_announcement",
     "write_workspace_file",
+    "write_workspace_binary_file",
 ]