Add artifact REST endpoints (list, get, download, restore)

Serve artifact metadata and bytes over HTTP with parent-derived
authorization: conversation-parented artifacts inherit conversation
access (owner, shared_with, or a public share token whose conversation
matches the parent), workflow-run artifacts check run ownership, and
access fails closed when the parent is missing or deleted.

Adds list/get/versions/download/restore routes, an authenticated
storage-agnostic download (sanitized Content-Disposition, 302 to a
short-lived private S3 presigned URL when that strategy is configured),
a generate_presigned_url primitive on the storage base and S3 backend,
and generalizes the tools artifact endpoint for documents/files. Shared
authorization helpers live in a dedicated module used by both surfaces.

application/api/user/artifacts/__init__.py

@@ -0,0 +1,5 @@
+"""Artifacts module."""
+
+from .routes import artifacts_ns
+
+__all__ = ["artifacts_ns"]

application/api/user/artifacts/authz.py

@@ -0,0 +1,68 @@
+"""Parent-derived authorization helpers for artifact access."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from flask import request
+
+from application.storage.db.repositories.agents import AgentsRepository
+from application.storage.db.repositories.conversations import ConversationsRepository
+from application.storage.db.repositories.shared_conversations import (
+    SharedConversationsRepository,
+)
+from application.storage.db.repositories.workflow_runs import WorkflowRunsRepository
+from application.storage.db.session import db_readonly
+
+
+def resolve_authenticated_user() -> Optional[str]:
+    """Resolve the caller to a user id via decoded JWT ``sub`` or api_key→owner."""
+    decoded_token = getattr(request, "decoded_token", None)
+    if decoded_token:
+        return decoded_token.get("sub")
+
+    api_key = request.args.get("api_key") or request.form.get("api_key")
+    if api_key:
+        with db_readonly() as conn:
+            agent = AgentsRepository(conn).find_by_key(api_key)
+        if agent:
+            return agent.get("user_id")
+    return None
+
+
+def user_can_access_conversation(
+    conn, conversation_id: str, user_id: Optional[str], share_token: Optional[str]
+) -> bool:
+    """Allow if the caller owns/shares the conversation, or holds a valid share token.
+
+    Reuses ``ConversationsRepository.get`` (owner OR ``shared_with``) so artifact
+    access tracks message access; a publicly shared link inherits download access
+    via its share token (see ``SharedConversationsRepository.find_by_uuid``).
+    """
+    if user_id:
+        if ConversationsRepository(conn).get(conversation_id, user_id) is not None:
+            return True
+    if share_token:
+        shared = SharedConversationsRepository(conn).find_by_uuid(share_token)
+        if shared and str(shared.get("conversation_id")) == str(conversation_id):
+            return True
+    return False
+
+
+def authorize_artifact(conn, artifact: dict, user_id: Optional[str]) -> bool:
+    """Authorize an artifact by resolving its parent; missing parent fails closed."""
+    conversation_id = artifact.get("conversation_id")
+    workflow_run_id = artifact.get("workflow_run_id")
+    share_token = request.args.get("share_token")
+
+    if conversation_id is not None:
+        return user_can_access_conversation(
+            conn, str(conversation_id), user_id, share_token
+        )
+    if workflow_run_id is not None:
+        if not user_id:
+            return False
+        run = WorkflowRunsRepository(conn).get(str(workflow_run_id))
+        return run is not None and run.get("user_id") == user_id
+    # No parent row reachable -> deny (e.g. deleted conversation/run).
+    return False

application/api/user/artifacts/routes.py

@@ -0,0 +1,372 @@
+"""Artifact metadata and download routes (parent-derived authz)."""
+
+from __future__ import annotations
+
+import re
+from typing import Optional
+
+from flask import current_app, jsonify, make_response, redirect, request
+from flask_restx import Namespace, Resource
+
+from application.api import api
+from application.api.user.artifacts.authz import (
+    authorize_artifact,
+    resolve_authenticated_user,
+    user_can_access_conversation,
+)
+from application.core.settings import settings
+from application.storage.db.base_repository import looks_like_uuid
+from application.storage.db.repositories.artifacts import ArtifactsRepository
+from application.storage.db.repositories.workflow_runs import WorkflowRunsRepository
+from application.storage.db.session import db_readonly, db_session
+from application.storage.storage_creator import StorageCreator
+
+artifacts_ns = Namespace("artifacts", description="Artifact operations", path="/api")
+
+# Presigned-URL TTL for private S3 artifact downloads (seconds).
+_PRESIGNED_URL_TTL = 300
+
+
+def _sanitize_header_filename(filename: Optional[str], fallback: str) -> str:
+    """Strip CRLF / quotes from a display filename for a Content-Disposition header."""
+    if not filename:
+        return fallback
+    cleaned = re.sub(r'[\r\n"]', "", str(filename)).strip()
+    return cleaned or fallback
+
+
+def _artifact_summary(artifact: dict) -> dict:
+    """Project an artifact identity row to its API metadata shape (owner id withheld)."""
+    return {
+        "id": str(artifact.get("id")),
+        "conversation_id": (
+            str(artifact["conversation_id"])
+            if artifact.get("conversation_id") is not None
+            else None
+        ),
+        "workflow_run_id": (
+            str(artifact["workflow_run_id"])
+            if artifact.get("workflow_run_id") is not None
+            else None
+        ),
+        "message_id": (
+            str(artifact["message_id"])
+            if artifact.get("message_id") is not None
+            else None
+        ),
+        "kind": artifact.get("kind"),
+        "title": artifact.get("title"),
+        "metadata": artifact.get("metadata"),
+        "current_version": artifact.get("current_version"),
+        "created_at": _iso(artifact.get("created_at")),
+        "updated_at": _iso(artifact.get("updated_at")),
+    }
+
+
+def _version_summary(version: dict, *, include_spec: bool = False) -> dict:
+    """Project a version row to its API metadata shape (storage_path withheld)."""
+    out = {
+        "version": version.get("version"),
+        "mime_type": version.get("mime_type"),
+        "filename": version.get("filename"),
+        "size": version.get("size"),
+        "sha256": version.get("sha256"),
+        "preview_text": version.get("preview_text"),
+        "produced_by": version.get("produced_by"),
+        "created_at": _iso(version.get("created_at")),
+    }
+    if include_spec:
+        out["spec"] = version.get("spec")
+    return out
+
+
+def _iso(value):
+    """ISO-format a datetime, passing through other values unchanged."""
+    return value.isoformat() if hasattr(value, "isoformat") else value
+
+
+@artifacts_ns.route("/artifacts")
+class ListArtifacts(Resource):
+    @api.doc(description="List artifacts for a conversation, workflow run, or the caller")
+    def get(self):
+        user_id = resolve_authenticated_user()
+        conversation_id = request.args.get("conversation_id")
+        workflow_run_id = request.args.get("workflow_run_id")
+        share_token = request.args.get("share_token")
+
+        if not user_id and not share_token:
+            return make_response(
+                jsonify({"success": False, "message": "Authentication required"}), 401
+            )
+
+        # Gate UUID-shape before any CAST(:id AS uuid) reaches the repo, so a
+        # malformed id is rejected cleanly instead of poisoning the transaction.
+        if conversation_id and not looks_like_uuid(conversation_id):
+            return make_response(
+                jsonify({"success": False, "message": "Invalid conversation_id"}), 400
+            )
+        if workflow_run_id and not looks_like_uuid(workflow_run_id):
+            return make_response(
+                jsonify({"success": False, "message": "Invalid workflow_run_id"}), 400
+            )
+
+        try:
+            with db_readonly() as conn:
+                if conversation_id:
+                    if not user_can_access_conversation(
+                        conn, conversation_id, user_id, share_token
+                    ):
+                        return make_response(
+                            jsonify({"success": False, "message": "Forbidden"}), 403
+                        )
+                    rows = ArtifactsRepository(conn).list_artifacts(
+                        conversation_id=conversation_id
+                    )
+                elif workflow_run_id:
+                    run = WorkflowRunsRepository(conn).get(workflow_run_id)
+                    if run is None or run.get("user_id") != user_id:
+                        return make_response(
+                            jsonify({"success": False, "message": "Forbidden"}), 403
+                        )
+                    rows = ArtifactsRepository(conn).list_artifacts(
+                        workflow_run_id=workflow_run_id
+                    )
+                else:
+                    if not user_id:
+                        return make_response(
+                            jsonify({"success": False, "message": "Authentication required"}),
+                            401,
+                        )
+                    rows = ArtifactsRepository(conn).list_artifacts(user_id=user_id)
+
+            return make_response(
+                jsonify(
+                    {"success": True, "artifacts": [_artifact_summary(r) for r in rows]}
+                ),
+                200,
+            )
+        except Exception as err:
+            current_app.logger.error(f"Error listing artifacts: {err}", exc_info=True)
+            return make_response(jsonify({"success": False}), 400)
+
+
+@artifacts_ns.route("/artifacts/<artifact_id>")
+class GetArtifact(Resource):
+    @api.doc(description="Get an artifact's metadata, version list, and current spec")
+    def get(self, artifact_id: str):
+        if not looks_like_uuid(artifact_id):
+            return make_response(
+                jsonify({"success": False, "message": "Artifact not found"}), 404
+            )
+        user_id = resolve_authenticated_user()
+        try:
+            with db_readonly() as conn:
+                repo = ArtifactsRepository(conn)
+                artifact = repo.get_artifact(artifact_id)
+                if artifact is None:
+                    return make_response(
+                        jsonify({"success": False, "message": "Artifact not found"}), 404
+                    )
+                if not authorize_artifact(conn, artifact, user_id):
+                    return make_response(
+                        jsonify({"success": False, "message": "Forbidden"}), 403
+                    )
+                versions = repo.list_versions(artifact_id)
+                current = repo.get_version(artifact_id, artifact.get("current_version"))
+
+            payload = _artifact_summary(artifact)
+            payload["versions"] = [_version_summary(v) for v in versions]
+            payload["spec"] = current.get("spec") if current else None
+            return make_response(jsonify({"success": True, "artifact": payload}), 200)
+        except Exception as err:
+            current_app.logger.error(f"Error retrieving artifact: {err}", exc_info=True)
+            return make_response(jsonify({"success": False}), 400)
+
+
+@artifacts_ns.route("/artifacts/<artifact_id>/versions/<int:version>")
+class GetArtifactVersion(Resource):
+    @api.doc(description="Get a single artifact version's metadata and spec")
+    def get(self, artifact_id: str, version: int):
+        if not looks_like_uuid(artifact_id):
+            return make_response(
+                jsonify({"success": False, "message": "Artifact not found"}), 404
+            )
+        user_id = resolve_authenticated_user()
+        try:
+            with db_readonly() as conn:
+                repo = ArtifactsRepository(conn)
+                artifact = repo.get_artifact(artifact_id)
+                if artifact is None:
+                    return make_response(
+                        jsonify({"success": False, "message": "Artifact not found"}), 404
+                    )
+                if not authorize_artifact(conn, artifact, user_id):
+                    return make_response(
+                        jsonify({"success": False, "message": "Forbidden"}), 403
+                    )
+                version_row = repo.get_version(artifact_id, version)
+            if version_row is None:
+                return make_response(
+                    jsonify({"success": False, "message": "Version not found"}), 404
+                )
+            return make_response(
+                jsonify(
+                    {"success": True, "version": _version_summary(version_row, include_spec=True)}
+                ),
+                200,
+            )
+        except Exception as err:
+            current_app.logger.error(
+                f"Error retrieving artifact version: {err}", exc_info=True
+            )
+            return make_response(jsonify({"success": False}), 400)
+
+
+@artifacts_ns.route("/artifacts/<artifact_id>/download")
+class DownloadArtifact(Resource):
+    @api.doc(description="Download an artifact's bytes (302 to a presigned URL on S3)")
+    def get(self, artifact_id: str):
+        if not looks_like_uuid(artifact_id):
+            return make_response(
+                jsonify({"success": False, "message": "Artifact not found"}), 404
+            )
+        user_id = resolve_authenticated_user()
+        version_arg = request.args.get("version")
+        try:
+            with db_readonly() as conn:
+                repo = ArtifactsRepository(conn)
+                artifact = repo.get_artifact(artifact_id)
+                if artifact is None:
+                    return make_response(
+                        jsonify({"success": False, "message": "Artifact not found"}), 404
+                    )
+                if not authorize_artifact(conn, artifact, user_id):
+                    return make_response(
+                        jsonify({"success": False, "message": "Forbidden"}), 403
+                    )
+                version = artifact.get("current_version")
+                if version_arg is not None:
+                    try:
+                        version = int(version_arg)
+                    except ValueError:
+                        return make_response(
+                            jsonify({"success": False, "message": "Invalid version"}), 400
+                        )
+                version_row = repo.get_version(artifact_id, version)
+
+            if version_row is None:
+                return make_response(
+                    jsonify({"success": False, "message": "Version not found"}), 404
+                )
+            # The object key is derived only from the stored path, never client input.
+            storage_path = version_row.get("storage_path")
+            if not storage_path:
+                return make_response(
+                    jsonify({"success": False, "message": "No file for this version"}), 404
+                )
+
+            filename = _sanitize_header_filename(
+                version_row.get("filename"), f"artifact-{artifact_id}"
+            )
+            mime_type = version_row.get("mime_type") or "application/octet-stream"
+            storage = StorageCreator.get_storage()
+
+            # With URL_STRATEGY=="s3" the contract is to hand back a presigned
+            # URL. If the active backend can't mint one, that's a config error:
+            # surface a 500 rather than silently proxying bytes from a backend
+            # the operator expected to be off the hot path.
+            if getattr(settings, "URL_STRATEGY", "backend") == "s3":
+                try:
+                    url = storage.generate_presigned_url(
+                        storage_path, expires_in=_PRESIGNED_URL_TTL
+                    )
+                except NotImplementedError:
+                    current_app.logger.error(
+                        "URL_STRATEGY=s3 but %s cannot mint presigned URLs",
+                        type(storage).__name__,
+                    )
+                    return make_response(
+                        jsonify(
+                            {"success": False, "message": "Storage misconfigured"}
+                        ),
+                        500,
+                    )
+                return redirect(url, code=302)
+
+            file_obj = storage.get_file(storage_path)
+            response = make_response(file_obj.read())
+            response.headers.set("Content-Type", mime_type)
+            response.headers.set(
+                "Content-Disposition", f'attachment; filename="{filename}"'
+            )
+            return response
+        except FileNotFoundError:
+            return make_response(
+                jsonify({"success": False, "message": "File not found"}), 404
+            )
+        except Exception as err:
+            current_app.logger.error(
+                f"Error downloading artifact: {err}", exc_info=True
+            )
+            return make_response(jsonify({"success": False}), 400)
+
+
+@artifacts_ns.route("/artifacts/<artifact_id>/restore")
+class RestoreArtifact(Resource):
+    @api.doc(description="Restore a prior version by appending it as the new current version")
+    def post(self, artifact_id: str):
+        if not looks_like_uuid(artifact_id):
+            return make_response(
+                jsonify({"success": False, "message": "Artifact not found"}), 404
+            )
+        user_id = resolve_authenticated_user()
+        data = request.get_json(silent=True) or {}
+        target_version = data.get("version")
+        if target_version is None:
+            return make_response(
+                jsonify({"success": False, "message": "Missing version"}), 400
+            )
+        try:
+            target_version = int(target_version)
+        except (ValueError, TypeError):
+            return make_response(
+                jsonify({"success": False, "message": "Invalid version"}), 400
+            )
+
+        try:
+            with db_session() as conn:
+                repo = ArtifactsRepository(conn)
+                artifact = repo.get_artifact(artifact_id)
+                if artifact is None:
+                    return make_response(
+                        jsonify({"success": False, "message": "Artifact not found"}), 404
+                    )
+                if not authorize_artifact(conn, artifact, user_id):
+                    return make_response(
+                        jsonify({"success": False, "message": "Forbidden"}), 403
+                    )
+                source = repo.get_version(artifact_id, target_version)
+                if source is None:
+                    return make_response(
+                        jsonify({"success": False, "message": "Version not found"}), 404
+                    )
+                new_version = repo.append_version(
+                    artifact_id,
+                    mime_type=source.get("mime_type"),
+                    filename=source.get("filename"),
+                    storage_path=source.get("storage_path"),
+                    size=source.get("size"),
+                    sha256=source.get("sha256"),
+                    spec=source.get("spec"),
+                    preview_text=source.get("preview_text"),
+                    produced_by=source.get("produced_by"),
+                )
+            return make_response(
+                jsonify(
+                    {"success": True, "version": _version_summary(new_version, include_spec=True)}
+                ),
+                200,
+            )
+        except Exception as err:
+            current_app.logger.error(f"Error restoring artifact: {err}", exc_info=True)
+            return make_response(jsonify({"success": False}), 400)

application/api/user/routes.py

@@ -13,6 +13,7 @@ from .agents import (
     agents_webhooks_ns,
 )
 from .analytics import analytics_ns
+from .artifacts import artifacts_ns
 from .attachments import attachments_ns
 from .conversations import conversations_ns
 from .me import me_ns
@@ -31,6 +32,9 @@ user = Blueprint("user", __name__)
 # Analytics
 api.add_namespace(analytics_ns)
 
+# Artifacts
+api.add_namespace(artifacts_ns)
+
 # Attachments
 api.add_namespace(attachments_ns)
 

application/api/user/tools/routes.py

@@ -15,9 +15,12 @@ from application.agents.default_tools import (
 from application.agents.tools.spec_parser import parse_spec
 from application.agents.tools.tool_manager import ToolManager
 from application.api import api
+from application.api.user.artifacts.authz import authorize_artifact
 from application.api.user.team_sharing import effective_write_owner, visible_with_access
 from application.core.url_validation import SSRFError, validate_url
 from application.security.encryption import decrypt_credentials, encrypt_credentials
+from application.storage.db.base_repository import looks_like_uuid
+from application.storage.db.repositories.artifacts import ArtifactsRepository
 from application.storage.db.repositories.notes import NotesRepository
 from application.storage.db.repositories.todos import TodosRepository
 from application.storage.db.repositories.user_tools import UserToolsRepository
@@ -934,6 +937,44 @@ class GetArtifact(Resource):
                     return make_response(
                         jsonify({"success": True, "artifact": artifact}), 200
                     )
+
+                # Generalized document/file artifacts live in the
+                # ``artifacts`` store, authorized by their parent (not user_id).
+                # Shape-gate the UUID lookup so a non-UUID id (legacy note/todo
+                # ObjectId already handled above) never reaches the CAST that
+                # would poison this read-only transaction.
+                artifact_doc = None
+                if looks_like_uuid(artifact_id):
+                    artifacts_repo = ArtifactsRepository(conn)
+                    artifact_doc = artifacts_repo.get_artifact(artifact_id)
+                if artifact_doc and authorize_artifact(conn, artifact_doc, user_id):
+                    current = artifacts_repo.get_version(
+                        artifact_id, artifact_doc.get("current_version")
+                    )
+                    artifact = {
+                        "artifact_type": (
+                            "document"
+                            if artifact_doc.get("kind") != "file"
+                            else "file"
+                        ),
+                        "data": {
+                            "id": str(artifact_doc.get("id")),
+                            "kind": artifact_doc.get("kind"),
+                            "title": artifact_doc.get("title"),
+                            "current_version": artifact_doc.get("current_version"),
+                            "mime_type": current.get("mime_type") if current else None,
+                            "filename": current.get("filename") if current else None,
+                            "size": current.get("size") if current else None,
+                            "download_url": (
+                                f"/api/artifacts/{artifact_id}/download"
+                                if current and current.get("storage_path")
+                                else None
+                            ),
+                        },
+                    }
+                    return make_response(
+                        jsonify({"success": True, "artifact": artifact}), 200
+                    )
         except Exception as err:
             current_app.logger.error(
                 f"Error retrieving artifact: {err}", exc_info=True

application/storage/base.py

@@ -37,6 +37,23 @@ class BaseStorage(ABC):
         """
         pass
 
