diff --git a/backend-java/src/main/java/io/agentflow/api/config/AgentflowProperties.java b/backend-java/src/main/java/io/agentflow/api/config/AgentflowProperties.java
index fb5dd16..e50b40b 100644
--- a/backend-java/src/main/java/io/agentflow/api/config/AgentflowProperties.java
+++ b/backend-java/src/main/java/io/agentflow/api/config/AgentflowProperties.java
@@ -15,6 +15,7 @@ public class AgentflowProperties {
     private Auth auth = new Auth();
     private Temporal temporal = new Temporal();
     private Retention retention = new Retention();
+    private Attachments attachments = new Attachments();
 
     public String getVersion() {
         return version;
@@ -88,6 +89,14 @@ public void setRetention(Retention retention) {
         this.retention = retention;
     }
 
+    public Attachments getAttachments() {
+        return attachments;
+    }
+
+    public void setAttachments(Attachments attachments) {
+        this.attachments = attachments;
+    }
+
     public static class Auth {
         private boolean enabled = false;
         private List<ApiKeyEntry> keys = List.of();
@@ -453,4 +462,25 @@ public void setPurgeBatchSize(int purgeBatchSize) {
             this.purgeBatchSize = purgeBatchSize;
         }
     }
+
+    public static class Attachments {
+        private String storageDir = "./data/attachments";
+        private long maxBytes = 10L * 1024 * 1024;
+
+        public String getStorageDir() {
+            return storageDir;
+        }
+
+        public void setStorageDir(String storageDir) {
+            this.storageDir = storageDir;
+        }
+
+        public long getMaxBytes() {
+            return maxBytes;
+        }
+
+        public void setMaxBytes(long maxBytes) {
+            this.maxBytes = maxBytes;
+        }
+    }
 }
diff --git a/backend-java/src/main/java/io/agentflow/api/controller/AttachmentsController.java b/backend-java/src/main/java/io/agentflow/api/controller/AttachmentsController.java
new file mode 100644
index 0000000..d5084a8
--- /dev/null
+++ b/backend-java/src/main/java/io/agentflow/api/controller/AttachmentsController.java
@@ -0,0 +1,52 @@
+package io.agentflow.api.controller;
+
+import io.agentflow.api.dto.AttachmentResponse;
+import io.agentflow.api.service.AttachmentService;
+import io.agentflow.api.service.AttachmentService.LoadedAttachment;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.multipart.MultipartFile;
+
+@RestController
+@RequestMapping("/v1/attachments")
+public class AttachmentsController {
+
+    private final AttachmentService service;
+
+    public AttachmentsController(AttachmentService service) {
+        this.service = service;
+    }
+
+    @PostMapping(consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
+    @ResponseStatus(HttpStatus.CREATED)
+    public AttachmentResponse upload(
+            @RequestParam("file") MultipartFile file,
+            @RequestParam(value = "caption", required = false) String caption) {
+        return service.upload(file, caption);
+    }
+
+    @GetMapping("/{id}")
+    public AttachmentResponse meta(@PathVariable String id) {
+        return service.getMeta(id);
+    }
+
+    @GetMapping("/{id}/content")
+    public ResponseEntity<byte[]> content(@PathVariable String id) {
+        LoadedAttachment loaded = service.getContent(id);
+        return ResponseEntity.ok()
+                .header(
+                        HttpHeaders.CONTENT_DISPOSITION,
+                        "inline; filename=\"" + loaded.entity().getFilename() + "\"")
+                .contentType(MediaType.parseMediaType(loaded.entity().getMediaType()))
+                .body(loaded.data());
+    }
+}
diff --git a/backend-java/src/main/java/io/agentflow/api/controller/GlobalExceptionHandler.java b/backend-java/src/main/java/io/agentflow/api/controller/GlobalExceptionHandler.java
index af2d7dd..821d668 100644
--- a/backend-java/src/main/java/io/agentflow/api/controller/GlobalExceptionHandler.java
+++ b/backend-java/src/main/java/io/agentflow/api/controller/GlobalExceptionHandler.java
@@ -3,6 +3,8 @@
 import io.agentflow.api.service.AgentNameConflictException;
 import io.agentflow.api.service.AgentNotFoundException;
 import io.agentflow.api.service.AgentVersionNotFoundException;
+import io.agentflow.api.service.AttachmentNotFoundException;
+import io.agentflow.api.service.AttachmentTooLargeException;
 import io.agentflow.api.service.RegressionExecutionException;
 import io.agentflow.api.service.RunComparisonValidationException;
 import io.agentflow.api.service.RunConflictException;
@@ -39,6 +41,22 @@ public ResponseEntity<Map<String, String>> handleThreadNotFound(ThreadNotFoundEx
         return ResponseEntity.status(HttpStatus.NOT_FOUND).body(Map.of("detail", ex.getMessage()));
     }
 
+    @ExceptionHandler(AttachmentNotFoundException.class)
+    public ResponseEntity<Map<String, String>> handleAttachmentNotFound(
+            AttachmentNotFoundException ex) {
+        return ResponseEntity.status(HttpStatus.NOT_FOUND)
+                .body(Map.of("detail", "Attachment not found: " + ex.getMessage()));
+    }
+
+    @ExceptionHandler(AttachmentTooLargeException.class)
+    public ResponseEntity<Map<String, String>> handleAttachmentTooLarge(
+            AttachmentTooLargeException ex) {
+        return ResponseEntity.status(HttpStatus.PAYLOAD_TOO_LARGE)
+                .body(Map.of(
+                        "detail",
+                        "Attachment too large: " + ex.getSize() + " > " + ex.getLimit() + " bytes"));
+    }
+
     @ExceptionHandler(RunNotFoundException.class)
     public ResponseEntity<Map<String, String>> handleRunNotFound(RunNotFoundException ex) {
         return ResponseEntity.status(HttpStatus.NOT_FOUND).body(Map.of("detail", ex.getMessage()));
diff --git a/backend-java/src/main/java/io/agentflow/api/dto/AttachmentResponse.java b/backend-java/src/main/java/io/agentflow/api/dto/AttachmentResponse.java
new file mode 100644
index 0000000..0593a0c
--- /dev/null
+++ b/backend-java/src/main/java/io/agentflow/api/dto/AttachmentResponse.java
@@ -0,0 +1,85 @@
+package io.agentflow.api.dto;
+
+import io.agentflow.api.entity.AttachmentEntity;
+import java.time.Instant;
+
+public class AttachmentResponse {
+
+    private String id;
+    private String tenantId;
+    private String runId;
+    private String messageId;
+    private String mediaType;
+    private String filename;
+    private long sizeBytes;
+    private String sha256;
+    private String caption;
+    private String url;
+    private Instant createdAt;
+    private Instant updatedAt;
+
+    public static AttachmentResponse fromEntity(AttachmentEntity entity) {
+        AttachmentResponse response = new AttachmentResponse();
+        response.id = entity.getId();
+        response.tenantId = entity.getTenantId();
+        response.runId = entity.getRunId();
+        response.messageId = entity.getMessageId();
+        response.mediaType = entity.getMediaType();
+        response.filename = entity.getFilename();
+        response.sizeBytes = entity.getSizeBytes();
+        response.sha256 = entity.getSha256();
+        response.caption = entity.getCaption();
+        response.url = "/v1/attachments/" + entity.getId();
+        response.createdAt = entity.getCreatedAt();
+        response.updatedAt = entity.getUpdatedAt();
+        return response;
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public String getTenantId() {
+        return tenantId;
+    }
+
+    public String getRunId() {
+        return runId;
+    }
+
+    public String getMessageId() {
+        return messageId;
+    }
+
+    public String getMediaType() {
+        return mediaType;
+    }
+
+    public String getFilename() {
+        return filename;
+    }
+
+    public long getSizeBytes() {
+        return sizeBytes;
+    }
+
+    public String getSha256() {
+        return sha256;
+    }
+
+    public String getCaption() {
+        return caption;
+    }
+
+    public String getUrl() {
+        return url;
+    }
+
+    public Instant getCreatedAt() {
+        return createdAt;
+    }
+
+    public Instant getUpdatedAt() {
+        return updatedAt;
+    }
+}
diff --git a/backend-java/src/main/java/io/agentflow/api/entity/AttachmentEntity.java b/backend-java/src/main/java/io/agentflow/api/entity/AttachmentEntity.java
new file mode 100644
index 0000000..00f6594
--- /dev/null
+++ b/backend-java/src/main/java/io/agentflow/api/entity/AttachmentEntity.java
@@ -0,0 +1,157 @@
+package io.agentflow.api.entity;
+
+import com.github.f4b6a3.ulid.UlidCreator;
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.PrePersist;
+import jakarta.persistence.PreUpdate;
+import jakarta.persistence.Table;
+import java.time.Instant;
+
+@Entity
+@Table(name = "attachments")
+public class AttachmentEntity {
+
+    @Id
+    @Column(length = 26)
+    private String id;
+
+    @Column(name = "tenant_id", nullable = false, length = 64)
+    private String tenantId = "default";
+
+    @Column(name = "run_id", length = 26)
+    private String runId;
+
+    @Column(name = "message_id", length = 26)
+    private String messageId;
+
+    @Column(name = "media_type", nullable = false, length = 128)
+    private String mediaType;
+
+    @Column(nullable = false, length = 512)
+    private String filename;
+
+    @Column(name = "storage_key", nullable = false, length = 512)
+    private String storageKey;
+
+    @Column(name = "size_bytes", nullable = false)
+    private long sizeBytes;
+
+    @Column(nullable = false, length = 64)
+    private String sha256;
+
+    @Column(columnDefinition = "TEXT")
+    private String caption;
+
+    @Column(name = "created_at", nullable = false)
+    private Instant createdAt;
+
+    @Column(name = "updated_at", nullable = false)
+    private Instant updatedAt;
+
+    @PrePersist
+    void onCreate() {
+        if (id == null) {
+            id = UlidCreator.getUlid().toString();
+        }
+        Instant now = Instant.now();
+        if (createdAt == null) {
+            createdAt = now;
+        }
+        updatedAt = now;
+    }
+
+    @PreUpdate
+    void onUpdate() {
+        updatedAt = Instant.now();
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getTenantId() {
+        return tenantId;
+    }
+
+    public void setTenantId(String tenantId) {
+        this.tenantId = tenantId;
+    }
+
+    public String getRunId() {
+        return runId;
+    }
+
+    public void setRunId(String runId) {
+        this.runId = runId;
+    }
+
+    public String getMessageId() {
+        return messageId;
+    }
+
+    public void setMessageId(String messageId) {
+        this.messageId = messageId;
+    }
+
+    public String getMediaType() {
+        return mediaType;
+    }
+
+    public void setMediaType(String mediaType) {
+        this.mediaType = mediaType;
+    }
+
+    public String getFilename() {
+        return filename;
+    }
+
+    public void setFilename(String filename) {
+        this.filename = filename;
+    }
+
+    public String getStorageKey() {
+        return storageKey;
+    }
+
+    public void setStorageKey(String storageKey) {
+        this.storageKey = storageKey;
+    }
+
+    public long getSizeBytes() {
+        return sizeBytes;
+    }
+
+    public void setSizeBytes(long sizeBytes) {
+        this.sizeBytes = sizeBytes;
+    }
+
+    public String getSha256() {
+        return sha256;
+    }
+
+    public void setSha256(String sha256) {
+        this.sha256 = sha256;
+    }
+
+    public String getCaption() {
+        return caption;
+    }
+
+    public void setCaption(String caption) {
+        this.caption = caption;
+    }
+
+    public Instant getCreatedAt() {
+        return createdAt;
+    }
+
+    public Instant getUpdatedAt() {
+        return updatedAt;
+    }
+}
diff --git a/backend-java/src/main/java/io/agentflow/api/repository/AttachmentRepository.java b/backend-java/src/main/java/io/agentflow/api/repository/AttachmentRepository.java
new file mode 100644
index 0000000..4e06886
--- /dev/null
+++ b/backend-java/src/main/java/io/agentflow/api/repository/AttachmentRepository.java
@@ -0,0 +1,21 @@
+package io.agentflow.api.repository;
+
+import io.agentflow.api.entity.AttachmentEntity;
+import java.util.List;
+import java.util.Optional;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface AttachmentRepository extends JpaRepository<AttachmentEntity, String> {
+
+    Optional<AttachmentEntity> findByIdAndTenantId(String id, String tenantId);
+
+    List<AttachmentEntity> findByRunIdAndTenantIdOrderByCreatedAtAsc(String runId, String tenantId);
+
+    List<AttachmentEntity> findByRunId(String runId);
+
+    List<AttachmentEntity> findByTenantId(String tenantId);
+
+    long deleteByRunId(String runId);
+
+    long deleteByTenantId(String tenantId);
+}
diff --git a/backend-java/src/main/java/io/agentflow/api/service/AttachmentNotFoundException.java b/backend-java/src/main/java/io/agentflow/api/service/AttachmentNotFoundException.java
new file mode 100644
index 0000000..225dac9
--- /dev/null
+++ b/backend-java/src/main/java/io/agentflow/api/service/AttachmentNotFoundException.java
@@ -0,0 +1,8 @@
+package io.agentflow.api.service;
+
+public class AttachmentNotFoundException extends RuntimeException {
+
+    public AttachmentNotFoundException(String attachmentId) {
+        super(attachmentId);
+    }
+}
diff --git a/backend-java/src/main/java/io/agentflow/api/service/AttachmentService.java b/backend-java/src/main/java/io/agentflow/api/service/AttachmentService.java
new file mode 100644
index 0000000..f35000b
--- /dev/null
+++ b/backend-java/src/main/java/io/agentflow/api/service/AttachmentService.java
@@ -0,0 +1,201 @@
+package io.agentflow.api.service;
+
+import io.agentflow.api.config.AgentflowProperties;
+import io.agentflow.api.dto.AttachmentResponse;
+import io.agentflow.api.entity.AttachmentEntity;
+import io.agentflow.api.repository.AttachmentRepository;
+import io.agentflow.api.security.AccessControl;
+import io.agentflow.api.security.Role;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.multipart.MultipartFile;
+
+@Service
+public class AttachmentService {
+
+    private final AttachmentRepository attachments;
+    private final AgentflowProperties props;
+
+    public AttachmentService(AttachmentRepository attachments, AgentflowProperties props) {
+        this.attachments = attachments;
+        this.props = props;
+    }
+
+    @Transactional
+    public AttachmentResponse upload(MultipartFile file, String caption) {
+        AccessControl.require(Role.OPERATOR);
+        String tenantId = AccessControl.tenantId(Role.OPERATOR);
+        byte[] data;
+        try {
+            data = file.getBytes();
+        } catch (IOException ex) {
+            throw new IllegalArgumentException("Failed to read upload", ex);
+        }
+        if (data.length == 0) {
+            throw new IllegalArgumentException("empty attachment");
+        }
+        long maxBytes = props.getAttachments().getMaxBytes();
+        if (data.length > maxBytes) {
+            throw new AttachmentTooLargeException(data.length, maxBytes);
+        }
+
+        AttachmentEntity entity = new AttachmentEntity();
+        entity.setTenantId(tenantId);
+        entity.setMediaType(
+                file.getContentType() == null || file.getContentType().isBlank()
+                        ? "application/octet-stream"
+                        : file.getContentType());
+        entity.setFilename(
+                file.getOriginalFilename() == null || file.getOriginalFilename().isBlank()
+                        ? "blob"
+                        : Path.of(file.getOriginalFilename()).getFileName().toString());
+        entity.setSizeBytes(data.length);
+        entity.setSha256(sha256Hex(data));
+        entity.setCaption(caption);
+        entity.setStorageKey("pending");
+        AttachmentEntity saved = attachments.save(entity);
+
+        String key = storageKey(tenantId, saved.getId(), saved.getFilename());
+        writeBlob(key, data);
+        saved.setStorageKey(key);
+        return AttachmentResponse.fromEntity(attachments.save(saved));
+    }
+
+    @Transactional(readOnly = true)
+    public AttachmentResponse getMeta(String id) {
+        return AttachmentResponse.fromEntity(require(id, Role.VIEWER));
+    }
+
+    @Transactional(readOnly = true)
+    public LoadedAttachment getContent(String id) {
+        AttachmentEntity entity = require(id, Role.VIEWER);
+        Path path = resolveKey(entity.getStorageKey());
+        if (!Files.isRegularFile(path)) {
+            throw new AttachmentNotFoundException(id);
+        }
+        try {
+            return new LoadedAttachment(entity, Files.readAllBytes(path));
+        } catch (IOException ex) {
+            throw new AttachmentNotFoundException(id);
+        }
+    }
+
+    @Transactional
+    public void bindInputAttachments(String runId, String tenantId, Map<String, Object> input) {
+        List<String> ids = extractAttachmentIds(input);
+        for (String attachmentId : ids) {
+            AttachmentEntity entity = attachments
+                    .findByIdAndTenantId(attachmentId, tenantId)
+                    .orElseThrow(() -> new AttachmentNotFoundException(attachmentId));
+            if (entity.getRunId() != null && !entity.getRunId().equals(runId)) {
+                throw new AttachmentNotFoundException(attachmentId);
+            }
+            entity.setRunId(runId);
+            attachments.save(entity);
+        }
+    }
+
+    @Transactional
+    public long eraseRun(String runId) {
+        List<AttachmentEntity> rows = attachments.findByRunId(runId);
+        for (AttachmentEntity row : rows) {
+            deleteBlobQuietly(row.getStorageKey());
+        }
+        return attachments.deleteByRunId(runId);
+    }
+
+    @Transactional
+    public long eraseTenant(String tenantId) {
+        List<AttachmentEntity> rows = attachments.findByTenantId(tenantId);
+        for (AttachmentEntity row : rows) {
+            deleteBlobQuietly(row.getStorageKey());
+        }
+        return attachments.deleteByTenantId(tenantId);
+    }
+
+    private AttachmentEntity require(String id, Role role) {
+        AccessControl.require(role);
+        String tenantId = AccessControl.tenantId(role);
+        return attachments
+                .findByIdAndTenantId(id, tenantId)
+                .orElseThrow(() -> new AttachmentNotFoundException(id));
+    }
+
+    @SuppressWarnings("unchecked")
+    private static List<String> extractAttachmentIds(Map<String, Object> input) {
+        if (input == null) {
+            return List.of();
+        }
+        Object raw = input.get("attachments");
+        if (!(raw instanceof List<?> list) || list.isEmpty()) {
+            return List.of();
+        }
+        List<String> ids = new ArrayList<>();
+        for (Object item : list) {
+            if (item instanceof String s) {
+                ids.add(s);
+            } else if (item instanceof Map<?, ?> map) {
+                Object id = map.get("id");
+                if (id instanceof String sid) {
+                    ids.add(sid);
+                }
+            }
+        }
+        return ids;
+    }
+
+    private void writeBlob(String key, byte[] data) {
+        Path path = resolveKey(key);
+        try {
+            Files.createDirectories(path.getParent());
+            Files.write(path, data);
+        } catch (IOException ex) {
+            throw new IllegalStateException("Failed to store attachment", ex);
+        }
+    }
+
+    private void deleteBlobQuietly(String key) {
+        try {
+            Files.deleteIfExists(resolveKey(key));
+        } catch (IOException ignored) {
+            // best-effort
+        }
+    }
+
+    private Path resolveKey(String key) {
+        Path root = Path.of(props.getAttachments().getStorageDir()).toAbsolutePath().normalize();
+        Path resolved = root.resolve(key).normalize();
+        if (!resolved.startsWith(root)) {
+            throw new IllegalArgumentException("invalid storage key");
+        }
+        return resolved;
+    }
+
+    private static String storageKey(String tenantId, String attachmentId, String filename) {
+        String safe = filename.replaceAll("[^A-Za-z0-9._+-]", "_");
+        if (safe.isBlank()) {
+            safe = "blob";
+        }
+        return tenantId + "/" + attachmentId + "/" + safe;
+    }
+
+    private static String sha256Hex(byte[] data) {
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            return HexFormat.of().formatHex(digest.digest(data));
+        } catch (NoSuchAlgorithmException ex) {
+            throw new IllegalStateException(ex);
+        }
+    }
+
+    public record LoadedAttachment(AttachmentEntity entity, byte[] data) {}
+}
diff --git a/backend-java/src/main/java/io/agentflow/api/service/AttachmentTooLargeException.java b/backend-java/src/main/java/io/agentflow/api/service/AttachmentTooLargeException.java
new file mode 100644
index 0000000..57b2e4d
--- /dev/null
+++ b/backend-java/src/main/java/io/agentflow/api/service/AttachmentTooLargeException.java
@@ -0,0 +1,21 @@
+package io.agentflow.api.service;
+
+public class AttachmentTooLargeException extends RuntimeException {
+
+    private final long size;
+    private final long limit;
+
+    public AttachmentTooLargeException(long size, long limit) {
+        super("attachment " + size + " bytes exceeds limit " + limit);
+        this.size = size;
+        this.limit = limit;
+    }
+
+    public long getSize() {
+        return size;
+    }
+
+    public long getLimit() {
+        return limit;
+    }
+}
diff --git a/backend-java/src/main/java/io/agentflow/api/service/RetentionService.java b/backend-java/src/main/java/io/agentflow/api/service/RetentionService.java
index ddacd9f..c769da4 100644
--- a/backend-java/src/main/java/io/agentflow/api/service/RetentionService.java
+++ b/backend-java/src/main/java/io/agentflow/api/service/RetentionService.java
@@ -29,6 +29,7 @@ public class RetentionService {
     private final RunRepository runs;
     private final MessageRepository messages;
     private final CheckpointRepository checkpoints;
+    private final AttachmentService attachments;
     private final StringRedisTemplate redis;
     private final AgentflowProperties props;
 
@@ -36,11 +37,13 @@ public RetentionService(
             RunRepository runs,
             MessageRepository messages,
             CheckpointRepository checkpoints,
+            AttachmentService attachments,
             StringRedisTemplate redis,
             AgentflowProperties props) {
         this.runs = runs;
         this.messages = messages;
         this.checkpoints = checkpoints;
+        this.attachments = attachments;
         this.redis = redis;
         this.props = props;
     }
@@ -59,6 +62,7 @@ public EraseRunDataResponse eraseRunData(String runId) {
 
         long msgDeleted = messages.deleteByRunId(runId);
         long cpDeleted = checkpoints.deleteByRunId(runId);
+        attachments.eraseRun(runId);
         clearTranscriptFields(run);
         runs.save(run);
         deleteEventLog(runId);
@@ -77,9 +81,11 @@ public EraseTenantDataResponse eraseTenantData() {
         for (RunEntity run : tenantRuns) {
             totalMessages += messages.deleteByRunId(run.getId());
             totalCheckpoints += checkpoints.deleteByRunId(run.getId());
+            attachments.eraseRun(run.getId());
             clearTranscriptFields(run);
             deleteEventLog(run.getId());
         }
+        attachments.eraseTenant(tenantId);
         runs.saveAll(tenantRuns);
 
         return new EraseTenantDataResponse(
@@ -114,6 +120,7 @@ public RetentionPurgeResponse purgeExpired(String tenantId, boolean dryRun) {
             for (RunEntity run : candidates) {
                 totalMessages += messages.deleteByRunId(run.getId());
                 totalCheckpoints += checkpoints.deleteByRunId(run.getId());
+                attachments.eraseRun(run.getId());
                 clearTranscriptFields(run);
                 deleteEventLog(run.getId());
             }
diff --git a/backend-java/src/main/java/io/agentflow/api/service/RunService.java b/backend-java/src/main/java/io/agentflow/api/service/RunService.java
index 58f628b..1b09974 100644
--- a/backend-java/src/main/java/io/agentflow/api/service/RunService.java
+++ b/backend-java/src/main/java/io/agentflow/api/service/RunService.java
@@ -60,6 +60,7 @@ public class RunService {
     private final ThreadRepository threads;
     private final JobProducer jobProducer;
     private final CancelSignal cancelSignal;
+    private final AttachmentService attachmentService;
 
     public RunService(
             RunRepository runs,
@@ -70,7 +71,8 @@ public RunService(
             AgentService agentService,
             ThreadRepository threads,
             JobProducer jobProducer,
-            CancelSignal cancelSignal) {
+            CancelSignal cancelSignal,
+            AttachmentService attachmentService) {
         this.runs = runs;
         this.steps = steps;
         this.messages = messages;
@@ -80,6 +82,7 @@ public RunService(
         this.threads = threads;
         this.jobProducer = jobProducer;
         this.cancelSignal = cancelSignal;
+        this.attachmentService = attachmentService;
     }
 
     @Transactional
@@ -115,6 +118,8 @@ public RunResponse create(RunCreateRequest req) {
                 Map.of(AGENT_VERSION_METADATA_KEY, agent.getVersion()));
         run.setMetadata(metadata);
         RunEntity saved = runs.save(run);
+        attachmentService.bindInputAttachments(
+                saved.getId(), agent.getTenantId(), saved.getInput());
         enqueueJobAfterCommit(saved.getId(), agent.getId(), adapter);
 
         return toResponse(saved);
diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml
index 8816f9d..143b25e 100644
--- a/backend-java/src/main/resources/application.yml
+++ b/backend-java/src/main/resources/application.yml
@@ -18,6 +18,10 @@ spring:
   mvc:
     async:
       request-timeout: -1
+  servlet:
+    multipart:
+      max-file-size: ${AGENTFLOW_ATTACHMENT_MAX_BYTES:10485760}
+      max-request-size: ${AGENTFLOW_ATTACHMENT_MAX_BYTES:10485760}
 
 server:
   port: ${AGENTFLOW_SERVER_PORT:8000}
@@ -68,10 +72,13 @@ agentflow:
       tenant-claim: ${AGENTFLOW_OIDC_TENANT_CLAIM:tenant_id}
       role-claim: ${AGENTFLOW_OIDC_ROLE_CLAIM:roles}
       default-role: ${AGENTFLOW_OIDC_DEFAULT_ROLE:viewer}
-  retention:
-    tenant-ttl-days: ${AGENTFLOW_DATA_RETENTION_TENANT_TTL_DAYS:0}
-    purge-interval-seconds: ${AGENTFLOW_DATA_RETENTION_PURGE_INTERVAL_SECONDS:3600}
-    purge-batch-size: ${AGENTFLOW_DATA_RETENTION_PURGE_BATCH_SIZE:100}
+    retention:
+      tenant-ttl-days: ${AGENTFLOW_DATA_RETENTION_TENANT_TTL_DAYS:0}
+      purge-interval-seconds: ${AGENTFLOW_DATA_RETENTION_PURGE_INTERVAL_SECONDS:3600}
+      purge-batch-size: ${AGENTFLOW_DATA_RETENTION_PURGE_BATCH_SIZE:100}
+  attachments:
+    storage-dir: ${AGENTFLOW_ATTACHMENT_STORAGE_DIR:./data/attachments}
+    max-bytes: ${AGENTFLOW_ATTACHMENT_MAX_BYTES:10485760}
 
 management:
   tracing:
diff --git a/backend-java/src/test/java/io/agentflow/api/service/RunServiceTest.java b/backend-java/src/test/java/io/agentflow/api/service/RunServiceTest.java
index 750ec8e..2c0dc16 100644
--- a/backend-java/src/test/java/io/agentflow/api/service/RunServiceTest.java
+++ b/backend-java/src/test/java/io/agentflow/api/service/RunServiceTest.java
@@ -81,6 +81,7 @@ private static RunService service(RunRepository runs, AgentService agents) {
                 agents,
                 mock(ThreadRepository.class),
                 mock(JobProducer.class),
-                mock(CancelSignal.class));
+                mock(CancelSignal.class),
+                mock(AttachmentService.class));
     }
 }
diff --git a/backend/alembic/env.py b/backend/alembic/env.py
index c6f05cb..2a3e03b 100644
--- a/backend/alembic/env.py
+++ b/backend/alembic/env.py
@@ -18,6 +18,7 @@
 from app.db.base import Base
 from app.models import (  # noqa: F401 -- ensure models are imported
     Agent,
+    Attachment,
     Checkpoint,
     Message,
     Project,
diff --git a/backend/alembic/versions/0007_attachments.py b/backend/alembic/versions/0007_attachments.py
new file mode 100644
index 0000000..abb04ad
--- /dev/null
+++ b/backend/alembic/versions/0007_attachments.py
@@ -0,0 +1,58 @@
+"""Add attachments table for multimodal file persistence.
+
+Revision ID: 0007_attachments
+Revises: 0006_threads
+Create Date: 2026-09-04
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "0007_attachments"
+down_revision = "0006_threads"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "attachments",
+        sa.Column("id", sa.String(26), primary_key=True),
+        sa.Column("tenant_id", sa.String(64), nullable=False),
+        sa.Column("run_id", sa.String(26), nullable=True),
+        sa.Column("message_id", sa.String(26), nullable=True),
+        sa.Column("media_type", sa.String(128), nullable=False),
+        sa.Column("filename", sa.String(512), nullable=False),
+        sa.Column("storage_key", sa.String(512), nullable=False),
+        sa.Column("size_bytes", sa.BigInteger(), nullable=False),
+        sa.Column("sha256", sa.String(64), nullable=False),
+        sa.Column("caption", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["run_id"], ["runs.id"], name="fk_attachments_run_id", ondelete="SET NULL"
+        ),
+        sa.ForeignKeyConstraint(
+            ["message_id"],
+            ["messages.id"],
+            name="fk_attachments_message_id",
+            ondelete="SET NULL",
+        ),
+    )
+    op.create_index("ix_attachments_tenant_id", "attachments", ["tenant_id"])
+    op.create_index("ix_attachments_run_id", "attachments", ["run_id"])
+    op.create_index("ix_attachments_message_id", "attachments", ["message_id"])
+    op.create_index(
+        "ix_attachments_storage_key", "attachments", ["storage_key"], unique=True
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_attachments_storage_key", table_name="attachments")
+    op.drop_index("ix_attachments_message_id", table_name="attachments")
+    op.drop_index("ix_attachments_run_id", table_name="attachments")
+    op.drop_index("ix_attachments_tenant_id", table_name="attachments")
+    op.drop_table("attachments")
diff --git a/backend/app/adapters/base.py b/backend/app/adapters/base.py
index 9a09982..292ee86 100644
--- a/backend/app/adapters/base.py
+++ b/backend/app/adapters/base.py
@@ -48,6 +48,8 @@ class AdapterContext:
     # Prior turns from the same thread (window-trimmed); empty when no thread.
     thread_id: str | None = None
     thread_messages: list[dict[str, Any]] = field(default_factory=list)
+    # Multimodal attachments bound to this run (metadata + storage_key); empty when none.
+    attachments: list[Any] = field(default_factory=list)
     step_index_base: int = 0
     emit: EmitCallback = field(
         default=None  # type: ignore[assignment]
@@ -105,9 +107,10 @@ async def emit_token_delta(
         """Stream an incremental token chunk to SSE subscribers (no DB write).
 
         ``part`` discriminates fine-grained streams: ``text`` (default) for the
-        visible reply, ``reasoning`` for model thinking / chain-of-thought.
-        Reasoning deltas are still ephemeral; persist the finished block with
-        ``emit_message(..., kind="reasoning")``.
+        visible reply, ``reasoning`` for model thinking / chain-of-thought, and
+        ``attachment`` for multimodal attachment announcements (payload may
+        include an ``attachment`` object). Reasoning / attachment deltas stay
+        ephemeral; persist finished blocks with ``emit_message``.
         """
         await self.emit(
             "token.delta",
diff --git a/backend/app/adapters/langgraph_adapter.py b/backend/app/adapters/langgraph_adapter.py
index 03084ee..3023d14 100644
--- a/backend/app/adapters/langgraph_adapter.py
+++ b/backend/app/adapters/langgraph_adapter.py
@@ -303,6 +303,8 @@ class _RunState:
     step_index: int = 0
     node_indices: dict[str, int] = field(default_factory=dict)
     emitted_system_prompts: set[str] = field(default_factory=set)
+    input_attachments: list[Any] = field(default_factory=list)
+    attachments_emitted: bool = False
 
     def next_step_index(self, node_id: str) -> int:
         if node_id not in self.node_indices:
@@ -359,6 +361,17 @@ async def run(self, ctx: AdapterContext) -> AdapterResult:
                 ctx.run_messages
             ),
         )
+        run_state.input_attachments = list(ctx.attachments or [])
+        # Skip re-emitting on retry/resume when transcript already has attachment rows.
+        if _messages_have_attachments(ctx.run_messages):
+            run_state.attachments_emitted = True
+        elif run_state.input_attachments:
+            from app.runtime.attachments import emit_input_attachments
+
+            await emit_input_attachments(
+                ctx, run_state.input_attachments, step_index=None, role="user"
+            )
+            run_state.attachments_emitted = True
 
         graph = StateGraph(dict)
         for node_spec in graph_spec.nodes:
@@ -589,6 +602,11 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]:
             ctx = run_state.ctx
             stream_tokens = run_state.config.get("stream_tokens", True)
             messages = _seed_agent_messages(state, system_prompt)
+            if run_state.input_attachments:
+                user_content = await _multimodal_user_content(
+                    str(state.get("input", "")), run_state.input_attachments
+                )
+                messages = _apply_multimodal_to_seed(messages, user_content)
             tool_results = dict(state.get("tool_results") or {})
             started = time.monotonic()
             total_in = 0
@@ -714,7 +732,13 @@ async def _run_one_tool(tc: dict[str, Any]) -> ToolOutcome:
                     messages.append({"role": "assistant", "content": final_reply})
 
                 latency_ms = int((time.monotonic() - started) * 1000)
-                cost_usd = estimate_cost_usd(model, total_in, total_out)
+                used_model = (
+                    (last_response.model if last_response else None) or model
+                )
+                routing_meta = (
+                    last_response.routing if last_response is not None else None
+                )
+                cost_usd = estimate_cost_usd(used_model, total_in, total_out)
                 await _emit_assistant_messages(
                     ctx,
                     step_index=step_idx,
@@ -799,10 +823,13 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]:
                 for m in (state.get("messages") or [])
                 if isinstance(m, dict) and m.get("role") != "system"
             ]
+            user_content = await _multimodal_user_content(
+                user_input, run_state.input_attachments
+            )
             messages = [
                 {"role": "system", "content": system_prompt},
                 *history,
-                {"role": "user", "content": user_input},
+                {"role": "user", "content": user_content},
             ]
             response = await self._invoke_model(
                 ctx,
@@ -1401,3 +1428,40 @@ async def _emit_assistant_messages(
             kind="reasoning",
         )
     await ctx.emit_message(role="assistant", content=content, step_index=step_index)
+
+
+def _messages_have_attachments(messages: list[dict[str, Any]] | None) -> bool:
+    if not messages:
+        return False
+    for message in messages:
+        extra = message.get("extra") or {}
+        if extra.get("kind") == "attachment" or extra.get("attachments"):
+            return True
+        if message.get("attachments"):
+            return True
+    return False
+
+
+async def _multimodal_user_content(
+    text: str, attachments: list[Any]
+) -> str | list[dict[str, Any]]:
+    if not attachments:
+        return text
+    from app.runtime.attachments import openai_user_content
+
+    return await openai_user_content(text, attachments)
+
+
+def _apply_multimodal_to_seed(
+    messages: list[dict[str, Any]],
+    user_content: str | list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Replace the trailing user turn with multimodal content when needed."""
+    if not messages or isinstance(user_content, str):
+        return messages
+    out = list(messages)
+    for i in range(len(out) - 1, -1, -1):
+        if isinstance(out[i], dict) and out[i].get("role") == "user":
+            out[i] = {**out[i], "content": user_content}
+            break
+    return out
diff --git a/backend/app/api/v1/attachments.py b/backend/app/api/v1/attachments.py
new file mode 100644
index 0000000..1228d0b
--- /dev/null
+++ b/backend/app/api/v1/attachments.py
@@ -0,0 +1,89 @@
+from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
+from fastapi.responses import Response
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.auth import AuthPrincipal, Role, require_role
+from app.db.session import get_session
+from app.schemas.attachment import AttachmentRead
+from app.services.attachment_service import (
+    AttachmentNotFound,
+    AttachmentService,
+    AttachmentTooLarge,
+)
+
+router = APIRouter(prefix="/attachments", tags=["attachments"])
+
+
+def get_attachment_service(
+    session: AsyncSession = Depends(get_session),
+) -> AttachmentService:
+    return AttachmentService(session=session)
+
+
+@router.post("", response_model=AttachmentRead, status_code=status.HTTP_201_CREATED)
+async def upload_attachment(
+    file: UploadFile = File(...),
+    caption: str | None = Form(default=None),
+    service: AttachmentService = Depends(get_attachment_service),
+    principal: AuthPrincipal = Depends(require_role(Role.OPERATOR)),
+) -> AttachmentRead:
+    data = await file.read()
+    try:
+        attachment = await service.create(
+            tenant_id=principal.tenant_id,
+            filename=file.filename or "blob",
+            media_type=file.content_type or "application/octet-stream",
+            data=data,
+            caption=caption,
+        )
+    except AttachmentTooLarge as exc:
+        raise HTTPException(
+            status_code=413,
+            detail=f"Attachment too large: {exc.size} > {exc.limit} bytes",
+        ) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    return service.to_read(attachment)
+
+
+@router.get("/{attachment_id}", response_model=AttachmentRead)
+async def get_attachment_meta(
+    attachment_id: str,
+    service: AttachmentService = Depends(get_attachment_service),
+    principal: AuthPrincipal = Depends(require_role(Role.VIEWER)),
+) -> AttachmentRead:
+    try:
+        attachment = await service.get(attachment_id, tenant_id=principal.tenant_id)
+    except AttachmentNotFound as exc:
+        raise HTTPException(
+            status_code=404, detail=f"Attachment not found: {exc}"
+        ) from exc
+    return service.to_read(attachment)
+
+
+@router.get("/{attachment_id}/content")
+async def download_attachment(
+    attachment_id: str,
+    service: AttachmentService = Depends(get_attachment_service),
+    principal: AuthPrincipal = Depends(require_role(Role.VIEWER)),
+) -> Response:
+    try:
+        attachment, data = await service.get_bytes(
+            attachment_id, tenant_id=principal.tenant_id
+        )
+    except AttachmentNotFound as exc:
+        raise HTTPException(
+            status_code=404, detail=f"Attachment not found: {exc}"
+        ) from exc
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=404, detail=f"Attachment blob missing: {attachment_id}"
+        ) from exc
+    headers = {
+        "Content-Disposition": f'inline; filename="{attachment.filename}"',
+    }
+    return Response(
+        content=data,
+        media_type=attachment.media_type,
+        headers=headers,
+    )
diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py
index 31bfa13..bbf3d8b 100644
--- a/backend/app/api/v1/router.py
+++ b/backend/app/api/v1/router.py
@@ -1,12 +1,22 @@
 from fastapi import APIRouter
 
-from app.api.v1 import agents, events, health, projects, retention, runs, threads
+from app.api.v1 import (
+    agents,
+    attachments,
+    events,
+    health,
+    projects,
+    retention,
+    runs,
+    threads,
+)
 
 api_router = APIRouter(prefix="/v1")
 api_router.include_router(health.router)
 api_router.include_router(projects.router)
 api_router.include_router(agents.router)
 api_router.include_router(threads.router)
+api_router.include_router(attachments.router)
 api_router.include_router(runs.router)
 api_router.include_router(retention.router)
 api_router.include_router(events.router)
diff --git a/backend/app/api/v1/runs.py b/backend/app/api/v1/runs.py
index 88067a6..af38f5d 100644
--- a/backend/app/api/v1/runs.py
+++ b/backend/app/api/v1/runs.py
@@ -14,6 +14,7 @@
 )
 from app.services.run_service import (
     AgentNotFound,
+    AttachmentRefNotFound,
     RunConflict,
     RunNotFound,
     RunService,
@@ -68,6 +69,10 @@ async def create_run(
         raise HTTPException(status_code=404, detail=f"Agent not found: {exc}") from exc
     except ThreadNotFound as exc:
         raise HTTPException(status_code=404, detail=f"Thread not found: {exc}") from exc
+    except AttachmentRefNotFound as exc:
+        raise HTTPException(
+            status_code=404, detail=f"Attachment not found: {exc}"
+        ) from exc
 
     await service.start_run(run.id)
     try:
diff --git a/backend/app/core/config.py b/backend/app/core/config.py
index 1520ffa..1be8347 100644
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@@ -324,6 +324,20 @@ class Settings(BaseSettings):
         description="Maximum runs purged per sweeper cycle per tenant.",
     )
 
+    attachment_storage_dir: str = Field(
+        default="./data/attachments",
+        description=(
+            "Local object-store root for multimodal attachments "
+            "(shared later with memory document chunks)."
+        ),
+    )
+    attachment_max_bytes: int = Field(
+        default=10 * 1024 * 1024,
+        ge=1024,
+        le=100 * 1024 * 1024,
+        description="Maximum upload size per attachment (default 10 MiB).",
+    )
+
 
 def effective_jobs_impl(settings: Settings | None = None) -> JobsImpl:
     """Resolve the job protocol Java and Python must agree on."""
diff --git a/backend/app/main.py b/backend/app/main.py
index 9b791e7..bf7ae35 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -20,7 +20,6 @@
 setup_logging()
 logger = get_logger("app")
 
-
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     load_adapter_plugins()
diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 8b44ae3..e19d767 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -1,4 +1,5 @@
 from app.models.agent import Agent, AgentVersion
+from app.models.attachment import Attachment
 from app.models.project import Project
 from app.models.run import Checkpoint, Message, Run, RunStatus, Step, ToolCall
 from app.models.thread import Thread
@@ -6,6 +7,7 @@
 __all__ = [
     "Agent",
     "AgentVersion",
+    "Attachment",
     "Checkpoint",
     "Message",
     "Project",
diff --git a/backend/app/models/attachment.py b/backend/app/models/attachment.py
new file mode 100644
index 0000000..363d539
--- /dev/null
+++ b/backend/app/models/attachment.py
@@ -0,0 +1,52 @@
+"""Multimodal attachment metadata (blobs live in object storage)."""
+
+from __future__ import annotations
+
+from sqlalchemy import BigInteger, ForeignKey, Index, String, Text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from ulid import ULID
+
+from app.db.base import Base
+from app.models.agent import DEFAULT_TENANT_ID
+
+
+def _ulid() -> str:
+    return str(ULID())
+
+
+class Attachment(Base):
+    """A file attached to a run / message.
+
+    ``Message.content`` stays plain text; attachment refs live in
+    ``Message.extra.attachments`` and this table. Binary bytes are stored
+    under ``storage_key`` in the configured object store (local FS by default;
+    the same store is reserved for future memory-layer document chunks).
+    """
+
+    __tablename__ = "attachments"
+    __table_args__ = (
+        Index("ix_attachments_tenant_id", "tenant_id"),
+        Index("ix_attachments_run_id", "run_id"),
+        Index("ix_attachments_message_id", "message_id"),
+        Index("ix_attachments_storage_key", "storage_key", unique=True),
+    )
+
+    id: Mapped[str] = mapped_column(String(26), primary_key=True, default=_ulid)
+    tenant_id: Mapped[str] = mapped_column(
+        String(64), default=DEFAULT_TENANT_ID, nullable=False
+    )
+    run_id: Mapped[str | None] = mapped_column(
+        ForeignKey("runs.id", ondelete="SET NULL"), nullable=True
+    )
+    message_id: Mapped[str | None] = mapped_column(
+        ForeignKey("messages.id", ondelete="SET NULL"), nullable=True
+    )
+    media_type: Mapped[str] = mapped_column(String(128), nullable=False)
+    filename: Mapped[str] = mapped_column(String(512), nullable=False)
+    storage_key: Mapped[str] = mapped_column(String(512), nullable=False)
+    size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    sha256: Mapped[str] = mapped_column(String(64), nullable=False)
+    caption: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+    run: Mapped["Run | None"] = relationship()  # noqa: F821
+    message: Mapped["Message | None"] = relationship()  # noqa: F821
diff --git a/backend/app/runtime/attachments.py b/backend/app/runtime/attachments.py
new file mode 100644
index 0000000..2c91c56
--- /dev/null
+++ b/backend/app/runtime/attachments.py
@@ -0,0 +1,112 @@
+"""Helpers for multimodal attachments in adapter prompts and SSE."""
+
+from __future__ import annotations
+
+import base64
+from typing import Any
+
+from app.adapters.base import AdapterContext
+from app.models.attachment import Attachment
+from app.runtime.object_store import ObjectStore, get_object_store
+from app.schemas.attachment import attachment_ref
+
+
+def is_image_media_type(media_type: str) -> bool:
+    return media_type.lower().startswith("image/")
+
+
+async def load_attachment_bytes(
+    attachment: Attachment,
+    *,
+    store: ObjectStore | None = None,
+) -> bytes:
+    return await (store or get_object_store()).get(attachment.storage_key)
+
+
+def data_url(media_type: str, data: bytes) -> str:
+    encoded = base64.b64encode(data).decode("ascii")
+    return f"data:{media_type};base64,{encoded}"
+
+
+async def openai_user_content(
+    text: str,
+    attachments: list[Attachment],
+    *,
+    store: ObjectStore | None = None,
+) -> str | list[dict[str, Any]]:
+    """Build OpenAI chat ``content`` — multimodal list when images are present."""
+    image_parts: list[dict[str, Any]] = []
+    notes: list[str] = []
+    for attachment in attachments:
+        if is_image_media_type(attachment.media_type):
+            blob = await load_attachment_bytes(attachment, store=store)
+            image_parts.append(
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": data_url(attachment.media_type, blob),
+                    },
+                }
+            )
+        else:
+            caption = attachment.caption or attachment.filename
+            notes.append(
+                f"[attachment {attachment.id} {attachment.media_type} {caption}]"
+            )
+
+    body = text or ""
+    if notes:
+        body = (body + "\n\n" if body else "") + "\n".join(notes)
+
+    if not image_parts:
+        return body
+
+    parts: list[dict[str, Any]] = [{"type": "text", "text": body or "(image)"}]
+    parts.extend(image_parts)
+    return parts
+
+
+async def emit_input_attachments(
+    ctx: AdapterContext,
+    attachments: list[Attachment],
+    *,
+    step_index: int | None = None,
+    role: str = "user",
+) -> list[dict[str, Any]]:
+    """Stream fine-grained attachment deltas, then persist a message with refs.
+
+    Returns the compact refs stored on ``Message.extra.attachments``.
+    """
+    if not attachments:
+        return []
+
+    refs = [attachment_ref(a) for a in attachments]
+    for ref in refs:
+        await ctx.emit_token_delta(
+            step_index=step_index if step_index is not None else 0,
+            delta=str(ref.get("filename") or ref.get("id") or ""),
+            role=role,
+            part="attachment",
+            attachment=ref,
+        )
+
+    captions = [str(a.caption or a.filename) for a in attachments]
+    content = "; ".join(captions) if captions else f"{len(attachments)} attachment(s)"
+    await ctx.emit_message(
+        role=role,
+        content=content,
+        step_index=step_index,
+        kind="attachment",
+        attachments=refs,
+    )
+    return refs
+
+
+def message_dict_with_attachments(msg: dict[str, Any]) -> dict[str, Any]:
+    """Keep attachment refs on chat dicts reconstructed from Message rows."""
+    payload = dict(msg)
+    extra = payload.get("extra") or {}
+    attachments = extra.get("attachments")
+    if attachments:
+        payload["attachments"] = attachments
+    return payload
diff --git a/backend/app/runtime/messages.py b/backend/app/runtime/messages.py
index 2d21e1d..92c9fdc 100644
--- a/backend/app/runtime/messages.py
+++ b/backend/app/runtime/messages.py
@@ -18,6 +18,13 @@ def message_row_to_dict(msg: Message) -> dict[str, Any]:
     tool_calls = extra.get("tool_calls")
     if tool_calls:
         payload["tool_calls"] = tool_calls
+    attachments = extra.get("attachments")
+    if attachments:
+        # Keep refs for adapters that rebuild multimodal content; binary stays
+        # in object storage and is loaded on demand.
+        payload["extra"] = {"attachments": attachments}
+        if extra.get("kind"):
+            payload["extra"]["kind"] = extra["kind"]
     return payload
 
 
diff --git a/backend/app/runtime/object_store.py b/backend/app/runtime/object_store.py
new file mode 100644
index 0000000..0d5026a
--- /dev/null
+++ b/backend/app/runtime/object_store.py
@@ -0,0 +1,89 @@
+"""Object storage for multimodal attachments (and future memory blobs).
+
+Default backend is a local filesystem under ``AGENTFLOW_ATTACHMENT_STORAGE_DIR``.
+The interface is intentionally small so an S3-compatible backend can replace
+it without changing Attachment rows (they only store ``storage_key``).
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Protocol
+
+from app.core.config import get_settings
+
+
+class ObjectStore(Protocol):
+    async def put(self, key: str, data: bytes, *, media_type: str) -> None: ...
+
+    async def get(self, key: str) -> bytes: ...
+
+    async def delete(self, key: str) -> None: ...
+
+    async def exists(self, key: str) -> bool: ...
+
+
+class LocalObjectStore:
+    """Filesystem-backed object store."""
+
+    def __init__(self, root: Path | str | None = None) -> None:
+        settings = get_settings()
+        self.root = Path(root or settings.attachment_storage_dir).resolve()
+        self.root.mkdir(parents=True, exist_ok=True)
+
+    def _path(self, key: str) -> Path:
+        # Prevent path traversal: only allow relative keys under root.
+        safe = Path(key)
+        if safe.is_absolute() or ".." in safe.parts:
+            raise ValueError(f"invalid storage key: {key!r}")
+        path = (self.root / safe).resolve()
+        if not str(path).startswith(str(self.root)):
+            raise ValueError(f"storage key escapes root: {key!r}")
+        return path
+
+    async def put(self, key: str, data: bytes, *, media_type: str) -> None:
+        path = self._path(key)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+
+    async def get(self, key: str) -> bytes:
+        path = self._path(key)
+        if not path.is_file():
+            raise FileNotFoundError(key)
+        return path.read_bytes()
+
+    async def delete(self, key: str) -> None:
+        path = self._path(key)
+        if path.is_file():
+            path.unlink()
+
+    async def exists(self, key: str) -> bool:
+        return self._path(key).is_file()
+
+
+_store: ObjectStore | None = None
+
+
+def get_object_store() -> ObjectStore:
+    global _store
+    if _store is None:
+        _store = LocalObjectStore()
+    return _store
+
+
+def set_object_store(store: ObjectStore | None) -> None:
+    """Override the process-wide store (tests)."""
+    global _store
+    _store = store
+
+
+def sha256_hex(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def attachment_storage_key(*, tenant_id: str, attachment_id: str, filename: str) -> str:
+    """Stable relative key: ``{tenant}/{id}/{safe_filename}``."""
+    safe_name = Path(filename).name or "blob"
+    safe_name = "".join(ch if ch.isalnum() or ch in "._-+" else "_" for ch in safe_name)
+    return f"{tenant_id}/{attachment_id}/{safe_name}"
diff --git a/backend/app/schemas/attachment.py b/backend/app/schemas/attachment.py
new file mode 100644
index 0000000..fcb87cf
--- /dev/null
+++ b/backend/app/schemas/attachment.py
@@ -0,0 +1,41 @@
+"""Attachment API schemas."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class AttachmentRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    tenant_id: str
+    run_id: str | None = None
+    message_id: str | None = None
+    media_type: str
+    filename: str
+    size_bytes: int
+    sha256: str
+    caption: str | None = None
+    url: str = Field(description="Relative download path under the API root.")
+    created_at: datetime
+    updated_at: datetime
+
+
+def attachment_url(attachment_id: str) -> str:
+    return f"/v1/attachments/{attachment_id}"
+
+
+def attachment_ref(attachment: object) -> dict[str, object]:
+    """Compact ref stored on ``Message.extra.attachments`` / SSE payloads."""
+    return {
+        "id": getattr(attachment, "id"),
+        "media_type": getattr(attachment, "media_type"),
+        "filename": getattr(attachment, "filename"),
+        "size_bytes": int(getattr(attachment, "size_bytes") or 0),
+        "url": attachment_url(str(getattr(attachment, "id"))),
+        "caption": getattr(attachment, "caption", None),
+        "sha256": getattr(attachment, "sha256", None),
+    }
diff --git a/backend/app/services/attachment_service.py b/backend/app/services/attachment_service.py
new file mode 100644
index 0000000..cfcfc6a
--- /dev/null
+++ b/backend/app/services/attachment_service.py
@@ -0,0 +1,248 @@
+"""Upload, bind, and erase multimodal attachments."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import get_settings
+from app.models.attachment import Attachment
+from app.runtime.object_store import (
+    ObjectStore,
+    attachment_storage_key,
+    get_object_store,
+    sha256_hex,
+)
+from app.schemas.attachment import AttachmentRead, attachment_ref, attachment_url
+
+
+class AttachmentNotFound(Exception):
+    def __init__(self, attachment_id: str) -> None:
+        self.attachment_id = attachment_id
+        super().__init__(attachment_id)
+
+
+class AttachmentTooLarge(Exception):
+    def __init__(self, size: int, limit: int) -> None:
+        self.size = size
+        self.limit = limit
+        super().__init__(f"attachment {size} bytes exceeds limit {limit}")
+
+
+class AttachmentService:
+    def __init__(
+        self,
+        session: AsyncSession,
+        *,
+        store: ObjectStore | None = None,
+    ) -> None:
+        self.session = session
+        self.store = store or get_object_store()
+
+    async def create(
+        self,
+        *,
+        tenant_id: str,
+        filename: str,
+        media_type: str,
+        data: bytes,
+        caption: str | None = None,
+    ) -> Attachment:
+        settings = get_settings()
+        if len(data) > settings.attachment_max_bytes:
+            raise AttachmentTooLarge(len(data), settings.attachment_max_bytes)
+        if not data:
+            raise ValueError("empty attachment")
+
+        attachment = Attachment(
+            tenant_id=tenant_id,
+            media_type=media_type or "application/octet-stream",
+            filename=filename or "blob",
+            storage_key="",  # filled after id is known
+            size_bytes=len(data),
+            sha256=sha256_hex(data),
+            caption=caption,
+        )
+        self.session.add(attachment)
+        await self.session.flush()
+
+        key = attachment_storage_key(
+            tenant_id=tenant_id,
+            attachment_id=attachment.id,
+            filename=attachment.filename,
+        )
+        await self.store.put(key, data, media_type=attachment.media_type)
+        attachment.storage_key = key
+        await self.session.commit()
+        await self.session.refresh(attachment)
+        return attachment
+
+    async def get(
+        self,
+        attachment_id: str,
+        *,
+        tenant_id: str | None = None,
+    ) -> Attachment:
+        attachment = await self.session.get(Attachment, attachment_id)
+        if attachment is None:
+            raise AttachmentNotFound(attachment_id)
+        if tenant_id is not None and attachment.tenant_id != tenant_id:
+            raise AttachmentNotFound(attachment_id)
+        return attachment
+
+    async def get_bytes(
+        self,
+        attachment_id: str,
+        *,
+        tenant_id: str | None = None,
+    ) -> tuple[Attachment, bytes]:
+        attachment = await self.get(attachment_id, tenant_id=tenant_id)
+        data = await self.store.get(attachment.storage_key)
+        return attachment, data
+
+    async def list_for_run(
+        self,
+        run_id: str,
+        *,
+        tenant_id: str | None = None,
+    ) -> list[Attachment]:
+        stmt = select(Attachment).where(Attachment.run_id == run_id)
+        if tenant_id is not None:
+            stmt = stmt.where(Attachment.tenant_id == tenant_id)
+        stmt = stmt.order_by(Attachment.created_at.asc())
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
+    async def bind_to_run(
+        self,
+        *,
+        run_id: str,
+        tenant_id: str,
+        attachment_ids: list[str],
+    ) -> list[Attachment]:
+        """Associate previously uploaded attachments with a run."""
+        if not attachment_ids:
+            return []
+        bound: list[Attachment] = []
+        for attachment_id in attachment_ids:
+            attachment = await self.get(attachment_id, tenant_id=tenant_id)
+            if attachment.run_id and attachment.run_id != run_id:
+                raise AttachmentNotFound(attachment_id)
+            attachment.run_id = run_id
+            bound.append(attachment)
+        await self.session.commit()
+        return bound
+
+    async def bind_message(
+        self,
+        *,
+        attachment_ids: list[str],
+        message_id: str,
+        tenant_id: str,
+    ) -> None:
+        if not attachment_ids:
+            return
+        for attachment_id in attachment_ids:
+            attachment = await self.get(attachment_id, tenant_id=tenant_id)
+            attachment.message_id = message_id
+        await self.session.commit()
+
+    async def resolve_input_attachments(
+        self,
+        run_input: dict[str, Any],
+        *,
+        tenant_id: str,
+        run_id: str | None = None,
+    ) -> list[Attachment]:
+        """Load attachments referenced by ``input.attachments`` (id list or refs)."""
+        raw = run_input.get("attachments")
+        if not isinstance(raw, list) or not raw:
+            return []
+        ids: list[str] = []
+        for item in raw:
+            if isinstance(item, str):
+                ids.append(item)
+            elif isinstance(item, dict) and isinstance(item.get("id"), str):
+                ids.append(item["id"])
+        if not ids:
+            return []
+        if run_id:
+            return await self.bind_to_run(
+                run_id=run_id, tenant_id=tenant_id, attachment_ids=ids
+            )
+        out: list[Attachment] = []
+        for attachment_id in ids:
+            out.append(await self.get(attachment_id, tenant_id=tenant_id))
+        return out
+
+    async def erase_run(self, run_id: str, *, tenant_id: str | None = None) -> int:
+        """Delete attachment rows + blobs for a run. Does not commit."""
+        stmt = select(Attachment).where(Attachment.run_id == run_id)
+        if tenant_id is not None:
+            stmt = stmt.where(Attachment.tenant_id == tenant_id)
+        result = await self.session.execute(stmt)
+        rows = list(result.scalars().all())
+        for row in rows:
+            try:
+                await self.store.delete(row.storage_key)
+            except Exception:
+                pass
+        if rows:
+            await self.session.execute(
+                delete(Attachment).where(Attachment.id.in_([r.id for r in rows]))
+            )
+        return len(rows)
+
+    async def erase_tenant(self, tenant_id: str) -> int:
+        """Delete all attachments for a tenant. Does not commit."""
+        stmt = select(Attachment).where(Attachment.tenant_id == tenant_id)
+        result = await self.session.execute(stmt)
+        rows = list(result.scalars().all())
+        for row in rows:
+            try:
+                await self.store.delete(row.storage_key)
+            except Exception:
+                pass
+        if rows:
+            await self.session.execute(
+                delete(Attachment).where(Attachment.tenant_id == tenant_id)
+            )
+        return len(rows)
+
+    def to_read(self, attachment: Attachment) -> AttachmentRead:
+        return AttachmentRead(
+            id=attachment.id,
+            tenant_id=attachment.tenant_id,
+            run_id=attachment.run_id,
+            message_id=attachment.message_id,
+            media_type=attachment.media_type,
+            filename=attachment.filename,
+            size_bytes=attachment.size_bytes,
+            sha256=attachment.sha256,
+            caption=attachment.caption,
+            url=attachment_url(attachment.id),
+            created_at=attachment.created_at,
+            updated_at=attachment.updated_at,
+        )
+
+
+class AttachmentErasureHook:
+    """Retention hook: drop attachment rows + object-store blobs with L0 erase."""
+
+    async def erase_run(self, session: AsyncSession, run: Any) -> dict[str, int]:
+        service = AttachmentService(session)
+        deleted = await service.erase_run(
+            run.id, tenant_id=getattr(run, "tenant_id", None)
+        )
+        return {"attachments_deleted": deleted}
+
+    async def erase_tenant(self, session: AsyncSession, tenant_id: str) -> dict[str, int]:
+        service = AttachmentService(session)
+        deleted = await service.erase_tenant(tenant_id)
+        return {"attachments_deleted": deleted}
+
+
+def refs_from_attachments(attachments: list[Attachment]) -> list[dict[str, Any]]:
+    return [attachment_ref(a) for a in attachments]
diff --git a/backend/app/services/retention_service.py b/backend/app/services/retention_service.py
index 49ea6ee..641567e 100644
--- a/backend/app/services/retention_service.py
+++ b/backend/app/services/retention_service.py
@@ -40,6 +40,15 @@ def register_memory_erasure_hook(hook: MemoryErasureHook) -> None:
     _memory_erasure_hooks.append(hook)
 
 
+def _register_builtin_hooks() -> None:
+    from app.services.attachment_service import AttachmentErasureHook
+
+    register_memory_erasure_hook(AttachmentErasureHook())
+
+
+_register_builtin_hooks()
+
+
 class RetentionService:
     def __init__(
         self,
@@ -79,7 +88,15 @@ async def erase_run_data(
             tenant_id=run.tenant_id,
             **counts,
         )
-        return {"messages_deleted": counts["messages"], "checkpoints_deleted": counts["checkpoints"]}
+        return {
+            "messages_deleted": counts["messages"],
+            "checkpoints_deleted": counts["checkpoints"],
+            **{
+                key: value
+                for key, value in counts.items()
+                if key not in ("messages", "checkpoints")
+            },
+        }
 
     async def erase_tenant_data(self, tenant_id: str) -> dict[str, int]:
         stmt = select(Run.id).where(Run.tenant_id == tenant_id)
diff --git a/backend/app/services/run_service.py b/backend/app/services/run_service.py
index 396dd02..6a5ea10 100644
--- a/backend/app/services/run_service.py
+++ b/backend/app/services/run_service.py
@@ -68,6 +68,14 @@ class AgentNotFound(Exception):
     pass
 
 
+class AttachmentRefNotFound(Exception):
+    """Raised when ``input.attachments`` references a missing / cross-tenant id."""
+
+    def __init__(self, attachment_id: str) -> None:
+        self.attachment_id = attachment_id
+        super().__init__(attachment_id)
+
+
 class ThreadNotFound(Exception):
     pass
 
@@ -140,6 +148,22 @@ async def create_run(
             ):
                 raise ThreadNotFound(thread_id)
 
+        from app.services.attachment_service import (
+            AttachmentNotFound,
+            AttachmentService,
+        )
+
+        # Validate attachment refs before inserting the run row.
+        attachment_service = AttachmentService(self.session)
+        try:
+            await attachment_service.resolve_input_attachments(
+                dict(payload.input or {}),
+                tenant_id=agent.tenant_id,
+                run_id=None,
+            )
+        except AttachmentNotFound as exc:
+            raise AttachmentRefNotFound(exc.attachment_id) from exc
+
         run = Run(
             tenant_id=agent.tenant_id,
             project_id=agent.project_id,
@@ -154,6 +178,12 @@ async def create_run(
         await self.session.commit()
         await self.session.refresh(run)
 
+        await attachment_service.resolve_input_attachments(
+            dict(payload.input or {}),
+            tenant_id=agent.tenant_id,
+            run_id=run.id,
+        )
+
         await self._broadcast("run.created", run.id, {"agent_id": agent.id})
         return run
 
@@ -557,6 +587,7 @@ async def _handle_event(
                 step = await self._find_step(run_id, raw_step_index)
                 if step is not None:
                     step_id = step.id
+            extra = dict(data.get("extra") or {})
             message = Message(
                 run_id=run_id,
                 index=index,
@@ -565,14 +596,31 @@ async def _handle_event(
                 name=data.get("name"),
                 content=data.get("content", ""),
                 tool_call_id=data.get("tool_call_id"),
-                extra=data.get("extra", {}),
+                extra=extra,
             )
             self.session.add(message)
             await self.session.commit()
+            await self.session.refresh(message)
+            attachment_ids = [
+                str(item.get("id"))
+                for item in (extra.get("attachments") or [])
+                if isinstance(item, dict) and item.get("id")
+            ]
+            if attachment_ids:
+                from app.services.attachment_service import AttachmentService
+
+                run = await self._get_run(run_id)
+                await AttachmentService(self.session).bind_message(
+                    attachment_ids=attachment_ids,
+                    message_id=message.id,
+                    tenant_id=run.tenant_id,
+                )
             data = {
                 **data,
+                "id": message.id,
                 "index": index,
                 "step_id": step_id,
+                "extra": extra,
             }
         elif event_type == "tool_call.started":
             step = await self._find_step(run_id, data["step_index"])
diff --git a/backend/app/services/thread_service.py b/backend/app/services/thread_service.py
index 921bb51..1bc2372 100644
--- a/backend/app/services/thread_service.py
+++ b/backend/app/services/thread_service.py
@@ -248,7 +248,7 @@ async def load_thread_window(
         messages: list[dict[str, Any]] = []
         for msg, _run in result.all():
             extra = msg.extra or {}
-            if extra.get("kind") in ("prompt_echo", "reasoning"):
+            if extra.get("kind") in ("prompt_echo", "reasoning", "attachment"):
                 continue
             payload = message_row_to_dict(msg)
             # Drop incomplete tool-call metadata from prior runs.
diff --git a/backend/app/worker/executor.py b/backend/app/worker/executor.py
index c22b4fd..7173e3e 100644
--- a/backend/app/worker/executor.py
+++ b/backend/app/worker/executor.py
@@ -146,6 +146,10 @@ async def execute(self, run_id: str, adapter_name: str) -> None:
                     agent_config=agent_config,
                 )
 
+            from app.services.attachment_service import AttachmentService
+
+            attachments = await AttachmentService(session).list_for_run(run.id)
+
             run.status = RunStatus.RUNNING
             await session.commit()
             await service._broadcast("run.started", run.id, {})
@@ -168,6 +172,7 @@ async def _emit(event_type: EventType, data: dict[str, Any]) -> None:
                 run_messages=run_messages,
                 thread_id=run.thread_id,
                 thread_messages=thread_messages,
+                attachments=attachments,
                 step_index_base=step_index_base,
                 emit=_emit,
             )
diff --git a/backend/tests/test_attachments.py b/backend/tests/test_attachments.py
new file mode 100644
index 0000000..450433b
--- /dev/null
+++ b/backend/tests/test_attachments.py
@@ -0,0 +1,207 @@
+"""Multimodal attachment upload, bind, stream, and erase."""
+
+from __future__ import annotations
+
+import asyncio
+from io import BytesIO
+from pathlib import Path
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.adapters import register_adapter
+from app.adapters.base import AdapterContext, AdapterResult, OrchestratorAdapter
+from app.models.run import RunStatus
+from app.runtime.object_store import LocalObjectStore, set_object_store
+from ulid import ULID
+
+
+@pytest.fixture
+def attachment_store(tmp_path: Path):
+    store = LocalObjectStore(tmp_path / "attachments")
+    set_object_store(store)
+    yield store
+    set_object_store(None)
+
+
+async def _poll_until(client, run_id: str, statuses: set[str], *, attempts: int = 80):
+    body: dict = {}
+    for _ in range(attempts):
+        detail = await client.get(f"/v1/runs/{run_id}")
+        body = detail.json()
+        if body["status"] in statuses:
+            return body
+        await asyncio.sleep(0.05)
+    return body
+
+
+@pytest.mark.asyncio
+async def test_upload_download_and_bind_attachment(client, attachment_store):
+    png = (
+        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01"
+        b"\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde"
+        b"\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f\x00\x00\x01\x01\x00\x05\x18\xd8N"
+        b"\x00\x00\x00\x00IEND\xaeB`\x82"
+    )
+    upload = await client.post(
+        "/v1/attachments",
+        files={"file": ("pixel.png", BytesIO(png), "image/png")},
+        data={"caption": "tiny"},
+    )
+    assert upload.status_code == 201, upload.text
+    meta = upload.json()
+    assert meta["filename"] == "pixel.png"
+    assert meta["media_type"] == "image/png"
+    assert meta["size_bytes"] == len(png)
+    assert meta["caption"] == "tiny"
+    assert meta["url"] == f"/v1/attachments/{meta['id']}"
+
+    content = await client.get(f"/v1/attachments/{meta['id']}/content")
+    assert content.status_code == 200
+    assert content.content == png
+    assert content.headers["content-type"].startswith("image/png")
+
+    agent = await client.post(
+        "/v1/agents",
+        json={"name": "attach-bot", "adapter": "echo", "config": {"delay": 0}},
+    )
+    assert agent.status_code == 201, agent.text
+    run = await client.post(
+        "/v1/runs",
+        json={
+            "agent_id": agent.json()["id"],
+            "input": {
+                "prompt": "describe",
+                "attachments": [{"id": meta["id"]}],
+            },
+        },
+    )
+    assert run.status_code == 202, run.text
+    body = await _poll_until(client, run.json()["id"], {"succeeded", "failed"})
+    assert body["status"] == "succeeded"
+
+    refreshed = await client.get(f"/v1/attachments/{meta['id']}")
+    assert refreshed.status_code == 200
+    assert refreshed.json()["run_id"] == body["id"]
+
+
+@pytest.mark.asyncio
+async def test_langgraph_streams_and_persists_attachments(client, attachment_store):
+    adapter_name = f"attach-capture-{ULID()}"
+    captured: list[AdapterContext] = []
+
+    class CaptureAdapter(OrchestratorAdapter):
+        name = adapter_name
+
+        async def run(self, ctx: AdapterContext) -> AdapterResult:
+            captured.append(ctx)
+            from app.runtime.attachments import emit_input_attachments
+
+            await ctx.emit_step_started(index=0, node="vision")
+            if ctx.attachments:
+                await emit_input_attachments(ctx, list(ctx.attachments), step_index=0)
+            prompt = str(ctx.input.get("prompt") or "")
+            await ctx.emit_message(role="user", content=prompt, step_index=0)
+            await ctx.emit_message(role="assistant", content=f"saw:{prompt}", step_index=0)
+            await ctx.emit_step_completed(
+                index=0, node="vision", output={"reply": f"saw:{prompt}"}
+            )
+            return AdapterResult(
+                status=RunStatus.SUCCEEDED, output={"reply": f"saw:{prompt}"}
+            )
+
+    register_adapter(adapter_name, CaptureAdapter())
+
+    png = b"fakepng-bytes"
+    upload = await client.post(
+        "/v1/attachments",
+        files={"file": ("chart.png", BytesIO(png), "image/png")},
+    )
+    assert upload.status_code == 201, upload.text
+    attachment_id = upload.json()["id"]
+
+    agent = await client.post(
+        "/v1/agents",
+        json={"name": "vision-bot", "adapter": adapter_name, "config": {}},
+    )
+    assert agent.status_code == 201, agent.text
+    run = await client.post(
+        "/v1/runs",
+        json={
+            "agent_id": agent.json()["id"],
+            "input": {
+                "prompt": "what is this?",
+                "attachments": [attachment_id],
+            },
+        },
+    )
+    assert run.status_code == 202, run.text
+    body = await _poll_until(client, run.json()["id"], {"succeeded", "failed"})
+    assert body["status"] == "succeeded"
+    assert len(captured) == 1
+    assert len(captured[0].attachments) == 1
+    assert captured[0].attachments[0].id == attachment_id
+
+    messages = body["messages"]
+    attachment_msgs = [
+        m for m in messages if (m.get("extra") or {}).get("kind") == "attachment"
+    ]
+    assert len(attachment_msgs) == 1
+    refs = attachment_msgs[0]["extra"]["attachments"]
+    assert refs[0]["id"] == attachment_id
+    assert refs[0]["filename"] == "chart.png"
+
+
+@pytest.mark.asyncio
+async def test_erase_run_deletes_attachment_blob(client, attachment_store, tmp_path):
+    png = b"erase-me"
+    upload = await client.post(
+        "/v1/attachments",
+        files={"file": ("gone.png", BytesIO(png), "image/png")},
+    )
+    assert upload.status_code == 201, upload.text
+    attachment_id = upload.json()["id"]
+    storage_key = None
+    # Resolve storage path after bind.
+    agent = await client.post(
+        "/v1/agents",
+        json={"name": "erase-attach", "adapter": "echo", "config": {"delay": 0}},
+    )
+    run = await client.post(
+        "/v1/runs",
+        json={
+            "agent_id": agent.json()["id"],
+            "input": {"prompt": "x", "attachments": [attachment_id]},
+        },
+    )
+    body = await _poll_until(client, run.json()["id"], {"succeeded", "failed"})
+    assert body["status"] == "succeeded"
+
+    meta = await client.get(f"/v1/attachments/{attachment_id}")
+    assert meta.status_code == 200
+
+    erase = await client.post(f"/v1/runs/{body['id']}/erase")
+    assert erase.status_code == 200, erase.text
+
+    missing = await client.get(f"/v1/attachments/{attachment_id}")
+    assert missing.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_missing_attachment_ref_returns_404(client, attachment_store):
+    agent = await client.post(
+        "/v1/agents",
+        json={"name": "missing-attach", "adapter": "echo", "config": {"delay": 0}},
+    )
+    assert agent.status_code == 201
+    run = await client.post(
+        "/v1/runs",
+        json={
+            "agent_id": agent.json()["id"],
+            "input": {
+                "prompt": "x",
+                "attachments": [{"id": "01MISSINGATTACHMENTID0000"}],
+            },
+        },
+    )
+    assert run.status_code == 404
diff --git a/docs/api-contract.md b/docs/api-contract.md
index 79719d4..fb9e19e 100644
--- a/docs/api-contract.md
+++ b/docs/api-contract.md
@@ -156,6 +156,32 @@ The first response is the newest page; `next_cursor` is an opaque
 Response: `Run[]` belonging to the thread, oldest first. 404 if the thread is
 not visible.
 
+### `POST /v1/attachments` → 201
+
+Multipart upload (`file` required, optional `caption`). Operator role.
+Persists bytes in the object store and returns metadata:
+
+```json
+{
+  "id": "01HZ...",
+  "tenant_id": "default",
+  "run_id": null,
+  "message_id": null,
+  "media_type": "image/png",
+  "filename": "chart.png",
+  "size_bytes": 1234,
+  "sha256": "…",
+  "caption": null,
+  "url": "/v1/attachments/01HZ...",
+  "created_at": "…",
+  "updated_at": "…"
+}
+```
+
+`GET /v1/attachments/{id}` returns the same metadata. `GET /v1/attachments/{id}/content`
+streams the binary. Cross-tenant ids return 404. Max size:
+`AGENTFLOW_ATTACHMENT_MAX_BYTES` (default 10 MiB).
+
 ### `POST /v1/runs` → 202
 
 Request:
@@ -163,13 +189,19 @@ Request:
 ```json
 {
   "agent_id": "01HZ...",
-  "input": { "prompt": "hi" },
+  "input": {
+    "prompt": "hi",
+    "attachments": [{ "id": "01HZ..." }]
+  },
   "metadata": {},
   "adapter": "echo",
   "thread_id": "01HZ..."
 }
 ```
 
+`input.attachments` may be a list of ids or `{ "id": "…" }` objects referencing
+previously uploaded attachments. Missing / cross-tenant ids → 404.
+
 `metadata`, `adapter`, and `thread_id` are optional. When `adapter` is omitted
 the agent's default adapter is used. When `thread_id` is set, the run is linked
 to that thread and the worker seeds prior thread turns into
@@ -384,11 +416,12 @@ The supported `type` values are:
 { "step_index": 0, "delta": "Hel", "role": "assistant", "part": "text" }
 ```
 
-`part` discriminates fine-grained streams: `"text"` (default, visible reply)
-or `"reasoning"` (model thinking / chain-of-thought). Finished reasoning is
-persisted as a `message.created` with `extra.kind = "reasoning"` so the
-console can show a collapsible block after reconnect; live deltas remain
-ephemeral.
+`part` discriminates fine-grained streams: `"text"` (default, visible reply),
+`"reasoning"` (model thinking / chain-of-thought), or `"attachment"` (multimodal
+file announcement; payload may include an `attachment` object). Finished
+reasoning is persisted as a `message.created` with `extra.kind = "reasoning"`;
+finished attachments as `extra.kind = "attachment"` plus
+`extra.attachments` refs. Live deltas remain ephemeral.
 
 `step.updated` flushes deferred metrics on a running step (tokens, latency)
 before `step.completed`:
@@ -607,6 +640,11 @@ ops dashboards.
 (content is the full chain-of-thought). These rows stay on the Run transcript
 for the console but are excluded from Thread L1 window seeding.
 
+`extra.kind = "attachment"` marks a multimodal attachment message. Content is
+a caption / filename summary; `extra.attachments` holds compact refs
+(`id`, `media_type`, `filename`, `url`, `size_bytes`). Binary bytes stay in
+object storage — `Message.content` is never multimodal JSON.
+
 ### `MessagePage`
 
 ```ts
diff --git a/docs/architecture.md b/docs/architecture.md
index 16372d9..5f70e8d 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -246,7 +246,21 @@ so operators see thinking as it streams. Thread L1 seeding skips reasoning
 rows (same as `prompt_echo`) so chain-of-thought is not replayed into the next
 prompt.
 
-Multimodal attachment persistence is still outstanding.
+## Fine-grained streaming (multimodal attachments)
+
+Attachments are uploaded first (`POST /v1/attachments`), then referenced from
+`POST /v1/runs` via `input.attachments: [{ "id": "…" }]`. Bytes live in a local
+object store (`AGENTFLOW_ATTACHMENT_STORAGE_DIR`); Postgres only stores metadata
+(`attachments` table). The same store is reserved for future memory document
+chunks — Message.content stays plain text.
+
+When a run starts, LangGraph (and adapters that call
+`emit_input_attachments`) streams `token.delta` with `part: "attachment"` for
+each file, then persists a `message.created` with
+`extra.kind = "attachment"` and `extra.attachments` refs. Image bytes are loaded
+into OpenAI-compatible multimodal `image_url` parts for the model call.
+Retention erase deletes attachment rows and blobs with L0 memory. Thread L1
+skips attachment rows (captions stay on the Run transcript for the console).
 
 ## Unit tests
 
diff --git a/docs/data-model.md b/docs/data-model.md
index b2e0717..5e884a3 100644
--- a/docs/data-model.md
+++ b/docs/data-model.md
@@ -16,8 +16,10 @@ erDiagram
     Run ||--o{ Step : has
     Run ||--o{ Message : has
     Run ||--o{ Checkpoint : has
+    Run ||--o{ Attachment : has
     Step ||--o{ ToolCall : has
     Step }o--|| Message : "optional step_id"
+    Message ||--o{ Attachment : "optional message_id"
 
     Project {
         string id PK
@@ -104,6 +106,18 @@ erDiagram
         string label
         json state
     }
+    Attachment {
+        string id PK
+        string tenant_id
+        string run_id FK
+        string message_id FK
+        string media_type
+        string filename
+        string storage_key
+        int size_bytes
+        string sha256
+        text caption
+    }
 ```
 
 ## Status state machine
@@ -146,6 +160,10 @@ stateDiagram-v2
 - **`Message.extra.kind = "reasoning"`** persists a finished thinking block
   (content = full reasoning text). Live chunks use SSE `token.delta` with
   `part: "reasoning"` and are not written to Redis replay or Postgres.
+- **`Message.extra.kind = "attachment"`** persists multimodal attachment refs
+  (`extra.attachments`). Bytes live in object storage keyed by
+  `attachments.storage_key`; `Message.content` stays plain text (caption).
+  Live announcements use SSE `token.delta` with `part: "attachment"`.
 - **`Thread` groups Runs for L1 short memory.** `Run.thread_id` is optional;
   when set, the worker seeds `AdapterContext.thread_messages` from prior runs
   in the same thread (window-trimmed). Messages remain Run-scoped rows.
@@ -176,5 +194,8 @@ stateDiagram-v2
 | `ix_messages_run_index` | stream messages in order |
 | `ix_tool_calls_step` | render tool calls inside a step |
 | `ix_checkpoints_run_index` | replay from the latest checkpoint |
+| `ix_attachments_tenant_id` | tenant-scoped attachment lookup |
+| `ix_attachments_run_id` | list attachments for a run / erase |
+| `ix_attachments_storage_key` | unique object-store key |
 | `ix_agent_versions_agent_id` | list version history for an agent |
 | `uq_agent_versions_agent_version` | one snapshot per (agent, version) |
diff --git a/docs/plan.md b/docs/plan.md
index 8aaaa2f..b4a7ce8 100644
--- a/docs/plan.md
+++ b/docs/plan.md
@@ -2,7 +2,7 @@
 
 架构与数据模型见 [architecture.md](architecture.md) 与 [data-model.md](data-model.md)。
 
-**最后核对：** 2026-09-03
+**最后核对：** 2026-09-04
 
 ## 优化方向
 
@@ -79,7 +79,7 @@
 - [ ] Agent 记忆服务（M1 Thread ✅；M2/M3 情景摘要 + 语义事实 + 文档 RAG 待做）
 - [x] 模型路由 / fallback 策略
 - [ ] 定时与批量 Run
-- [x] 细粒度流式：推理块（`token.delta.part` + `Message.extra.kind=reasoning`）；多模态附件（DB 持久化）待做
+- [x] 细粒度流式：推理块（`token.delta.part` + `Message.extra.kind=reasoning`）；多模态附件（`token.delta.part=attachment` + `attachments` 表 / 对象存储 + `Message.extra.kind=attachment`）
 
 ## Agent Memory 专项
 
@@ -165,7 +165,7 @@ L0 属于执行与审计；L1–L3 才是「Agent Memory」。L0 不能删，但
 - 事实冲突：同一 `key`（如 `user.preferred_language`）新写入覆盖旧值并留 `superseded_by`。
 - 记忆评测：回归套件增加「是否错误召回过期事实 / 是否漏召回」用例，挂到现有 `regression-executions`。
 - Procedural：把成功的 tool 序列收成 `kind=procedure`，仅对同 agent + 相似 input 检索；默认关闭。
-- 多模态附件：与 Phase 5 流式附件共用对象存储，记忆层只存引用与 caption embedding。
+- 多模态附件：与 Phase 5 流式附件共用对象存储（`attachments` 表 + local FS；记忆层日后只存引用与 caption embedding）✅
 
 ---
 
diff --git a/frontend/components/EventStream.tsx b/frontend/components/EventStream.tsx
index bb7597a..1fd6c84 100644
--- a/frontend/components/EventStream.tsx
+++ b/frontend/components/EventStream.tsx
@@ -26,7 +26,12 @@ function formatEventData(type: string, data: Record<string, unknown>): string {
     return parts.join(" · ");
   }
   if (type === "token.delta") {
-    const part = data.part === "reasoning" ? "reasoning" : "text";
+    const part =
+      data.part === "reasoning"
+        ? "reasoning"
+        : data.part === "attachment"
+          ? "attachment"
+          : "text";
     const delta = typeof data.delta === "string" ? data.delta : "";
     const preview = delta.length > 40 ? `${delta.slice(0, 40)}…` : delta;
     return `part=${part} · ${JSON.stringify(preview)}`;
diff --git a/frontend/components/MessagesPanel.tsx b/frontend/components/MessagesPanel.tsx
index e403dfa..30639d5 100644
--- a/frontend/components/MessagesPanel.tsx
+++ b/frontend/components/MessagesPanel.tsx
@@ -21,9 +21,27 @@ function isReasoning(message: Message): boolean {
   return kind === "reasoning" || kind === "streaming_reasoning";
 }
 
+function isAttachment(message: Message): boolean {
+  const kind = message.extra?.kind;
+  if (kind === "attachment" || kind === "streaming_attachment") return true;
+  const attachments = message.extra?.attachments;
+  return Array.isArray(attachments) && attachments.length > 0;
+}
+
 function isStreaming(message: Message): boolean {
   const kind = message.extra?.kind;
-  return kind === "streaming" || kind === "streaming_reasoning";
+  return (
+    kind === "streaming" ||
+    kind === "streaming_reasoning" ||
+    kind === "streaming_attachment"
+  );
+}
+
+function attachmentList(message: Message): Array<Record<string, unknown>> {
+  const raw = message.extra?.attachments;
+  return Array.isArray(raw)
+    ? raw.filter((item): item is Record<string, unknown> => !!item && typeof item === "object")
+    : [];
 }
 
 function mergeMessages(...groups: Message[][]): Message[] {
@@ -118,6 +136,10 @@ export function MessagesPanel({
     () => displayed.filter(isReasoning).length,
     [displayed],
   );
+  const attachmentCount = useMemo(
+    () => displayed.filter(isAttachment).length,
+    [displayed],
+  );
 
   async function loadOlder() {
     if (loading || !hasMore) return;
@@ -160,6 +182,9 @@ export function MessagesPanel({
             {reasoningCount > 0
               ? ` · ${reasoningCount} reasoning block${reasoningCount === 1 ? "" : "s"}`
               : ""}
+            {attachmentCount > 0
+              ? ` · ${attachmentCount} attachment message${attachmentCount === 1 ? "" : "s"}`
+              : ""}
             {hasMore ? " · older available" : ""}
           </span>
         ) : null}
@@ -214,6 +239,68 @@ export function MessagesPanel({
               />
             );
           }
+          if (isAttachment(message)) {
+            const items = attachmentList(message);
+            return (
+              <li
+                key={key}
+                className="rounded border border-border bg-surface p-3 text-sm"
+              >
+                <div className="mb-2 text-xs uppercase tracking-wide text-muted">
+                  {message.role} · attachment
+                  {isStreaming(message) ? " · streaming…" : ""}
+                  <span className="ml-2 font-mono normal-case">
+                    #{message.index}
+                  </span>
+                </div>
+                {message.content ? (
+                  <div className="mb-2 whitespace-pre-wrap">{message.content}</div>
+                ) : null}
+                <ul className="space-y-2">
+                  {items.map((item, idx) => {
+                    const id = typeof item.id === "string" ? item.id : `att-${idx}`;
+                    const filename =
+                      typeof item.filename === "string" ? item.filename : id;
+                    const mediaType =
+                      typeof item.media_type === "string" ? item.media_type : "";
+                    const url =
+                      typeof item.url === "string"
+                        ? `/api${item.url}/content`
+                        : null;
+                    const isImage = mediaType.startsWith("image/");
+                    return (
+                      <li
+                        key={id}
+                        className="rounded border border-dashed border-border p-2"
+                      >
+                        <div className="text-xs text-muted">
+                          {filename}
+                          {mediaType ? ` · ${mediaType}` : ""}
+                        </div>
+                        {url && isImage ? (
+                          // eslint-disable-next-line @next/next/no-img-element
+                          <img
+                            src={url}
+                            alt={filename}
+                            className="mt-2 max-h-48 max-w-full rounded object-contain"
+                          />
+                        ) : url ? (
+                          <a
+                            className="mt-1 inline-block text-xs underline"
+                            href={url}
+                            target="_blank"
+                            rel="noreferrer"
+                          >
+                            Download
+                          </a>
+                        ) : null}
+                      </li>
+                    );
+                  })}
+                </ul>
+              </li>
+            );
+          }
           return (
             <li
               key={key}
diff --git a/frontend/lib/api.ts b/frontend/lib/api.ts
index af9adb9..125b4de 100644
--- a/frontend/lib/api.ts
+++ b/frontend/lib/api.ts
@@ -2,6 +2,7 @@ import type {
   Agent,
   AgentVersion,
   AgentVersionDiff,
+  Attachment,
   MessagePage,
   Run,
   Thread,
@@ -117,6 +118,25 @@ export const api = {
     const runs = await request<Run[]>(`/v1/threads/${id}/runs`);
     return runs.map(normalizeRun);
   },
+  uploadAttachment: async (file: File, caption?: string) => {
+    const body = new FormData();
+    body.append("file", file);
+    if (caption) body.append("caption", caption);
+    const response = await fetch("/api/v1/attachments", {
+      method: "POST",
+      headers: { ...authHeaders() },
+      body,
+      cache: "no-store",
+    });
+    if (!response.ok) {
+      const detail = await response.text();
+      throw new Error(`${response.status} ${response.statusText}: ${detail}`);
+    }
+    return response.json() as Promise<Attachment>;
+  },
+  getAttachment: (id: string) =>
+    request<Attachment>(`/v1/attachments/${id}`),
+  attachmentContentUrl: (id: string) => `/api/v1/attachments/${id}/content`,
   listAgents: () => request<Agent[]>("/v1/agents"),
   createAgent: (body: {
     name: string;
diff --git a/frontend/lib/run-events.ts b/frontend/lib/run-events.ts
index b47a52b..bc6371f 100644
--- a/frontend/lib/run-events.ts
+++ b/frontend/lib/run-events.ts
@@ -181,7 +181,10 @@ function appendMessage(messages: Message[], data: Record<string, unknown>, at: s
   return [...messages, message].sort((a, b) => a.index - b.index);
 }
 
-function streamingDraftId(stepIndex: number, part: "text" | "reasoning"): string {
+function streamingDraftId(
+  stepIndex: number,
+  part: "text" | "reasoning" | "attachment",
+): string {
   return `sse-stream-${stepIndex}-${part}`;
 }
 
@@ -192,10 +195,84 @@ function appendTokenDelta(
 ): Message[] {
   const stepIndex = data.step_index;
   if (typeof stepIndex !== "number") return messages;
+
+  const part =
+    data.part === "reasoning"
+      ? "reasoning"
+      : data.part === "attachment"
+        ? "attachment"
+        : "text";
+
+  if (part === "attachment") {
+    const attachment = asRecord(data.attachment);
+    if (!attachment.id && typeof data.delta !== "string") return messages;
+    const draftId = streamingDraftId(stepIndex, "attachment");
+    const existing = messages.find((message) => message.id === draftId);
+    const nextAttachment = {
+      id: typeof attachment.id === "string" ? attachment.id : String(data.delta ?? ""),
+      media_type:
+        typeof attachment.media_type === "string" ? attachment.media_type : "",
+      filename:
+        typeof attachment.filename === "string"
+          ? attachment.filename
+          : typeof data.delta === "string"
+            ? data.delta
+            : "attachment",
+      size_bytes:
+        typeof attachment.size_bytes === "number" ? attachment.size_bytes : 0,
+      url: typeof attachment.url === "string" ? attachment.url : undefined,
+      caption:
+        typeof attachment.caption === "string" ? attachment.caption : null,
+    };
+    if (existing) {
+      const prev = Array.isArray(existing.extra.attachments)
+        ? [...(existing.extra.attachments as Record<string, unknown>[])]
+        : [];
+      if (!prev.some((item) => item.id === nextAttachment.id)) {
+        prev.push(nextAttachment);
+      }
+      return messages.map((message) =>
+        message.id === draftId
+          ? {
+              ...message,
+              content: prev
+                .map((item) => String(item.filename || item.id || ""))
+                .join("; "),
+              extra: {
+                ...message.extra,
+                kind: "streaming_attachment",
+                attachments: prev,
+              },
+              created_at: at,
+            }
+          : message,
+      );
+    }
+    const index =
+      messages.length > 0
+        ? Math.max(...messages.map((message) => message.index)) + 1
+        : 0;
+    const draft: Message = {
+      id: draftId,
+      index,
+      step_id: null,
+      role: "user",
+      name: null,
+      content: String(nextAttachment.filename),
+      tool_call_id: null,
+      extra: {
+        kind: "streaming_attachment",
+        step_index: stepIndex,
+        attachments: [nextAttachment],
+      },
+      created_at: at,
+    };
+    return [...messages, draft].sort((a, b) => a.index - b.index);
+  }
+
   const delta = typeof data.delta === "string" ? data.delta : "";
   if (!delta) return messages;
 
-  const part = data.part === "reasoning" ? "reasoning" : "text";
   const draftId = streamingDraftId(stepIndex, part);
   const existing = messages.find((message) => message.id === draftId);
   if (existing) {
@@ -234,9 +311,15 @@ function dropStreamingDrafts(
 ): Message[] {
   const stepIndex =
     typeof data.step_index === "number" ? data.step_index : null;
+  const kind = asRecord(data.extra).kind;
+  // Attachment messages may omit step_index; always drop attachment drafts.
+  if (kind === "attachment") {
+    return messages.filter(
+      (message) => !message.id.startsWith("sse-stream-") || !message.id.endsWith("-attachment"),
+    );
+  }
   if (stepIndex == null) return messages;
 
-  const kind = asRecord(data.extra).kind;
   const dropIds = new Set<string>();
   if (kind === "reasoning") {
     dropIds.add(streamingDraftId(stepIndex, "reasoning"));
diff --git a/frontend/lib/types.ts b/frontend/lib/types.ts
index be5a7fe..c39004f 100644
--- a/frontend/lib/types.ts
+++ b/frontend/lib/types.ts
@@ -61,6 +61,21 @@ export interface MessagePage {
   has_more: boolean;
 }
 
+export interface Attachment {
+  id: string;
+  tenant_id: string;
+  run_id: string | null;
+  message_id: string | null;
+  media_type: string;
+  filename: string;
+  size_bytes: number;
+  sha256: string;
+  caption: string | null;
+  url: string;
+  created_at: string;
+  updated_at: string;
+}
+
 export interface Checkpoint {
   id: string;
   index: number;