+    def generate_presigned_url(self, path: str, expires_in: int = 300) -> str:
+        """Return a short-lived presigned download URL; not all backends support it.
+
+        Args:
+            path: Path to the file
+            expires_in: TTL of the signed URL in seconds
+
+        Returns:
+            str: A presigned URL granting time-limited read access
+
+        Raises:
+            NotImplementedError: If the backend cannot mint presigned URLs.
+        """
+        raise NotImplementedError(
+            f"{type(self).__name__} does not support presigned URLs"
+        )
+
     @abstractmethod
     def process_file(self, path: str, processor_func: Callable, **kwargs):
         """

application/storage/s3.py

@@ -121,6 +121,16 @@ class S3Storage(BaseStorage):
         file_obj.seek(0)
         return file_obj
 
+    def generate_presigned_url(self, path: str, expires_in: int = 300) -> str:
+        """Return a short-lived presigned GET URL for a private object (TTL <= 1h)."""
+        path = self._validate_path(path)
+        expires_in = min(expires_in, 3600)
+        return self.s3.generate_presigned_url(
+            "get_object",
+            Params={"Bucket": self.bucket_name, "Key": path},
+            ExpiresIn=expires_in,
+        )
+
     def delete_file(self, path: str) -> bool:
         """Delete a file from S3 storage."""
         path = self._validate_path(path)

tests/api/user/test_artifacts_routes.py

@@ -0,0 +1,564 @@
+"""Route-layer tests for the artifacts API (parent-derived authz)."""
+
+from __future__ import annotations
+
+import io
+import uuid
+from contextlib import contextmanager
+
+import pytest
+from flask import request
+from sqlalchemy import text
+
+from application.storage.db.repositories.artifacts import ArtifactsRepository
+from application.storage.db.repositories.conversations import ConversationsRepository
+from application.storage.db.repositories.shared_conversations import (
+    SharedConversationsRepository,
+)
+from application.storage.db.repositories.workflow_runs import WorkflowRunsRepository
+
+
+OWNER = "artifact_owner"
+STRANGER = "artifact_stranger"
+SHARED_USER = "artifact_shared_user"
+
+
+@pytest.fixture
+def _patch_db(pg_conn, monkeypatch):
+    """Redirect the routes' ``db_readonly`` / ``db_session`` to the test conn."""
+
+    @contextmanager
+    def _use_conn():
+        yield pg_conn
+
+    monkeypatch.setattr("application.api.user.artifacts.routes.db_readonly", _use_conn)
+    monkeypatch.setattr("application.api.user.artifacts.routes.db_session", _use_conn)
+    return pg_conn
+
+
+@pytest.fixture
+def token_owner():
+    return {"sub": OWNER, "email": "owner@example.com"}
+
+
+def _make_conversation(conn, user_id=OWNER):
+    return ConversationsRepository(conn).create(user_id, name="conv")
+
+
+def _make_workflow(conn, user_id=OWNER):
+    res = conn.execute(
+        text("INSERT INTO workflows (user_id, name) VALUES (:u, 'wf') RETURNING id"),
+        {"u": user_id},
+    )
+    return str(res.fetchone()[0])
+
+
+def _make_artifact(conn, **kwargs):
+    return ArtifactsRepository(conn).create_artifact(
+        kwargs.pop("user_id", OWNER),
+        kwargs.pop("kind", "document"),
+        **kwargs,
+    )
+
+
+def _call(flask_app, resource_cls, *args, token=None, query=None, json_body=None, method="get"):
+    with flask_app.app_context():
+        with flask_app.test_request_context(query_string=query or {}, json=json_body):
+            request.decoded_token = token
+            resource = resource_cls()
+            return getattr(resource, method)(*args)
+
+
+# ---------------------------------------------------------------------------
+# List
+# ---------------------------------------------------------------------------
+@pytest.mark.unit
+class TestListArtifacts:
+    def test_owner_lists_conversation_artifacts(
+        self, _patch_db, flask_app, token_owner
+    ):
+        from application.api.user.artifacts.routes import ListArtifacts
+
+        conv = _make_conversation(_patch_db)
+        conv_id = str(conv["id"])
+        _make_artifact(_patch_db, conversation_id=conv_id, title="a1")
+        _make_artifact(_patch_db, conversation_id=conv_id, title="a2")
+
+        resp = _call(
+            flask_app, ListArtifacts, token=token_owner,
+            query={"conversation_id": conv_id},
+        )
+        assert resp.status_code == 200
+        assert len(resp.json["artifacts"]) == 2
+
+    def test_stranger_denied_conversation_list(
+        self, _patch_db, flask_app
+    ):
+        from application.api.user.artifacts.routes import ListArtifacts
+
+        conv = _make_conversation(_patch_db)
+        _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+
+        resp = _call(
+            flask_app, ListArtifacts,
+            token={"sub": STRANGER},
+            query={"conversation_id": str(conv["id"])},
+        )
+        assert resp.status_code == 403
+
+    def test_no_filter_scopes_to_user(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import ListArtifacts
+
+        owner_conv = _make_conversation(_patch_db)
+        owned = _make_artifact(
+            _patch_db, conversation_id=str(owner_conv["id"]), user_id=OWNER
+        )
+        stranger = _make_artifact(
+            _patch_db, conversation_id=str(_make_conversation(_patch_db)["id"]),
+            user_id=STRANGER,
+        )
+
+        resp = _call(flask_app, ListArtifacts, token=token_owner)
+        assert resp.status_code == 200
+        # The owner id is accounting-only and must not leak in the summary.
+        assert all("user_id" not in a for a in resp.json["artifacts"])
+        returned_ids = {a["id"] for a in resp.json["artifacts"]}
+        assert str(owned["id"]) in returned_ids
+        assert str(stranger["id"]) not in returned_ids
+
+    def test_unauthenticated_401(self, _patch_db, flask_app):
+        from application.api.user.artifacts.routes import ListArtifacts
+
+        resp = _call(flask_app, ListArtifacts, token=None)
+        assert resp.status_code == 401
+
+    def test_non_uuid_conversation_id_400(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import ListArtifacts
+
+        # A malformed id must be rejected before reaching CAST(:id AS uuid),
+        # which would otherwise raise a DataError and poison the transaction.
+        resp = _call(
+            flask_app, ListArtifacts, token=token_owner,
+            query={"conversation_id": "not-a-uuid"},
+        )
+        assert resp.status_code == 400
+
+    def test_non_uuid_workflow_run_id_400(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import ListArtifacts
+
+        resp = _call(
+            flask_app, ListArtifacts, token=token_owner,
+            query={"workflow_run_id": "1234"},
+        )
+        assert resp.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Get + versions
+# ---------------------------------------------------------------------------
+@pytest.mark.unit
+class TestGetArtifact:
+    def test_owner_gets_artifact_with_versions(
+        self, _patch_db, flask_app, token_owner
+    ):
+        from application.api.user.artifacts.routes import GetArtifact
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(
+            _patch_db, conversation_id=str(conv["id"]), spec={"body": "v1"}
+        )
+        ArtifactsRepository(_patch_db).append_version(art["id"], spec={"body": "v2"})
+
+        resp = _call(flask_app, GetArtifact, art["id"], token=token_owner)
+        assert resp.status_code == 200
+        assert resp.json["artifact"]["current_version"] == 2
+        assert len(resp.json["artifact"]["versions"]) == 2
+        assert resp.json["artifact"]["spec"] == {"body": "v2"}
+
+    def test_stranger_denied_403(self, _patch_db, flask_app):
+        from application.api.user.artifacts.routes import GetArtifact
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+
+        resp = _call(flask_app, GetArtifact, art["id"], token={"sub": STRANGER})
+        assert resp.status_code == 403
+
+    def test_missing_parent_fails_closed(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import GetArtifact
+        from sqlalchemy import text
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+        # Delete the parent conversation -> the artifact must no longer be reachable.
+        _patch_db.execute(
+            text("DELETE FROM conversations WHERE id = CAST(:id AS uuid)"),
+            {"id": str(conv["id"])},
+        )
+
+        resp = _call(flask_app, GetArtifact, art["id"], token=token_owner)
+        assert resp.status_code == 403
+
+    def test_unknown_artifact_404(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import GetArtifact
+
+        resp = _call(flask_app, GetArtifact, str(uuid.uuid4()), token=token_owner)
+        assert resp.status_code == 404
+
+    def test_workflow_run_owner_access(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import GetArtifact
+
+        wf_id = _make_workflow(_patch_db, OWNER)
+        run = WorkflowRunsRepository(_patch_db).create(wf_id, OWNER, "completed")
+        art = _make_artifact(_patch_db, workflow_run_id=str(run["id"]))
+
+        resp = _call(flask_app, GetArtifact, art["id"], token=token_owner)
+        assert resp.status_code == 200
+
+        resp_other = _call(
+            flask_app, GetArtifact, art["id"], token={"sub": STRANGER}
+        )
+        assert resp_other.status_code == 403
+
+
+@pytest.mark.unit
+class TestGetArtifactVersion:
+    def test_version_returns_spec(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import GetArtifactVersion
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(
+            _patch_db, conversation_id=str(conv["id"]), spec={"body": "one"}
+        )
+        resp = _call(flask_app, GetArtifactVersion, art["id"], 1, token=token_owner)
+        assert resp.status_code == 200
+        assert resp.json["version"]["spec"] == {"body": "one"}
+
+    def test_missing_version_404(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import GetArtifactVersion
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+        resp = _call(flask_app, GetArtifactVersion, art["id"], 99, token=token_owner)
+        assert resp.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# Shared-conversation inheritance
+# ---------------------------------------------------------------------------
+@pytest.mark.unit
+class TestSharedAccess:
+    def test_shared_with_user_can_get(self, _patch_db, flask_app):
+        from application.api.user.artifacts.routes import GetArtifact
+
+        conv = _make_conversation(_patch_db)
+        ConversationsRepository(_patch_db).add_shared_user(str(conv["id"]), SHARED_USER)
+        art = _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+
+        resp = _call(flask_app, GetArtifact, art["id"], token={"sub": SHARED_USER})
+        assert resp.status_code == 200
+
+    def test_share_token_holder_can_download(self, _patch_db, flask_app, monkeypatch):
+        from application.api.user.artifacts.routes import DownloadArtifact
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(
+            _patch_db,
+            conversation_id=str(conv["id"]),
+            filename="report.pdf",
+            mime_type="application/pdf",
+            storage_path="inputs/owner/artifacts/x/v1/report.pdf",
+        )
+        share = SharedConversationsRepository(_patch_db).create(
+            str(conv["id"]), OWNER
+        )
+
+        storage = _FakeStorage(b"PDFDATA")
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.StorageCreator.get_storage",
+            lambda: storage,
+        )
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.settings.URL_STRATEGY",
+            "backend",
+            raising=False,
+        )
+
+        # Anonymous link holder (no JWT) supplies the share token.
+        resp = _call(
+            flask_app, DownloadArtifact, art["id"], token=None,
+            query={"share_token": str(share["uuid"])},
+        )
+        assert resp.status_code == 200
+        assert resp.data == b"PDFDATA"
+
+
+# ---------------------------------------------------------------------------
+# Download
+# ---------------------------------------------------------------------------
+class _FakeStorage:
+    def __init__(self, data: bytes):
+        self._data = data
+
+    def get_file(self, path):
+        return io.BytesIO(self._data)
+
+    def generate_presigned_url(self, path, expires_in=300):
+        return f"https://signed.example/{path}?exp={expires_in}"
+
+
+@pytest.mark.unit
+class TestDownloadArtifact:
+    def _seed(self, conn, **over):
+        conv = _make_conversation(conn)
+        defaults = dict(
+            conversation_id=str(conv["id"]),
+            filename="deck.pptx",
+            mime_type="application/vnd.ms-powerpoint",
+            storage_path="inputs/owner/artifacts/x/v1/deck.pptx",
+        )
+        defaults.update(over)
+        return _make_artifact(conn, **defaults)
+
+    def test_local_streams_bytes_with_content_disposition(
+        self, _patch_db, flask_app, token_owner, monkeypatch
+    ):
+        from application.api.user.artifacts.routes import DownloadArtifact
+
+        art = self._seed(_patch_db)
+        storage = _FakeStorage(b"BINARY")
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.StorageCreator.get_storage",
+            lambda: storage,
+        )
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.settings.URL_STRATEGY",
+            "backend",
+            raising=False,
+        )
+
+        resp = _call(flask_app, DownloadArtifact, art["id"], token=token_owner)
+        assert resp.status_code == 200
+        assert resp.data == b"BINARY"
+        assert resp.headers["Content-Disposition"] == 'attachment; filename="deck.pptx"'
+        assert resp.headers["Content-Type"] == "application/vnd.ms-powerpoint"
+
+    def test_s3_strategy_redirects_to_presigned(
+        self, _patch_db, flask_app, token_owner, monkeypatch
+    ):
+        from application.api.user.artifacts.routes import DownloadArtifact
+
+        art = self._seed(_patch_db)
+        storage = _FakeStorage(b"unused")
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.StorageCreator.get_storage",
+            lambda: storage,
+        )
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.settings.URL_STRATEGY",
+            "s3",
+            raising=False,
+        )
+
+        resp = _call(flask_app, DownloadArtifact, art["id"], token=token_owner)
+        assert resp.status_code == 302
+        assert resp.headers["Location"].startswith("https://signed.example/")
+
+    def test_stranger_denied(self, _patch_db, flask_app, monkeypatch):
+        from application.api.user.artifacts.routes import DownloadArtifact
+
+        art = self._seed(_patch_db)
+        resp = _call(flask_app, DownloadArtifact, art["id"], token={"sub": STRANGER})
+        assert resp.status_code == 403
+
+    def test_s3_strategy_misconfigured_backend_500(
+        self, _patch_db, flask_app, token_owner, monkeypatch
+    ):
+        from application.api.user.artifacts.routes import DownloadArtifact
+
+        art = self._seed(_patch_db)
+
+        class _NoPresignStorage:
+            def get_file(self, path):
+                return io.BytesIO(b"unused")
+
+            def generate_presigned_url(self, path, expires_in=300):
+                raise NotImplementedError("backend cannot mint presigned URLs")
+
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.StorageCreator.get_storage",
+            lambda: _NoPresignStorage(),
+        )
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.settings.URL_STRATEGY",
+            "s3",
+            raising=False,
+        )
+
+        resp = _call(flask_app, DownloadArtifact, art["id"], token=token_owner)
+        assert resp.status_code == 500
+
+    def test_crlf_filename_sanitized_in_header(
+        self, _patch_db, flask_app, token_owner, monkeypatch
+    ):
+        from application.api.user.artifacts.routes import DownloadArtifact
+
+        art = self._seed(_patch_db, filename='a"\r\nInjected: x.txt')
+        storage = _FakeStorage(b"X")
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.StorageCreator.get_storage",
+            lambda: storage,
+        )
+        monkeypatch.setattr(
+            "application.api.user.artifacts.routes.settings.URL_STRATEGY",
+            "backend",
+            raising=False,
+        )
+        resp = _call(flask_app, DownloadArtifact, art["id"], token=token_owner)
+        disposition = resp.headers["Content-Disposition"]
+        assert "\r" not in disposition and "\n" not in disposition
+        assert disposition == 'attachment; filename="aInjected: x.txt"'
+
+
+# ---------------------------------------------------------------------------
+# Restore
+# ---------------------------------------------------------------------------
+@pytest.mark.unit
+class TestRestoreArtifact:
+    def test_restore_appends_new_version(self, _patch_db, flask_app, token_owner):
+        from application.api.user.artifacts.routes import RestoreArtifact
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(
+            _patch_db, conversation_id=str(conv["id"]), spec={"body": "v1"}
+        )
+        ArtifactsRepository(_patch_db).append_version(art["id"], spec={"body": "v2"})
+
+        resp = _call(
+            flask_app, RestoreArtifact, art["id"], token=token_owner,
+            json_body={"version": 1}, method="post",
+        )
+        assert resp.status_code == 200
+        assert resp.json["version"]["version"] == 3
+        assert resp.json["version"]["spec"] == {"body": "v1"}
+        assert ArtifactsRepository(_patch_db).get_artifact(art["id"])["current_version"] == 3
+
+    def test_restore_missing_version_field_400(
+        self, _patch_db, flask_app, token_owner
+    ):
+        from application.api.user.artifacts.routes import RestoreArtifact
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+        resp = _call(
+            flask_app, RestoreArtifact, art["id"], token=token_owner,
+            json_body={}, method="post",
+        )
+        assert resp.status_code == 400
+
+    def test_restore_stranger_denied(self, _patch_db, flask_app):
+        from application.api.user.artifacts.routes import RestoreArtifact
+
+        conv = _make_conversation(_patch_db)
+        art = _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+        resp = _call(
+            flask_app, RestoreArtifact, art["id"], token={"sub": STRANGER},
+            json_body={"version": 1}, method="post",
+        )
+        assert resp.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# Malformed artifact_id path segment -> 404 (no CAST(:id AS uuid) txn poison)
+# ---------------------------------------------------------------------------
+@pytest.mark.unit
+class TestMalformedArtifactId:
+    @pytest.mark.parametrize(
+        "resource_name, extra_args, method, json_body",
+        [
+            ("GetArtifact", (), "get", None),
+            ("GetArtifactVersion", (1,), "get", None),
+            ("DownloadArtifact", (), "get", None),
+            ("RestoreArtifact", (), "post", {"version": 1}),
+        ],
+    )
+    def test_non_uuid_id_returns_404(
+        self, _patch_db, flask_app, token_owner, resource_name,
+        extra_args, method, json_body,
+    ):
+        import application.api.user.artifacts.routes as routes_mod
+
+        resource_cls = getattr(routes_mod, resource_name)
+        resp = _call(
+            flask_app, resource_cls, "legacy-mongo-objectid-aabbcc", *extra_args,
+            token=token_owner, method=method, json_body=json_body,
+        )
+        assert resp.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# api_key -> agent-owner principal resolution
+# ---------------------------------------------------------------------------
+@pytest.mark.unit
+class TestApiKeyPrincipal:
+    def test_api_key_owner_can_get_owned_artifact(
+        self, _patch_db, flask_app, monkeypatch
+    ):
+        from application.api.user.artifacts import authz
+        from application.api.user.artifacts.routes import GetArtifact
+        from application.storage.db.repositories.agents import AgentsRepository
+
+        conv = _make_conversation(_patch_db, user_id=OWNER)
+        art = _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+
+        # The api_key path opens its own readonly conn inside authz; point it at
+        # the test conn and resolve the key to the artifact's owning agent.
+        @contextmanager
+        def _use_conn():
+            yield _patch_db
+
+        monkeypatch.setattr(authz, "db_readonly", _use_conn)
+        monkeypatch.setattr(
+            AgentsRepository, "find_by_key",
+            lambda self, key: {"user_id": OWNER} if key == "secret-key" else None,
+        )
+
+        # No JWT -> principal must be resolved from the api_key query param.
+        resp = _call(
+            flask_app, GetArtifact, art["id"], token=None,
+            query={"api_key": "secret-key"},
+        )
+        assert resp.status_code == 200
+        assert resp.json["artifact"]["id"] == str(art["id"])
+
+    def test_api_key_resolving_to_stranger_denied(
+        self, _patch_db, flask_app, monkeypatch
+    ):
+        from application.api.user.artifacts import authz
+        from application.api.user.artifacts.routes import GetArtifact
+        from application.storage.db.repositories.agents import AgentsRepository
+
+        conv = _make_conversation(_patch_db, user_id=OWNER)
+        art = _make_artifact(_patch_db, conversation_id=str(conv["id"]))
+
+        @contextmanager
+        def _use_conn():
+            yield _patch_db
+
+        monkeypatch.setattr(authz, "db_readonly", _use_conn)
+        monkeypatch.setattr(
+            AgentsRepository, "find_by_key",
+            lambda self, key: {"user_id": STRANGER},
+        )
+
+        resp = _call(
+            flask_app, GetArtifact, art["id"], token=None,
+            query={"api_key": "stranger-key"},
+        )
+        assert resp.status_code == 403
+
+
+# A full-app integration test (namespace registration + auth before_request)
+# is covered by the e2e suite under tests/e2e; the unit tests above exercise
+# the malformed-id gate and the api_key principal path directly against the
+# Resource handlers without building the whole app.

tests/api/user/test_tools_artifact_generalized.py

@@ -0,0 +1,107 @@
+"""Tests for the generalized GET /api/tools/artifact/<id> document branch."""
+
+from __future__ import annotations
+
+import uuid
+from contextlib import contextmanager
+
+import pytest
+from flask import request
+
+from application.storage.db.repositories.artifacts import ArtifactsRepository
+from application.storage.db.repositories.conversations import ConversationsRepository
+from application.storage.db.repositories.notes import NotesRepository
+from application.storage.db.repositories.user_tools import UserToolsRepository
+
+
+@pytest.fixture
+def _patch_db(pg_conn, monkeypatch):
+    """Point both the tools and artifacts route DB helpers at the test conn."""
+
+    @contextmanager
+    def _use_conn():
+        yield pg_conn
+
+    monkeypatch.setattr("application.api.user.tools.routes.db_readonly", _use_conn)
+    monkeypatch.setattr("application.api.user.artifacts.routes.db_readonly", _use_conn)
+    return pg_conn
+
+
+def _get(flask_app, artifact_id, token):
+    from application.api.user.tools.routes import GetArtifact
+
+    with flask_app.app_context():
+        with flask_app.test_request_context():
+            request.decoded_token = token
+            return GetArtifact().get(artifact_id)
+
+
+@pytest.mark.unit
+class TestGeneralizedToolsArtifact:
+    def test_document_branch_returns_metadata(
+        self, _patch_db, flask_app, decoded_token
+    ):
+        conv = ConversationsRepository(_patch_db).create(
+            decoded_token["sub"], name="conv"
+        )
+        art = ArtifactsRepository(_patch_db).create_artifact(
+            decoded_token["sub"],
+            "document",
+            conversation_id=str(conv["id"]),
+            title="Report",
+            filename="report.pdf",
+            mime_type="application/pdf",
+            storage_path="inputs/u/artifacts/x/v1/report.pdf",
+        )
+
+        resp = _get(flask_app, art["id"], decoded_token)
+        assert resp.status_code == 200
+        body = resp.json["artifact"]
+        assert body["artifact_type"] == "document"
+        assert body["data"]["title"] == "Report"
+        assert body["data"]["filename"] == "report.pdf"
+        assert body["data"]["download_url"] == f"/api/artifacts/{art['id']}/download"
+
+    def test_file_kind_maps_to_file_type(self, _patch_db, flask_app, decoded_token):
+        conv = ConversationsRepository(_patch_db).create(
+            decoded_token["sub"], name="conv"
+        )
+        art = ArtifactsRepository(_patch_db).create_artifact(
+            decoded_token["sub"],
+            "file",
+            conversation_id=str(conv["id"]),
+            storage_path="inputs/u/artifacts/y/v1/data.bin",
+        )
+        resp = _get(flask_app, art["id"], decoded_token)
+        assert resp.status_code == 200
+        assert resp.json["artifact"]["artifact_type"] == "file"
+
+    def test_stranger_document_not_found(self, _patch_db, flask_app, decoded_token):
+        conv = ConversationsRepository(_patch_db).create("other_owner", name="conv")
+        art = ArtifactsRepository(_patch_db).create_artifact(
+            "other_owner",
+            "document",
+            conversation_id=str(conv["id"]),
+            storage_path="inputs/o/artifacts/z/v1/x.pdf",
+        )
+        # Parent-derived authz denies a stranger -> falls through to 404.
+        resp = _get(flask_app, art["id"], decoded_token)
+        assert resp.status_code == 404
+
+    def test_note_branch_still_works(self, _patch_db, flask_app, decoded_token):
+        tool = UserToolsRepository(_patch_db).create(
+            user_id=decoded_token["sub"], name="notes_tool"
+        )
+        note = NotesRepository(_patch_db).upsert(
+            user_id=decoded_token["sub"],
+            tool_id=str(tool["id"]),
+            title="t",
+            content="a\nb",
+        )
+        resp = _get(flask_app, str(note["id"]), decoded_token)
+        assert resp.status_code == 200
+        assert resp.json["artifact"]["artifact_type"] == "note"
+
+    def test_unknown_id_404(self, _patch_db, flask_app, decoded_token):
+        resp = _get(flask_app, str(uuid.uuid4()), decoded_token)
+        assert resp.status_code == 404