diff --git a/internal/api/backup_destinations.go b/internal/api/backup_destinations.go
index 0be4b45..88da8ea 100644
--- a/internal/api/backup_destinations.go
+++ b/internal/api/backup_destinations.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"net/url"
 	"strings"
 
 	"github.com/flatrun/agent/internal/backup"
@@ -13,6 +14,12 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
+// managedStorePort is the fallback S3 API port used when a managed
+// destination's stored endpoint carries none. A managed object store is reached
+// over the shared object-storage network (see objectStorageNetworkName); the
+// agent, a host process, dials the container's address on that network.
+const managedStorePort = "9000"
+
 // buildStore resolves a destination plus its referenced credential into a
 // ready-to-use remote Store.
 func (s *Server) buildStore(dest config.BackupDestination) (backup.Store, error) {
@@ -25,9 +32,15 @@ func (s *Server) buildStore(dest config.BackupDestination) (backup.Store, error)
 		if cred.Kind != models.CredentialKindS3 {
 			return nil, fmt.Errorf("destination %q: credential %q is not an s3 credential", dest.Name, dest.CredentialID)
 		}
+		endpoint := dest.Endpoint
+		if dest.Kind == "managed" && dest.Deployment != "" {
+			if resolved, err := s.resolveManagedEndpoint(dest); err == nil && resolved != "" {
+				endpoint = resolved
+			}
+		}
 		return backup.NewS3Store(backup.S3Config{
 			Name:         dest.Name,
-			Endpoint:     dest.Endpoint,
+			Endpoint:     endpoint,
 			Region:       dest.Region,
 			Bucket:       dest.Bucket,
 			Prefix:       dest.Prefix,
@@ -40,6 +53,32 @@ func (s *Server) buildStore(dest config.BackupDestination) (backup.Store, error)
 	}
 }
 
+// resolveManagedEndpoint returns the address a managed store is reachable at
+// right now. A managed object store only exposes its port on an internal
+// compose network and its container IP changes across recreates, so the
+// endpoint is resolved live from the deployment rather than trusted from
+// stored config. The scheme and port of the stored endpoint are preserved.
+func (s *Server) resolveManagedEndpoint(dest config.BackupDestination) (string, error) {
+	if s.manager == nil {
+		return "", fmt.Errorf("docker manager unavailable")
+	}
+	ip, err := s.manager.ContainerPrimaryIP(dest.Deployment, objectStorageNetworkName(s.config))
+	if err != nil {
+		return "", err
+	}
+
+	u, err := url.Parse(dest.Endpoint)
+	if err != nil || u.Host == "" {
+		return "http://" + ip + ":" + managedStorePort, nil
+	}
+	port := u.Port()
+	if port == "" {
+		port = managedStorePort
+	}
+	u.Host = ip + ":" + port
+	return u.String(), nil
+}
+
 // applyBackupDestinations rebuilds the backup manager's remote stores from the
 // current config. It is called at startup and by the runtime applier when the
 // destinations config changes.
diff --git a/internal/api/backup_detect.go b/internal/api/backup_detect.go
new file mode 100644
index 0000000..8d03773
--- /dev/null
+++ b/internal/api/backup_detect.go
@@ -0,0 +1,301 @@
+package api
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/flatrun/agent/internal/backup"
+	"github.com/flatrun/agent/internal/docker"
+	"github.com/flatrun/agent/pkg/models"
+)
+
+// effectiveBackupSpec returns the deployment's configured backup spec, or, when
+// no databases are configured, one with databases auto-detected from the
+// deployment. This is what makes a database-server deployment (including the
+// shared infrastructure database) get a dump without any manual setup.
+func (s *Server) effectiveBackupSpec(d *models.Deployment) *backup.BackupSpec {
+	var spec *backup.BackupSpec
+	if d.Metadata != nil && d.Metadata.Backup != nil {
+		spec = d.Metadata.Backup
+	}
+	if spec != nil && len(spec.Databases) > 0 {
+		return spec
+	}
+
+	detected := s.detectBackupDatabases(d)
+	if len(detected) == 0 {
+		return spec
+	}
+
+	out := backup.BackupSpec{}
+	if spec != nil {
+		out = *spec
+	}
+	out.Databases = detected
+	// A database server's live data directory is captured by its dump, so drop
+	// it from the file copy: it is redundant and, taken hot, potentially
+	// inconsistent. Other data (app files, config) is still copied.
+	out.ExcludePatterns = mergeUnique(out.ExcludePatterns, s.dbServerDataDirs(d))
+	return &out
+}
+
+// dbServerDataDirs returns the bind-mount directory names of the deployment's
+// own database services, so the live database files can be excluded from the
+// file backup in favour of the logical dump.
+func (s *Server) dbServerDataDirs(d *models.Deployment) []string {
+	content, err := os.ReadFile(filepath.Join(d.Path, "docker-compose.yml"))
+	if err != nil {
+		return nil
+	}
+	compose, err := docker.ParseComposeYAML(string(content))
+	if err != nil {
+		return nil
+	}
+	services, ok := compose["services"].(map[string]interface{})
+	if !ok {
+		return nil
+	}
+
+	seen := map[string]bool{}
+	var dirs []string
+	for _, raw := range services {
+		svc, ok := raw.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		image, _ := svc["image"].(string)
+		if dbTypeFromImage(image) == "" {
+			continue
+		}
+		for _, host := range serviceBindHostPaths(svc) {
+			base := filepath.Base(host)
+			if base == "." || base == "/" || base == "" || seen[base] {
+				continue
+			}
+			seen[base] = true
+			dirs = append(dirs, base)
+		}
+	}
+	return dirs
+}
+
+func serviceBindHostPaths(svc map[string]interface{}) []string {
+	vols, ok := svc["volumes"].([]interface{})
+	if !ok {
+		return nil
+	}
+	var out []string
+	for _, v := range vols {
+		switch vv := v.(type) {
+		case string:
+			host := vv
+			if i := strings.Index(vv, ":"); i >= 0 {
+				host = vv[:i]
+			}
+			if strings.HasPrefix(host, ".") || strings.Contains(host, "/") {
+				out = append(out, host)
+			}
+		case map[string]interface{}:
+			if t, _ := vv["type"].(string); t == "bind" {
+				if src, _ := vv["source"].(string); src != "" {
+					out = append(out, src)
+				}
+			}
+		}
+	}
+	return out
+}
+
+func mergeUnique(a, b []string) []string {
+	seen := map[string]bool{}
+	var out []string
+	for _, list := range [][]string{a, b} {
+		for _, s := range list {
+			if s == "" || seen[s] {
+				continue
+			}
+			seen[s] = true
+			out = append(out, s)
+		}
+	}
+	return out
+}
+
+// detectBackupDatabases finds database-server services in a deployment's compose
+// and returns a spec to dump each in full (pg_dumpall / --all-databases) using
+// its root credentials. A shared database deployment thus backs up every app's
+// data in one dump; a standalone or sidecar database backs up its own.
+func (s *Server) detectBackupDatabases(d *models.Deployment) []models.DatabaseBackupSpec {
+	content, err := os.ReadFile(filepath.Join(d.Path, "docker-compose.yml"))
+	if err != nil {
+		return nil
+	}
+	compose, err := docker.ParseComposeYAML(string(content))
+	if err != nil {
+		return nil
+	}
+	services, ok := compose["services"].(map[string]interface{})
+	if !ok {
+		return nil
+	}
+
+	env, _ := s.readDeploymentEnvMap(d.Name)
+
+	var specs []models.DatabaseBackupSpec
+	for name, raw := range services {
+		svc, ok := raw.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		image, _ := svc["image"].(string)
+		dbType := dbTypeFromImage(image)
+		if dbType == "" {
+			continue
+		}
+		user, password := dbRootCreds(dbType, svc, env)
+		specs = append(specs, models.DatabaseBackupSpec{
+			Service:      name,
+			Type:         dbType,
+			Container:    docker.ContainerNameForService(string(content), d.Name, name),
+			AllDatabases: true,
+			User:         user,
+			Password:     password,
+		})
+	}
+
+	// Per-app slice: a deployment using the shared database dumps just its own
+	// database from the shared server, so its (typically more frequent) backups
+	// are self-contained rather than depending on the shared database's own,
+	// possibly less frequent, global backup.
+	specs = append(specs, s.detectSharedDatabaseSlices(d, env)...)
+	return specs
+}
+
+func (s *Server) detectSharedDatabaseSlices(d *models.Deployment, env map[string]string) []models.DatabaseBackupSpec {
+	if d.Metadata == nil {
+		return nil
+	}
+	shared := s.config.Infrastructure.Database
+	if shared.Container == "" {
+		return nil
+	}
+
+	var specs []models.DatabaseBackupSpec
+	for _, dbc := range d.Metadata.Databases {
+		if !dbc.IsShared && dbc.Mode != "shared" {
+			continue
+		}
+		dbName := ""
+		if dbc.DatabaseName != "" {
+			dbName = dbc.DatabaseName
+		} else if dbc.EnvPrefix != "" {
+			dbName = env[dbc.EnvPrefix+"_DATABASE"]
+		}
+		if dbName == "" {
+			dbName = env["DB_DATABASE"]
+		}
+		if dbName == "" {
+			continue
+		}
+		dbType := normalizeDBType(dbc.Type)
+		if dbType == "" {
+			dbType = normalizeDBType(shared.Type)
+		}
+		if dbType == "" {
+			continue
+		}
+		label := dbc.Alias
+		if label == "" {
+			label = "app"
+		}
+		specs = append(specs, models.DatabaseBackupSpec{
+			Service:   label,
+			Type:      dbType,
+			Container: shared.Container,
+			Database:  dbName,
+			User:      shared.RootUser,
+			Password:  shared.RootPassword,
+		})
+	}
+	return specs
+}
+
+func normalizeDBType(t string) string {
+	switch strings.ToLower(t) {
+	case "mysql", "mariadb":
+		return "mysql"
+	case "postgres", "postgresql":
+		return "postgres"
+	}
+	return ""
+}
+
+func dbTypeFromImage(image string) string {
+	img := strings.ToLower(image)
+	switch {
+	case strings.Contains(img, "postgres") || strings.Contains(img, "postgis"):
+		return "postgres"
+	case strings.Contains(img, "mariadb") || strings.Contains(img, "mysql") || strings.Contains(img, "percona"):
+		return "mysql"
+	}
+	return ""
+}
+
+func dbRootCreds(dbType string, svc map[string]interface{}, env map[string]string) (user, password string) {
+	switch dbType {
+	case "postgres":
+		user = serviceEnvValue(svc, env, "POSTGRES_USER")
+		if user == "" {
+			user = "postgres"
+		}
+		password = serviceEnvValue(svc, env, "POSTGRES_PASSWORD")
+	case "mysql":
+		user = "root"
+		password = serviceEnvValue(svc, env, "MYSQL_ROOT_PASSWORD")
+	}
+	return user, password
+}
+
+// serviceEnvValue resolves an env var for a service: the deployment's env file
+// wins, then the service's own environment block (resolving a ${VAR} reference
+// back against the env file).
+func serviceEnvValue(svc map[string]interface{}, env map[string]string, key string) string {
+	if v, ok := env[key]; ok && v != "" {
+		return v
+	}
+	switch e := svc["environment"].(type) {
+	case map[string]interface{}:
+		if raw, ok := e[key]; ok {
+			return resolveEnvRef(fmt.Sprint(raw), env)
+		}
+	case []interface{}:
+		for _, item := range e {
+			str, ok := item.(string)
+			if !ok {
+				continue
+			}
+			if k, val, found := strings.Cut(str, "="); found && k == key {
+				return resolveEnvRef(val, env)
+			}
+		}
+	}
+	return ""
+}
+
+func resolveEnvRef(v string, env map[string]string) string {
+	v = strings.TrimSpace(v)
+	if strings.HasPrefix(v, "${") && strings.HasSuffix(v, "}") {
+		inner := v[2 : len(v)-1]
+		name, def := inner, ""
+		if i := strings.Index(inner, ":-"); i >= 0 {
+			name, def = inner[:i], inner[i+2:]
+		}
+		if val, ok := env[name]; ok && val != "" {
+			return val
+		}
+		return def
+	}
+	return v
+}
diff --git a/internal/api/backup_detect_test.go b/internal/api/backup_detect_test.go
new file mode 100644
index 0000000..65b2a8f
--- /dev/null
+++ b/internal/api/backup_detect_test.go
@@ -0,0 +1,97 @@
+package api
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/flatrun/agent/pkg/config"
+	"github.com/flatrun/agent/pkg/models"
+)
+
+func TestDetectBackupDatabases_DBServerGlobalDump(t *testing.T) {
+	dir := t.TempDir()
+	srv := &Server{config: &config.Config{DeploymentsPath: dir}}
+
+	name := "shared-db"
+	dep := filepath.Join(dir, name)
+	if err := os.MkdirAll(dep, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	compose := "name: shared-db\nservices:\n  postgres:\n    image: postgres:16\n    environment:\n      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}\n"
+	os.WriteFile(filepath.Join(dep, "docker-compose.yml"), []byte(compose), 0o644)
+	os.WriteFile(filepath.Join(dep, ".env"), []byte("POSTGRES_PASSWORD=topsecret\n"), 0o600)
+
+	specs := srv.detectBackupDatabases(&models.Deployment{Name: name, Path: dep})
+	if len(specs) != 1 {
+		t.Fatalf("want 1 detected db, got %d: %#v", len(specs), specs)
+	}
+	s := specs[0]
+	if s.Type != "postgres" || !s.AllDatabases || s.User != "postgres" || s.Password != "topsecret" || s.Container == "" {
+		t.Fatalf("unexpected detected spec: %#v", s)
+	}
+}
+
+func TestDbTypeFromImage(t *testing.T) {
+	cases := map[string]string{
+		"postgres:16": "postgres", "postgis/postgis": "postgres",
+		"mariadb:11": "mysql", "mysql:8": "mysql", "redis:7": "", "nginx": "",
+	}
+	for img, want := range cases {
+		if got := dbTypeFromImage(img); got != want {
+			t.Errorf("%s: got %q want %q", img, got, want)
+		}
+	}
+}
+
+func TestDetectBackupDatabases_SharedAppSlice(t *testing.T) {
+	dir := t.TempDir()
+	srv := &Server{config: &config.Config{
+		DeploymentsPath: dir,
+		Infrastructure: config.InfrastructureConfig{Database: config.SharedDatabaseConfig{
+			Enabled: true, Type: "postgres", Container: "shared-pg", RootUser: "postgres", RootPassword: "rootpw",
+		}},
+	}}
+	name := "wordpress"
+	dep := filepath.Join(dir, name)
+	os.MkdirAll(dep, 0o755)
+	os.WriteFile(filepath.Join(dep, "docker-compose.yml"), []byte("services:\n  app:\n    image: wordpress:6\n"), 0o644)
+	os.WriteFile(filepath.Join(dep, ".env"), []byte("DB_DATABASE=wordpress_db\nDB_PASSWORD=apppw\n"), 0o600)
+
+	specs := srv.detectBackupDatabases(&models.Deployment{
+		Name: name, Path: dep,
+		Metadata: &models.ServiceMetadata{Databases: []models.DatabaseConfig{{Alias: "primary", Type: "postgres", Mode: "shared", IsShared: true}}},
+	})
+	if len(specs) != 1 {
+		t.Fatalf("want 1 slice, got %d: %#v", len(specs), specs)
+	}
+	s := specs[0]
+	if s.Container != "shared-pg" || s.Database != "wordpress_db" || s.User != "postgres" || s.Password != "rootpw" || s.AllDatabases {
+		t.Fatalf("unexpected slice: %#v", s)
+	}
+}
+
+func TestEffectiveBackupSpec_ExcludesDBDataDir(t *testing.T) {
+	dir := t.TempDir()
+	srv := &Server{config: &config.Config{DeploymentsPath: dir}}
+	name := "pg"
+	dep := filepath.Join(dir, name)
+	os.MkdirAll(dep, 0o755)
+	compose := "name: pg\nservices:\n  postgres:\n    image: postgres:16\n    volumes:\n      - ./data:/var/lib/postgresql/data\n    environment:\n      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}\n"
+	os.WriteFile(filepath.Join(dep, "docker-compose.yml"), []byte(compose), 0o644)
+	os.WriteFile(filepath.Join(dep, ".env"), []byte("POSTGRES_PASSWORD=pw\n"), 0o600)
+
+	spec := srv.effectiveBackupSpec(&models.Deployment{Name: name, Path: dep})
+	if spec == nil || len(spec.Databases) != 1 {
+		t.Fatalf("expected a detected database: %#v", spec)
+	}
+	found := false
+	for _, e := range spec.ExcludePatterns {
+		if e == "data" {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("expected 'data' excluded, got %#v", spec.ExcludePatterns)
+	}
+}
diff --git a/internal/api/backup_handlers.go b/internal/api/backup_handlers.go
index 697a651..3bc92f2 100644
--- a/internal/api/backup_handlers.go
+++ b/internal/api/backup_handlers.go
@@ -91,10 +91,7 @@ func (s *Server) createBackup(c *gin.Context) {
 		return
 	}
 
-	var spec *backup.BackupSpec
-	if deployment.Metadata != nil && deployment.Metadata.Backup != nil {
-		spec = deployment.Metadata.Backup
-	}
+	spec := s.effectiveBackupSpec(deployment)
 
 	jobID := s.backupManager.StartBackupJob(req.DeploymentName, spec)
 	c.JSON(http.StatusAccepted, gin.H{"job_id": jobID, "message": "Backup job started"})
@@ -113,10 +110,7 @@ func (s *Server) createDeploymentBackup(c *gin.Context) {
 		return
 	}
 
-	var spec *backup.BackupSpec
-	if deployment.Metadata != nil && deployment.Metadata.Backup != nil {
-		spec = deployment.Metadata.Backup
-	}
+	spec := s.effectiveBackupSpec(deployment)
 
 	jobID := s.backupManager.StartBackupJob(deploymentName, spec)
 	c.JSON(http.StatusAccepted, gin.H{"job_id": jobID, "message": "Backup job started"})
@@ -209,10 +203,7 @@ func (s *Server) getDeploymentBackupConfig(c *gin.Context) {
 		return
 	}
 
-	var spec *backup.BackupSpec
-	if deployment.Metadata != nil && deployment.Metadata.Backup != nil {
-		spec = deployment.Metadata.Backup
-	}
+	spec := s.effectiveBackupSpec(deployment)
 
 	c.JSON(http.StatusOK, gin.H{"backup_config": spec})
 }
diff --git a/internal/api/object_browser.go b/internal/api/object_browser.go
new file mode 100644
index 0000000..86cc6e9
--- /dev/null
+++ b/internal/api/object_browser.go
@@ -0,0 +1,236 @@
+package api
+
+import (
+	"fmt"
+	"net/http"
+	"path"
+	"strconv"
+
+	"github.com/flatrun/agent/internal/backup"
+	"github.com/flatrun/agent/pkg/config"
+	"github.com/gin-gonic/gin"
+)
+
+// bucketStatsCap bounds how many objects are counted per bucket for the bucket
+// list, so a very large bucket does not stall the page.
+const bucketStatsCap = 10000
+
+// defaultObjectPage is the object-listing page size when none is requested.
+const defaultObjectPage = 200
+
+// storeS3ByName resolves a registered object store by name into a usable S3
+// client (plus its destination record), writing an error response and returning
+// false on failure.
+func (s *Server) storeS3ByName(c *gin.Context) (*backup.S3Store, config.BackupDestination, bool) {
+	dest, ok := s.findDestinationByName(c.Param("name"))
+	if !ok {
+		c.JSON(http.StatusNotFound, gin.H{"error": "object store not found"})
+		return nil, config.BackupDestination{}, false
+	}
+	store, err := s.buildStore(dest)
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return nil, config.BackupDestination{}, false
+	}
+	s3store, ok := store.(*backup.S3Store)
+	if !ok {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "store is not S3-compatible"})
+		return nil, config.BackupDestination{}, false
+	}
+	return s3store, dest, true
+}
+
+// scopedStore returns the store scoped to the request's target bucket, which is
+// the ?bucket query when given, otherwise the store's configured bucket.
+func scopedStore(store *backup.S3Store, dest config.BackupDestination, c *gin.Context) *backup.S3Store {
+	bucket := c.Query("bucket")
+	if bucket == "" {
+		bucket = dest.Bucket
+	}
+	return store.WithBucket(bucket)
+}
+
+type bucketInfo struct {
+	Name      string `json:"name"`
+	Objects   int    `json:"objects"`
+	Size      int64  `json:"size"`
+	Truncated bool   `json:"truncated"`
+	IsBackup  bool   `json:"is_backup"`
+}
+
+// listStoreBuckets lists every bucket on the store's server with its object
+// count and total size, and marks the backup bucket.
+func (s *Server) listStoreBuckets(c *gin.Context) {
+	store, dest, ok := s.storeS3ByName(c)
+	if !ok {
+		return
+	}
+	ctx := c.Request.Context()
+	names, err := store.ListBuckets(ctx)
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+
+	out := make([]bucketInfo, 0, len(names))
+	for _, n := range names {
+		count, size, truncated, _ := store.WithBucket(n).BucketStats(ctx, bucketStatsCap)
+		out = append(out, bucketInfo{
+			Name:      n,
+			Objects:   count,
+			Size:      size,
+			Truncated: truncated,
+			IsBackup:  n == dest.Bucket,
+		})
+	}
+	c.JSON(http.StatusOK, gin.H{"buckets": out, "backup_bucket": dest.Bucket})
+}
+
+// deleteStoreBucket removes an empty bucket. The store's backup bucket is
+// protected, so browsing cannot destroy the backup target.
+func (s *Server) deleteStoreBucket(c *gin.Context) {
+	store, dest, ok := s.storeS3ByName(c)
+	if !ok {
+		return
+	}
+	bucket := c.Param("bucket")
+	if bucket == dest.Bucket {
+		c.JSON(http.StatusForbidden, gin.H{"error": fmt.Sprintf("the backup bucket %q cannot be deleted", dest.Bucket)})
+		return
+	}
+	if err := store.WithBucket(bucket).DeleteBucket(c.Request.Context()); err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error() + " (a bucket must be empty to delete)"})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"message": "Bucket deleted", "bucket": bucket})
+}
+
+// createStoreBucket creates a new bucket on the store's server.
+func (s *Server) createStoreBucket(c *gin.Context) {
+	store, _, ok := s.storeS3ByName(c)
+	if !ok {
+		return
+	}
+	var req struct {
+		Bucket string `json:"bucket" binding:"required"`
+	}
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	if err := store.WithBucket(req.Bucket).EnsureBucket(c.Request.Context()); err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusCreated, gin.H{"message": "Bucket created", "bucket": req.Bucket})
+}
+
+// listStoreObjects returns one page of objects in the target bucket, with a
+// continuation token for the next page.
+func (s *Server) listStoreObjects(c *gin.Context) {
+	store, dest, ok := s.storeS3ByName(c)
+	if !ok {
+		return
+	}
+	limit := int32(defaultObjectPage)
+	if v, err := strconv.Atoi(c.Query("limit")); err == nil && v > 0 && v <= 1000 {
+		limit = int32(v)
+	}
+
+	objects, next, err := scopedStore(store, dest, c).ListObjectsPage(c.Request.Context(), c.Query("prefix"), c.Query("token"), limit)
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	if objects == nil {
+		objects = []backup.ObjectInfo{}
+	}
+	c.JSON(http.StatusOK, gin.H{"objects": objects, "next_token": next})
+}
+
+// uploadStoreObject stores an uploaded file at the given key (defaulting to the
+// file's name) in the target bucket.
+func (s *Server) uploadStoreObject(c *gin.Context) {
+	store, dest, ok := s.storeS3ByName(c)
+	if !ok {
+		return
+	}
+	target := scopedStore(store, dest, c)
+	fileHeader, err := c.FormFile("file")
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "file is required"})
+		return
+	}
+	key := c.PostForm("key")
+	if key == "" {
+		key = fileHeader.Filename
+	}
+	f, err := fileHeader.Open()
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	defer f.Close()
+
+	if err := target.Put(c.Request.Context(), key, f, fileHeader.Size); err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusCreated, gin.H{"message": "Object uploaded", "key": key})
+}
+
+// downloadStoreObject streams an object back to the caller as an attachment.
+func (s *Server) downloadStoreObject(c *gin.Context) {
+	store, dest, ok := s.storeS3ByName(c)
+	if !ok {
+		return
+	}
+	key := c.Query("key")
+	if key == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "key is required"})
+		return
+	}
+	reader, err := scopedStore(store, dest, c).Open(c.Request.Context(), key)
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	defer reader.Close()
+
+	if c.Query("inline") == "true" {
+		c.DataFromReader(http.StatusOK, -1, "application/octet-stream", reader, nil)
+		return
+	}
+	c.Header("Content-Disposition", fmt.Sprintf("attachment; filename=%q", path.Base(key)))
+	c.DataFromReader(http.StatusOK, -1, "application/octet-stream", reader, nil)
+}
+
+// deleteStoreObject removes a single object from the target bucket. Deletes in
+// the store's backup bucket are refused, so browsing cannot destroy backups;
+// objects in any other bucket are freely removable.
+func (s *Server) deleteStoreObject(c *gin.Context) {
+	store, dest, ok := s.storeS3ByName(c)
+	if !ok {
+		return
+	}
+	bucket := c.Query("bucket")
+	if bucket == "" {
+		bucket = dest.Bucket
+	}
+	if bucket == dest.Bucket {
+		c.JSON(http.StatusForbidden, gin.H{
+			"error": fmt.Sprintf("deletes are disabled in the backup bucket %q to protect backups; use another bucket", dest.Bucket),
+		})
+		return
+	}
+	key := c.Query("key")
+	if key == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "key is required"})
+		return
+	}
+	if err := store.WithBucket(bucket).Delete(c.Request.Context(), key); err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"message": "Object deleted", "key": key})
+}
diff --git a/internal/api/object_consume.go b/internal/api/object_consume.go
new file mode 100644
index 0000000..7c542bc
--- /dev/null
+++ b/internal/api/object_consume.go
@@ -0,0 +1,127 @@
+package api
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+
+	"github.com/flatrun/agent/internal/docker"
+	"github.com/gin-gonic/gin"
+)
+
+// attachStoreToDeployment injects a store's connection details into another
+// deployment's environment so its app can use the store directly. A managed
+// store is reached inside the cluster by its container name on the shared
+// object-storage network (which the app is joined to); an external store by its
+// public URL. The deployment must be restarted for the change to take effect.
+func (s *Server) attachStoreToDeployment(c *gin.Context) {
+	dest, ok := s.findDestinationByName(c.Param("name"))
+	if !ok {
+		c.JSON(http.StatusNotFound, gin.H{"error": "object store not found"})
+		return
+	}
+
+	var req struct {
+		Deployment string `json:"deployment" binding:"required"`
+		Prefix     string `json:"prefix"`
+	}
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	prefix := req.Prefix
+	if prefix == "" {
+		prefix = "S3_"
+	}
+
+	deployDir := filepath.Join(s.config.DeploymentsPath, req.Deployment)
+	composePath := filepath.Join(deployDir, "docker-compose.yml")
+	composeContent, err := os.ReadFile(composePath)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "could not read deployment compose: " + err.Error()})
+		return
+	}
+
+	cred, err := s.credentialsManager.GetGenericCredential(dest.CredentialID)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "store has no usable credential: " + err.Error()})
+		return
+	}
+
+	endpoint := dest.Endpoint
+	joinNetwork := ""
+	if dest.Kind == "managed" && dest.Deployment != "" {
+		port := managedStorePort
+		if u, perr := url.Parse(dest.Endpoint); perr == nil && u.Port() != "" {
+			port = u.Port()
+		}
+		endpoint = fmt.Sprintf("http://%s:%s", dest.Deployment, port)
+		joinNetwork = objectStorageNetworkName(s.config)
+	}
+
+	vars := map[string]string{
+		prefix + "ENDPOINT":          endpoint,
+		prefix + "BUCKET":            dest.Bucket,
+		prefix + "REGION":            dest.Region,
+		prefix + "ACCESS_KEY_ID":     cred.Data["access_key_id"],
+		prefix + "SECRET_ACCESS_KEY": cred.Data["secret_access_key"],
+		prefix + "USE_PATH_STYLE":    strconv.FormatBool(dest.UsePathStyle),
+	}
+
+	// Upsert into the deployment's .env.flatrun, preserving anything already there.
+	var envVars []EnvVar
+	if existing, rerr := os.ReadFile(filepath.Join(deployDir, ".env.flatrun")); rerr == nil {
+		envVars = parseEnvContent(string(existing))
+	}
+	index := make(map[string]int, len(envVars))
+	for i, e := range envVars {
+		index[e.Key] = i
+	}
+	for k, v := range vars {
+		if i, ok := index[k]; ok {
+			envVars[i].Value = v
+		} else {
+			envVars = append(envVars, EnvVar{Key: k, Value: v})
+		}
+	}
+	if err := s.writeEnvFile(req.Deployment, envVars); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to write environment: " + err.Error()})
+		return
+	}
+
+	// Wire the service to load that env file, and (managed) join the store's network.
+	updated, err := docker.EnsureServiceEnvFile(string(composeContent), ".env.flatrun")
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update compose: " + err.Error()})
+		return
+	}
+	if joinNetwork != "" {
+		if s.networksManager != nil {
+			_ = s.networksManager.EnsureNetwork(joinNetwork)
+		}
+		if withNet, nerr := docker.AddNetworkToCompose(updated, joinNetwork); nerr == nil {
+			updated = withNet
+		}
+	}
+	if err := os.WriteFile(composePath, []byte(updated), 0644); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to write compose: " + err.Error()})
+		return
+	}
+
+	keys := make([]string, 0, len(vars))
+	for k := range vars {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	c.JSON(http.StatusOK, gin.H{
+		"message":  "Store attached. Restart the deployment to apply.",
+		"keys":     keys,
+		"endpoint": endpoint,
+		"network":  joinNetwork,
+	})
+}
diff --git a/internal/api/object_replicate.go b/internal/api/object_replicate.go
new file mode 100644
index 0000000..088f6d9
--- /dev/null
+++ b/internal/api/object_replicate.go
@@ -0,0 +1,128 @@
+package api
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+
+	"github.com/flatrun/agent/internal/backup"
+	"github.com/gin-gonic/gin"
+)
+
+// s3StoreByName resolves a registered store by name into an S3 client without
+// writing a response, for callers that handle their own errors.
+func (s *Server) s3StoreByName(name string) (*backup.S3Store, error) {
+	dest, ok := s.findDestinationByName(name)
+	if !ok {
+		return nil, fmt.Errorf("object store %q not found", name)
+	}
+	store, err := s.buildStore(dest)
+	if err != nil {
+		return nil, err
+	}
+	s3store, ok := store.(*backup.S3Store)
+	if !ok {
+		return nil, fmt.Errorf("store %q is not S3-compatible", name)
+	}
+	return s3store, nil
+}
+
+// copyObject streams a source object through a temp file before uploading it.
+// The S3 client must sign the request body, which requires a seekable reader;
+// an object read stream is not seekable, so it is buffered to disk first. Disk
+// (not memory) keeps large backup archives from exhausting RAM.
+func copyObject(ctx context.Context, src, dst *backup.S3Store, key string) (int64, error) {
+	reader, err := src.Open(ctx, key)
+	if err != nil {
+		return 0, err
+	}
+	defer reader.Close()
+
+	tmp, err := os.CreateTemp("", "flatrun-replicate-*")
+	if err != nil {
+		return 0, err
+	}
+	defer os.Remove(tmp.Name())
+	defer tmp.Close()
+
+	n, err := io.Copy(tmp, reader)
+	if err != nil {
+		return 0, err
+	}
+	if _, err := tmp.Seek(0, io.SeekStart); err != nil {
+		return 0, err
+	}
+	if err := dst.Put(ctx, key, tmp, n); err != nil {
+		return 0, err
+	}
+	return n, nil
+}
+
+// replicateStore copies every object from the store named in the path to a
+// target store. It is incremental: an object already present in the target at
+// the same size is skipped, so re-running only moves what changed. This backs
+// offsite copies (managed to external) and local caches (external to managed).
+func (s *Server) replicateStore(c *gin.Context) {
+	src, _, ok := s.storeS3ByName(c)
+	if !ok {
+		return
+	}
+
+	var req struct {
+		Target string `json:"target" binding:"required"`
+	}
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	if req.Target == c.Param("name") {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "a store cannot replicate to itself"})
+		return
+	}
+
+	dst, err := s.s3StoreByName(req.Target)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	ctx := c.Request.Context()
+	if err := dst.EnsureBucket(ctx); err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": "target bucket unavailable: " + err.Error()})
+		return
+	}
+
+	objects, err := src.ListObjects(ctx, "")
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": "could not list source: " + err.Error()})
+		return
+	}
+
+	var copied, skipped, failed int
+	var bytes int64
+	for _, o := range objects {
+		if info, err := dst.Stat(ctx, o.Key); err == nil && info.Size == o.Size {
+			skipped++
+			continue
+		}
+		n, err := copyObject(ctx, src, dst, o.Key)
+		if err != nil {
+			failed++
+			continue
+		}
+		copied++
+		bytes += n
+	}
+
+	c.JSON(http.StatusOK, gin.H{
+		"message":       fmt.Sprintf("Replicated %d object(s) to %s", copied, req.Target),
+		"target":        req.Target,
+		"copied":        copied,
+		"skipped":       skipped,
+		"failed":        failed,
+		"bytes_copied":  bytes,
+		"total_objects": len(objects),
+	})
+}
diff --git a/internal/api/object_store_list_test.go b/internal/api/object_store_list_test.go
new file mode 100644
index 0000000..d04d433
--- /dev/null
+++ b/internal/api/object_store_list_test.go
@@ -0,0 +1,68 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/flatrun/agent/pkg/config"
+	"github.com/gin-gonic/gin"
+)
+
+// The object store template must carry its declared S3 bootstrap contract
+// through the list API, which is what the picker forwards to auto-register it.
+func TestListTemplates_ObjectStoreCarriesContract(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	srv := &Server{config: &config.Config{DeploymentsPath: t.TempDir()}}
+
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("GET", "/templates", nil)
+	srv.listTemplates(c)
+
+	var resp struct {
+		Templates []Template `json:"templates"`
+	}
+	if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	var minio *Template
+	for i := range resp.Templates {
+		if resp.Templates[i].ID == "minio" {
+			minio = &resp.Templates[i]
+		}
+	}
+	if minio == nil {
+		t.Fatal("minio template missing from type=all listing")
+	}
+	if minio.ObjectStore == nil {
+		t.Fatal("minio template does not carry an object_store contract")
+	}
+	if minio.ObjectStore.APIPort != 9000 || minio.ObjectStore.AccessKeyEnv != "MINIO_ROOT_USER" {
+		t.Fatalf("unexpected contract: %+v", *minio.ObjectStore)
+	}
+}
+
+// A storage template is a normal, standalone template: it must appear in the
+// default catalog listing, not only under type=all.
+func TestListTemplates_ObjectStoreVisibleInDefault(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	srv := &Server{config: &config.Config{DeploymentsPath: t.TempDir()}}
+
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("GET", "/templates", nil)
+	srv.listTemplates(c)
+
+	var resp struct {
+		Templates []Template `json:"templates"`
+	}
+	_ = json.Unmarshal(w.Body.Bytes(), &resp)
+	for _, tpl := range resp.Templates {
+		if tpl.ID == "minio" {
+			return
+		}
+	}
+	t.Fatal("minio (storage category) should appear in the default listing")
+}
diff --git a/internal/api/object_stores.go b/internal/api/object_stores.go
new file mode 100644
index 0000000..93420f7
--- /dev/null
+++ b/internal/api/object_stores.go
@@ -0,0 +1,244 @@
+package api
+
+import (
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/flatrun/agent/internal/backup"
+	"github.com/flatrun/agent/pkg/config"
+	"github.com/flatrun/agent/pkg/models"
+	"github.com/gin-gonic/gin"
+)
+
+const (
+	defaultManagedBucket     = "backups"
+	defaultManagedRegion     = "us-east-1"
+	managedProvisionAttempts = 20
+	managedProvisionInterval = 500 * time.Millisecond
+)
+
+// provisionManagedObjectStoreRequest carries the S3 bootstrap contract for a
+// deployment FlatRun runs. Credentials come either literally (a custom store
+// whose secrets the caller knows) or by naming the deployment env vars that
+// hold them (a template store, where the UI forwards the template's declared
+// object_store contract). The agent stays free of any per-image knowledge.
+type provisionManagedObjectStoreRequest struct {
+	Deployment string `json:"deployment" binding:"required"`
+	StoreName  string `json:"store_name"`
+	Bucket     string `json:"bucket"`
+
+	AccessKeyEnv string `json:"access_key_env"`
+	SecretKeyEnv string `json:"secret_key_env"`
+	AccessKey    string `json:"access_key"`
+	SecretKey    string `json:"secret_key"`
+
+	APIPort      int    `json:"api_port"`
+	Region       string `json:"region"`
+	UsePathStyle *bool  `json:"use_path_style"`
+}
+
+// provisionManagedObjectStore turns a deployment FlatRun runs into a connected
+// store: it resolves the store's S3 credentials, stores a credential, ensures
+// the bucket exists, and registers a managed backup destination pointing at the
+// deployment. It is the auto-register step behind the "Deploy a local store"
+// and "use an existing deployment" flows.
+func (s *Server) provisionManagedObjectStore(c *gin.Context) {
+	if s.backupManager == nil {
+		c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Backup manager not enabled"})
+		return
+	}
+
+	var req provisionManagedObjectStoreRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	if req.APIPort <= 0 {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "api_port is required"})
+		return
+	}
+
+	accessKey, secretKey, err := s.resolveStoreCredentials(req)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	storeName := strings.TrimSpace(req.StoreName)
+	if storeName == "" {
+		storeName = req.Deployment
+	}
+	bucket := strings.TrimSpace(req.Bucket)
+	if bucket == "" {
+		bucket = defaultManagedBucket
+	}
+	region := strings.TrimSpace(req.Region)
+	if region == "" {
+		region = defaultManagedRegion
+	}
+	usePathStyle := true
+	if req.UsePathStyle != nil {
+		usePathStyle = *req.UsePathStyle
+	}
+
+	// A create returns before the store finishes starting, so wait for the
+	// container to become reachable before making the bucket on it.
+	endpoint, err := s.waitForManagedEndpoint(req.Deployment, req.APIPort)
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": fmt.Sprintf("object store %q is not reachable yet: %v", req.Deployment, err)})
+		return
+	}
+
+	cred, err := s.credentialsManager.CreateGenericCredential(storeName+"-keys", models.CredentialKindS3, map[string]string{
+		"access_key_id":     accessKey,
+		"secret_access_key": secretKey,
+	})
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	enabled := true
+	dest := config.BackupDestination{
+		Name:         storeName,
+		Type:         "s3",
+		Kind:         "managed",
+		Deployment:   req.Deployment,
+		Endpoint:     endpoint,
+		Region:       region,
+		Bucket:       bucket,
+		CredentialID: cred.ID,
+		UsePathStyle: usePathStyle,
+		Enabled:      &enabled,
+	}
+
+	if err := s.ensureManagedBucket(c, dest); err != nil {
+		_ = s.credentialsManager.DeleteGenericCredential(cred.ID)
+		c.JSON(http.StatusBadGateway, gin.H{"error": fmt.Sprintf("could not create bucket on %q: %v", storeName, err)})
+		return
+	}
+
+	updated := append(append([]config.BackupDestination{}, s.config.Backup.Destinations...), dest)
+	outcome, err := s.applyConfigUpdate("backup.destinations", updated)
+	if err != nil {
+		_ = s.credentialsManager.DeleteGenericCredential(cred.ID)
+		respondAPIError(c, err)
+		return
+	}
+
+	resp := gin.H{
+		"message":     "Managed object store connected",
+		"destination": dest,
+		"credential":  cred,
+		"applied":     outcome.Applied,
+	}
+	if outcome.ApplyErr != nil {
+		resp["apply_error"] = outcome.ApplyErr.Error()
+	}
+	c.JSON(http.StatusCreated, resp)
+}
+
+// resolveStoreCredentials returns the S3 access key and secret for a store,
+// taken from the request when supplied directly, otherwise read from the
+// deployment env vars the request names.
+func (s *Server) resolveStoreCredentials(req provisionManagedObjectStoreRequest) (string, string, error) {
+	if req.AccessKey != "" && req.SecretKey != "" {
+		return req.AccessKey, req.SecretKey, nil
+	}
+	if req.AccessKeyEnv == "" || req.SecretKeyEnv == "" {
+		return "", "", fmt.Errorf("provide S3 credentials, or the env keys that hold them")
+	}
+
+	env, err := s.readDeploymentEnvMap(req.Deployment)
+	if err != nil {
+		return "", "", err
+	}
+	access, secret := env[req.AccessKeyEnv], env[req.SecretKeyEnv]
+	if access == "" || secret == "" {
+		return "", "", fmt.Errorf("object store %q has no S3 credentials in %s / %s", req.Deployment, req.AccessKeyEnv, req.SecretKeyEnv)
+	}
+	return access, secret, nil
+}
+
+// readDeploymentEnvMap reads a deployment's generated env into a key/value map.
+// A template writes its env to whatever file it declares (MinIO uses .env,
+// others use .env.flatrun), so every env file in the deployment is read and
+// merged rather than assuming one fixed name. The FlatRun-managed .env.flatrun
+// wins where a key appears in more than one.
+func (s *Server) readDeploymentEnvMap(name string) (map[string]string, error) {
+	dir := filepath.Join(s.config.DeploymentsPath, name)
+	matches, _ := filepath.Glob(filepath.Join(dir, ".env*"))
+
+	env := make(map[string]string)
+	found := false
+	for _, path := range matches {
+		base := filepath.Base(path)
+		if strings.HasSuffix(base, ".example") || strings.HasSuffix(base, ".sample") {
+			continue
+		}
+		content, err := os.ReadFile(path)
+		if err != nil {
+			continue
+		}
+		found = true
+		for _, v := range parseEnvContent(string(content)) {
+			if base == ".env.flatrun" || env[v.Key] == "" {
+				env[v.Key] = v.Value
+			}
+		}
+	}
+	if !found {
+		return nil, fmt.Errorf("could not read object store environment in %s", dir)
+	}
+	return env, nil
+}
+
+// waitForManagedEndpoint polls for the deployment's container to become
+// reachable and returns the address its S3 API is served on.
+func (s *Server) waitForManagedEndpoint(deployment string, apiPort int) (string, error) {
+	if s.manager == nil {
+		return "", fmt.Errorf("docker manager unavailable")
+	}
+	var lastErr error
+	for i := 0; i < managedProvisionAttempts; i++ {
+		ip, err := s.manager.ContainerPrimaryIP(deployment, objectStorageNetworkName(s.config))
+		if err == nil && ip != "" {
+			return fmt.Sprintf("http://%s:%d", ip, apiPort), nil
+		}
+		lastErr = err
+		time.Sleep(managedProvisionInterval)
+	}
+	if lastErr == nil {
+		lastErr = fmt.Errorf("container did not report an address")
+	}
+	return "", lastErr
+}
+
+// ensureManagedBucket makes the destination's bucket if it is missing. The
+// container reports an address a moment before its S3 API is actually serving,
+// so the bucket call is retried over a short warmup window.
+func (s *Server) ensureManagedBucket(c *gin.Context, dest config.BackupDestination) error {
+	store, err := s.buildStore(dest)
+	if err != nil {
+		return err
+	}
+	s3store, ok := store.(*backup.S3Store)
+	if !ok {
+		return nil
+	}
+
+	var lastErr error
+	for i := 0; i < managedProvisionAttempts; i++ {
+		if err := s3store.EnsureBucket(c.Request.Context()); err == nil {
+			return nil
+		} else {
+			lastErr = err
+		}
+		time.Sleep(managedProvisionInterval)
+	}
+	return lastErr
+}
diff --git a/internal/api/object_stores_test.go b/internal/api/object_stores_test.go
index 19b9906..8f729b7 100644
--- a/internal/api/object_stores_test.go
+++ b/internal/api/object_stores_test.go
@@ -6,6 +6,8 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/flatrun/agent/internal/auth"
@@ -65,6 +67,12 @@ func setupObjectStoreTestServer(t *testing.T) (*Server, *gin.Engine, func()) {
 	protected.PUT("/storage-credentials/:id", mw.RequirePermission(auth.PermBackupsWrite), server.updateStorageCredential)
 	protected.DELETE("/storage-credentials/:id", mw.RequirePermission(auth.PermBackupsDelete), server.deleteStorageCredential)
 	protected.GET("/backup-destinations", mw.RequirePermission(auth.PermBackupsRead), server.listBackupDestinations)
+	protected.POST("/object-stores/provision-managed", mw.RequirePermission(auth.PermBackupsWrite), server.provisionManagedObjectStore)
+	protected.GET("/object-stores/:name/objects", mw.RequirePermission(auth.PermBackupsRead), server.listStoreObjects)
+	protected.POST("/object-stores/:name/objects", mw.RequirePermission(auth.PermBackupsWrite), server.uploadStoreObject)
+	protected.GET("/object-stores/:name/objects/download", mw.RequirePermission(auth.PermBackupsRead), server.downloadStoreObject)
+	protected.DELETE("/object-stores/:name/objects", mw.RequirePermission(auth.PermBackupsWrite), server.deleteStoreObject)
+	protected.POST("/object-stores/:name/attach", mw.RequirePermission(auth.PermDeploymentsWrite), server.attachStoreToDeployment)
 
 	cleanup := func() {
 		authManager.Close()
@@ -178,6 +186,161 @@ func TestStorageCredential_DeleteInUseConflicts(t *testing.T) {
 	}
 }
 
+func TestProvisionManagedObjectStore_RequiresApiPort(t *testing.T) {
+	_, router, cleanup := setupObjectStoreTestServer(t)
+	defer cleanup()
+	token := objStoreLogin(t, router)
+
+	res := osReq(t, router, http.MethodPost, "/api/object-stores/provision-managed", token, map[string]any{
+		"deployment": "some-store",
+	})
+	if res.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400 without an api_port, got %d %s", res.Code, res.Body.String())
+	}
+}
+
+func TestProvisionManagedObjectStore_MissingEnvReturns400(t *testing.T) {
+	_, router, cleanup := setupObjectStoreTestServer(t)
+	defer cleanup()
+	token := objStoreLogin(t, router)
+
+	res := osReq(t, router, http.MethodPost, "/api/object-stores/provision-managed", token, map[string]any{
+		"deployment":     "no-such-store",
+		"access_key_env": "MINIO_ROOT_USER",
+		"secret_key_env": "MINIO_ROOT_PASSWORD",
+		"api_port":       9000,
+	})
+	if res.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400 for a deployment with no environment, got %d %s", res.Code, res.Body.String())
+	}
+}
+
+// Credentials must be read from whatever env file the template wrote. The MinIO
+// template writes .env (not .env.flatrun), so a deployment with only a .env must
+// still resolve credentials (getting past resolution to the reachability step,
+// 502, rather than failing at 400 for missing env).
+func TestProvisionManagedObjectStore_ReadsPlainEnvFile(t *testing.T) {
+	server, router, cleanup := setupObjectStoreTestServer(t)
+	defer cleanup()
+	token := objStoreLogin(t, router)
+
+	deployment := "minio-store"
+	deployDir := filepath.Join(server.config.DeploymentsPath, deployment)
+	if err := os.MkdirAll(deployDir, 0o755); err != nil {
+		t.Fatalf("mkdir deployment: %v", err)
+	}
+	env := "MINIO_ROOT_USER=flatrun\nMINIO_ROOT_PASSWORD=generated-secret\n"
+	if err := os.WriteFile(filepath.Join(deployDir, ".env"), []byte(env), 0o600); err != nil {
+		t.Fatalf("write env: %v", err)
+	}
+
+	res := osReq(t, router, http.MethodPost, "/api/object-stores/provision-managed", token, map[string]any{
+		"deployment":     deployment,
+		"access_key_env": "MINIO_ROOT_USER",
+		"secret_key_env": "MINIO_ROOT_PASSWORD",
+		"api_port":       9000,
+	})
+	if res.Code != http.StatusBadGateway {
+		t.Fatalf("expected 502 (creds resolved from .env, store unreachable), got %d %s", res.Code, res.Body.String())
+	}
+}
+
+// A store whose credentials resolve but that is not reachable (no docker in the
+// unit environment) must fail before any credential or destination is created,
+// so a failed provision leaves no orphaned state behind. This holds for the
+// template path (credentials read from named env vars).
+func TestProvisionManagedObjectStore_UnreachableLeavesNoState(t *testing.T) {
+	server, router, cleanup := setupObjectStoreTestServer(t)
+	defer cleanup()
+	token := objStoreLogin(t, router)
+
+	deployment := "my-store"
+	deployDir := filepath.Join(server.config.DeploymentsPath, deployment)
+	if err := os.MkdirAll(deployDir, 0o755); err != nil {
+		t.Fatalf("mkdir deployment: %v", err)
+	}
+	env := "MINIO_ROOT_USER=flatrun\nMINIO_ROOT_PASSWORD=generated-secret\n"
+	if err := os.WriteFile(filepath.Join(deployDir, ".env.flatrun"), []byte(env), 0o600); err != nil {
+		t.Fatalf("write env: %v", err)
+	}
+
+	res := osReq(t, router, http.MethodPost, "/api/object-stores/provision-managed", token, map[string]any{
+		"deployment":     deployment,
+		"access_key_env": "MINIO_ROOT_USER",
+		"secret_key_env": "MINIO_ROOT_PASSWORD",
+		"api_port":       9000,
+	})
+	if res.Code != http.StatusBadGateway {
+		t.Fatalf("expected 502 when the store is unreachable, got %d %s", res.Code, res.Body.String())
+	}
+	if creds := server.credentialsManager.ListGenericCredentials(models.CredentialKindS3); len(creds) != 0 {
+		t.Fatalf("failed provision left a credential behind: %#v", creds)
+	}
+	if len(server.config.Backup.Destinations) != 0 {
+		t.Fatalf("failed provision left a destination behind: %#v", server.config.Backup.Destinations)
+	}
+}
+
+func TestListStoreObjects_UnknownStore404(t *testing.T) {
+	_, router, cleanup := setupObjectStoreTestServer(t)
+	defer cleanup()
+	token := objStoreLogin(t, router)
+
+	res := osReq(t, router, http.MethodGet, "/api/object-stores/nope/objects", token, nil)
+	if res.Code != http.StatusNotFound {
+		t.Fatalf("expected 404 for an unknown store, got %d %s", res.Code, res.Body.String())
+	}
+}
+
+func TestAttachStoreToDeployment_WritesEnvAndWiresCompose(t *testing.T) {
+	server, router, cleanup := setupObjectStoreTestServer(t)
+	defer cleanup()
+	token := objStoreLogin(t, router)
+
+	cred, err := server.credentialsManager.CreateGenericCredential("app-keys", models.CredentialKindS3, map[string]string{
+		"access_key_id": "AKIATEST", "secret_access_key": "shh",
+	})
+	if err != nil {
+		t.Fatalf("create credential: %v", err)
+	}
+	server.config.Backup.Destinations = []config.BackupDestination{{
+		Name: "r2", Type: "s3", Kind: "external", Endpoint: "https://s3.example.com",
+		Region: "us-east-1", Bucket: "assets", CredentialID: cred.ID, UsePathStyle: true,
+	}}
+
+	app := "webapp"
+	appDir := filepath.Join(server.config.DeploymentsPath, app)
+	if err := os.MkdirAll(appDir, 0o755); err != nil {
+		t.Fatalf("mkdir: %v", err)
+	}
+	compose := "services:\n  app:\n    image: node:20-alpine\n"
+	if err := os.WriteFile(filepath.Join(appDir, "docker-compose.yml"), []byte(compose), 0o644); err != nil {
+		t.Fatalf("write compose: %v", err)
+	}
+
+	res := osReq(t, router, http.MethodPost, "/api/object-stores/r2/attach", token, map[string]any{
+		"deployment": app,
+	})
+	if res.Code != http.StatusOK {
+		t.Fatalf("attach: %d %s", res.Code, res.Body.String())
+	}
+
+	env, err := os.ReadFile(filepath.Join(appDir, ".env.flatrun"))
+	if err != nil {
+		t.Fatalf("read env: %v", err)
+	}
+	for _, want := range []string{"S3_ENDPOINT=https://s3.example.com", "S3_BUCKET=assets", "S3_ACCESS_KEY_ID=AKIATEST", "S3_SECRET_ACCESS_KEY=shh"} {
+		if !strings.Contains(string(env), want) {
+			t.Fatalf("env missing %q:\n%s", want, env)
+		}
+	}
+
+	updated, _ := os.ReadFile(filepath.Join(appDir, "docker-compose.yml"))
+	if !strings.Contains(string(updated), "env_file") || !strings.Contains(string(updated), ".env.flatrun") {
+		t.Fatalf("compose not wired to env file:\n%s", updated)
+	}
+}
+
 func TestBackupDestinations_ListReportsKind(t *testing.T) {
 	server, router, cleanup := setupObjectStoreTestServer(t)
 	defer cleanup()
diff --git a/internal/api/server.go b/internal/api/server.go
index a24f6ff..7287893 100644
--- a/internal/api/server.go
+++ b/internal/api/server.go
@@ -118,6 +118,17 @@ type Server struct {
 	statsAt    time.Time
 }
 
+// objectStorageNetworkName resolves the shared network self-hosted object
+// stores join. It reuses the database network unless a dedicated one is
+// configured, so object storage rides the same data-backend network as
+// databases by default.
+func objectStorageNetworkName(cfg *config.Config) string {
+	if n := cfg.Infrastructure.DefaultObjectStorageNetwork; n != "" {
+		return n
+	}
+	return cfg.Infrastructure.DefaultDatabaseNetwork
+}
+
 func New(cfg *config.Config, configPath string) *Server {
 	if cfg.Logging.Level == "debug" {
 		gin.SetMode(gin.DebugMode)
@@ -125,6 +136,10 @@ func New(cfg *config.Config, configPath string) *Server {
 		gin.SetMode(gin.ReleaseMode)
 	}
 
+	// Exposed to compose substitution so an object-store template joins the
+	// configured network (${FLATRUN_OBJECT_NETWORK:-database}) on every up.
+	os.Setenv("FLATRUN_OBJECT_NETWORK", objectStorageNetworkName(cfg))
+
 	router := gin.Default()
 
 	if cfg.API.EnableCORS {
@@ -733,6 +748,18 @@ func (s *Server) setupRoutes() {
 			protected.GET("/backup-destinations", s.authMiddleware.RequirePermission(auth.PermBackupsRead), s.listBackupDestinations)
 			protected.POST("/backup-destinations/test", s.authMiddleware.RequirePermission(auth.PermBackupsWrite), s.testBackupDestination)
 
+			// Object stores
+			protected.POST("/object-stores/provision-managed", s.authMiddleware.RequirePermission(auth.PermBackupsWrite), s.provisionManagedObjectStore)
+			protected.GET("/object-stores/:name/buckets", s.authMiddleware.RequirePermission(auth.PermBackupsRead), s.listStoreBuckets)
+			protected.POST("/object-stores/:name/buckets", s.authMiddleware.RequirePermission(auth.PermBackupsWrite), s.createStoreBucket)
+			protected.DELETE("/object-stores/:name/buckets/:bucket", s.authMiddleware.RequirePermission(auth.PermBackupsDelete), s.deleteStoreBucket)
+			protected.GET("/object-stores/:name/objects", s.authMiddleware.RequirePermission(auth.PermBackupsRead), s.listStoreObjects)
+			protected.POST("/object-stores/:name/objects", s.authMiddleware.RequirePermission(auth.PermBackupsWrite), s.uploadStoreObject)
+			protected.GET("/object-stores/:name/objects/download", s.authMiddleware.RequirePermission(auth.PermBackupsRead), s.downloadStoreObject)
+			protected.DELETE("/object-stores/:name/objects", s.authMiddleware.RequirePermission(auth.PermBackupsWrite), s.deleteStoreObject)
+			protected.POST("/object-stores/:name/attach", s.authMiddleware.RequirePermission(auth.PermDeploymentsWrite), s.attachStoreToDeployment)
+			protected.POST("/object-stores/:name/replicate", s.authMiddleware.RequirePermission(auth.PermBackupsWrite), s.replicateStore)
+
 			// Scheduler endpoints
 			protected.GET("/scheduler/tasks", s.authMiddleware.RequirePermission(auth.PermSchedulerRead), s.listScheduledTasks)
 			protected.GET("/scheduler/tasks/:id", s.authMiddleware.RequirePermission(auth.PermSchedulerRead), s.getScheduledTask)
@@ -1197,6 +1224,19 @@ func (s *Server) createDeployment(c *gin.Context) {
 		s.processTemplateFiles(req.Name, req.TemplateID, allEnvVars)
 		s.processTemplateEnv(req.Name, req.TemplateID, req.ComposeContent, allEnvVars)
 		s.applyTemplateMountOwnership(req.Name, req.TemplateID)
+
+		// An object store joins the shared object-storage network; ensure it
+		// exists so the container can start (it is declared external).
+		if md, err := s.loadTemplateMetadata(req.TemplateID); err == nil && md.ObjectStore != nil {
+			if objNet := objectStorageNetworkName(s.config); objNet != "" && s.networksManager != nil {
+				if err := s.networksManager.EnsureNetwork(objNet); err != nil {
+					c.JSON(http.StatusInternalServerError, gin.H{
+						"error": fmt.Sprintf("Failed to ensure object storage network %q exists: %v", objNet, err),
+					})
+					return
+				}
+			}
+		}
 	}
 
 	// Seeding reads the images, so it runs before the deployment is started and
@@ -3341,17 +3381,30 @@ type TemplateFile struct {
 }
 
 type TemplateMetadata struct {
-	Name          string          `yaml:"name"`
-	Description   string          `yaml:"description"`
-	Icon          string          `yaml:"icon"`
-	Logo          string          `yaml:"logo"`
-	Category      string          `yaml:"category"`
-	Type          string          `yaml:"type"`
-	Priority      int             `yaml:"priority"`
-	ContainerPort int             `yaml:"container_port"`
-	Mounts        []TemplateMount `yaml:"mounts"`
-	Files         []TemplateFile  `yaml:"files"`
-	Env           TemplateEnv     `yaml:"env"`
+	Name          string               `yaml:"name"`
+	Description   string               `yaml:"description"`
+	Icon          string               `yaml:"icon"`
+	Logo          string               `yaml:"logo"`
+	Category      string               `yaml:"category"`
+	Type          string               `yaml:"type"`
+	ObjectStore   *TemplateObjectStore `yaml:"object_store"`
+	Priority      int                  `yaml:"priority"`
+	ContainerPort int                  `yaml:"container_port"`
+	Mounts        []TemplateMount      `yaml:"mounts"`
+	Files         []TemplateFile       `yaml:"files"`
+	Env           TemplateEnv          `yaml:"env"`
+}
+
+// TemplateObjectStore is a template's declaration that it runs an S3-compatible
+// object store, and how to bootstrap it: which generated env vars carry the
+// root credentials and which port serves the S3 API. Declaring this keeps the
+// agent free of any per-image (MinIO, Garage, ...) knowledge.
+type TemplateObjectStore struct {
+	AccessKeyEnv string `json:"access_key_env" yaml:"access_key_env"`
+	SecretKeyEnv string `json:"secret_key_env" yaml:"secret_key_env"`
+	APIPort      int    `json:"api_port" yaml:"api_port"`
+	Region       string `json:"region,omitempty" yaml:"region,omitempty"`
+	UsePathStyle bool   `json:"use_path_style" yaml:"use_path_style"`
 }
 
 // TemplateEnv describes how a platform's environment file is produced. The
@@ -3398,17 +3451,18 @@ func templateMountHostPath(m TemplateMount) string {
 }
 
 type Template struct {
-	ID            string          `json:"id"`
-	Name          string          `json:"name" yaml:"name"`
-	Description   string          `json:"description" yaml:"description"`
-	Icon          string          `json:"icon" yaml:"icon"`
-	Logo          string          `json:"logo" yaml:"logo"`
-	Category      string          `json:"category" yaml:"category"`
-	Priority      int             `json:"priority" yaml:"priority"`
-	ContainerPort int             `json:"container_port" yaml:"container_port"`
-	Mounts        []TemplateMount `json:"mounts" yaml:"mounts"`
-	Files         []TemplateFile  `json:"files" yaml:"files"`
-	Content       string          `json:"content"`
+	ID            string               `json:"id"`
+	Name          string               `json:"name" yaml:"name"`
+	Description   string               `json:"description" yaml:"description"`
+	Icon          string               `json:"icon" yaml:"icon"`
+	Logo          string               `json:"logo" yaml:"logo"`
+	Category      string               `json:"category" yaml:"category"`
+	ObjectStore   *TemplateObjectStore `json:"object_store,omitempty" yaml:"object_store,omitempty"`
+	Priority      int                  `json:"priority" yaml:"priority"`
+	ContainerPort int                  `json:"container_port" yaml:"container_port"`
+	Mounts        []TemplateMount      `json:"mounts" yaml:"mounts"`
+	Files         []TemplateFile       `json:"files" yaml:"files"`
+	Content       string               `json:"content"`
 }
 
 func (s *Server) listTemplates(c *gin.Context) {
@@ -3491,6 +3545,7 @@ func (s *Server) listTemplates(c *gin.Context) {
 				Icon:          metadata.Icon,
 				Logo:          metadata.Logo,
 				Category:      metadata.Category,
+				ObjectStore:   metadata.ObjectStore,
 				Priority:      metadata.Priority,
 				ContainerPort: metadata.ContainerPort,
 				Mounts:        metadata.Mounts,
diff --git a/internal/backup/exclude_test.go b/internal/backup/exclude_test.go
new file mode 100644
index 0000000..d1c9130
--- /dev/null
+++ b/internal/backup/exclude_test.go
@@ -0,0 +1,42 @@
+package backup
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestMatchesExclude(t *testing.T) {
+	if !matchesExclude("data", []string{"data"}) {
+		t.Fatal("exact match failed")
+	}
+	if !matchesExclude("data", []string{"da*"}) {
+		t.Fatal("glob match failed")
+	}
+	if matchesExclude("uploads", []string{"data"}) {
+		t.Fatal("should not match")
+	}
+}
+
+func TestBackupMountedData_SkipsExcluded(t *testing.T) {
+	m, err := NewManager(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	dep := t.TempDir()
+	for _, d := range []string{"data", "uploads"} {
+		os.MkdirAll(filepath.Join(dep, d), 0o755)
+		os.WriteFile(filepath.Join(dep, d, "f"), []byte("x"), 0o644)
+	}
+	out := t.TempDir()
+	var md BackupMetadata
+	if err := m.backupMountedData(dep, out, &md, []string{"data"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(filepath.Join(out, "data", "data")); err == nil {
+		t.Fatal("excluded 'data' dir should not be copied")
+	}
+	if _, err := os.Stat(filepath.Join(out, "data", "uploads")); err != nil {
+		t.Fatal("'uploads' should be copied")
+	}
+}
diff --git a/internal/backup/manager.go b/internal/backup/manager.go
index 72c7980..4d14c29 100644
--- a/internal/backup/manager.go
+++ b/internal/backup/manager.go
@@ -100,7 +100,11 @@ func (m *Manager) CreateBackup(ctx context.Context, deploymentName string, spec
 		backup.Components = append(backup.Components, "metadata")
 	}
 
-	if err := m.backupMountedData(deploymentPath, tempDir, &metadata); err != nil {
+	var excludes []string
+	if spec != nil {
+		excludes = spec.ExcludePatterns
+	}
+	if err := m.backupMountedData(deploymentPath, tempDir, &metadata, excludes); err != nil {
 		log.Printf("Backup: mounted data warning: %v", err)
 	}
 	if len(metadata.Components.MountedData) > 0 {
@@ -217,7 +221,7 @@ func (m *Manager) backupMetadataFile(deploymentPath, tempDir string, metadata *B
 	return nil
 }
 
-func (m *Manager) backupMountedData(deploymentPath, tempDir string, metadata *BackupMetadata) error {
+func (m *Manager) backupMountedData(deploymentPath, tempDir string, metadata *BackupMetadata, excludes []string) error {
 	dataDir := filepath.Join(tempDir, "data")
 	if err := os.MkdirAll(dataDir, 0755); err != nil {
 		return fmt.Errorf("failed to create data backup directory: %w", err)
@@ -225,6 +229,10 @@ func (m *Manager) backupMountedData(deploymentPath, tempDir string, metadata *Ba
 
 	commonDataDirs := []string{"data", "uploads", "storage", "config", "logs"}
 	for _, dir := range commonDataDirs {
+		if matchesExclude(dir, excludes) {
+			log.Printf("Backup: skipping %s (excluded, e.g. a database's live files captured by its dump)", dir)
+			continue
+		}
 		srcPath := filepath.Join(deploymentPath, dir)
 		if info, err := os.Stat(srcPath); err == nil && info.IsDir() {
 			destPath := filepath.Join(dataDir, dir)
@@ -239,6 +247,20 @@ func (m *Manager) backupMountedData(deploymentPath, tempDir string, metadata *Ba
 	return nil
 }
 
+// matchesExclude reports whether a mounted-data directory name matches any
+// exclude pattern (glob or exact).
+func matchesExclude(name string, patterns []string) bool {
+	for _, p := range patterns {
+		if p == name {
+			return true
+		}
+		if ok, err := filepath.Match(p, name); err == nil && ok {
+			return true
+		}
+	}
+	return false
+}
+
 func (m *Manager) backupContainerData(ctx context.Context, deploymentName string, paths []ContainerPath, tempDir string, metadata *BackupMetadata) error {
 	containerDir := filepath.Join(tempDir, "container_data")
 	if err := os.MkdirAll(containerDir, 0755); err != nil {
@@ -303,12 +325,18 @@ func (m *Manager) backupDatabases(ctx context.Context, deploymentName string, da
 }
 
 func (m *Manager) dumpMySQL(ctx context.Context, deploymentName string, db *DatabaseSpec, dbDir string) (string, error) {
-	containerName := fmt.Sprintf("%s-%s", deploymentName, db.Service)
-	if db.Service == deploymentName || db.Service == "" {
-		containerName = deploymentName
+	containerName := db.Container
+	if containerName == "" {
+		containerName = fmt.Sprintf("%s-%s", deploymentName, db.Service)
+		if db.Service == deploymentName || db.Service == "" {
+			containerName = deploymentName
+		}
 	}
 
-	dumpFile := filepath.Join(dbDir, fmt.Sprintf("%s_mysql.sql", db.Service))
+	label := db.Service
+	if label == "" {
+		label = deploymentName
+	}
 
 	host := db.Host
 	if host == "" {
@@ -324,15 +352,19 @@ func (m *Manager) dumpMySQL(ctx context.Context, deploymentName string, db *Data
 	}
 
 	args := []string{"exec"}
-
 	if db.Password != "" {
 		args = append(args, "-e", "MYSQL_PWD="+db.Password)
 	}
+	args = append(args, containerName, "mysqldump", "-h", host, "-u", user,
+		"--single-transaction", "--routines", "--triggers")
 
-	args = append(args, containerName, "mysqldump",
-		"-h", host,
-		"-u", user,
-		"--single-transaction", "--routines", "--triggers", database)
+	dumpFile := filepath.Join(dbDir, fmt.Sprintf("%s_mysql.sql", label))
+	if db.AllDatabases {
+		args = append(args, "--all-databases")
+		dumpFile = filepath.Join(dbDir, fmt.Sprintf("%s_mysql_all.sql", label))
+	} else {
+		args = append(args, database)
+	}
 
 	cmd := exec.CommandContext(ctx, "docker", args...)
 	output, err := cmd.Output()
@@ -348,12 +380,18 @@ func (m *Manager) dumpMySQL(ctx context.Context, deploymentName string, db *Data
 }
 
 func (m *Manager) dumpPostgres(ctx context.Context, deploymentName string, db *DatabaseSpec, dbDir string) (string, error) {
-	containerName := fmt.Sprintf("%s-%s", deploymentName, db.Service)
-	if db.Service == deploymentName || db.Service == "" {
-		containerName = deploymentName
+	containerName := db.Container
+	if containerName == "" {
+		containerName = fmt.Sprintf("%s-%s", deploymentName, db.Service)
+		if db.Service == deploymentName || db.Service == "" {
+			containerName = deploymentName
+		}
 	}
 
-	dumpFile := filepath.Join(dbDir, fmt.Sprintf("%s_postgres.sql", db.Service))
+	label := db.Service
+	if label == "" {
+		label = deploymentName
+	}
 
 	user := db.User
 	if user == "" {
@@ -364,15 +402,18 @@ func (m *Manager) dumpPostgres(ctx context.Context, deploymentName string, db *D
 		database = deploymentName
 	}
 
-	args := []string{
-		"exec",
-	}
-
+	args := []string{"exec"}
 	if db.Password != "" {
 		args = append(args, "-e", fmt.Sprintf("PGPASSWORD=%s", db.Password))
 	}
 
-	args = append(args, containerName, "pg_dump", "-U", user, database)
+	dumpFile := filepath.Join(dbDir, fmt.Sprintf("%s_postgres.sql", label))
+	if db.AllDatabases {
+		args = append(args, containerName, "pg_dumpall", "-U", user)
+		dumpFile = filepath.Join(dbDir, fmt.Sprintf("%s_postgres_all.sql", label))
+	} else {
+		args = append(args, containerName, "pg_dump", "-U", user, database)
+	}
 
 	cmd := exec.CommandContext(ctx, "docker", args...)
 	output, err := cmd.Output()
diff --git a/internal/backup/store_s3.go b/internal/backup/store_s3.go
index 046c20c..ac2cbf7 100644
--- a/internal/backup/store_s3.go
+++ b/internal/backup/store_s3.go
@@ -63,6 +63,27 @@ func NewS3Store(cfg S3Config) (*S3Store, error) {
 
 func (s *S3Store) Name() string { return s.cfg.Name }
 
+// EnsureBucket creates the store's bucket when it does not already exist. A
+// freshly deployed object store starts empty, so a managed store needs its
+// bucket made before the first backup can be mirrored to it.
+func (s *S3Store) EnsureBucket(ctx context.Context) error {
+	_, err := s.client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(s.cfg.Bucket)})
+	if err == nil {
+		return nil
+	}
+
+	_, err = s.client.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(s.cfg.Bucket)})
+	if err != nil {
+		var owned *s3types.BucketAlreadyOwnedByYou
+		var exists *s3types.BucketAlreadyExists
+		if errors.As(err, &owned) || errors.As(err, &exists) {
+			return nil
+		}
+		return fmt.Errorf("create bucket %q: %w", s.cfg.Bucket, err)
+	}
+	return nil
+}
+
 // fullKey prepends the destination prefix to a relative backup key.
 func (s *S3Store) fullKey(key string) string {
 	if s.cfg.Prefix == "" {
@@ -133,6 +154,121 @@ func (s *S3Store) List(ctx context.Context, prefix string) ([]ObjectInfo, error)
 	return out, nil
 }
 
+// ListBuckets returns the names of every bucket reachable with the store's
+// credentials. A store is configured with one bucket, but the server it points
+// at (a self-hosted MinIO, say) can host many.
+func (s *S3Store) ListBuckets(ctx context.Context) ([]string, error) {
+	out, err := s.client.ListBuckets(ctx, &s3.ListBucketsInput{})
+	if err != nil {
+		return nil, fmt.Errorf("s3 list buckets: %w", err)
+	}
+	names := make([]string, 0, len(out.Buckets))
+	for _, b := range out.Buckets {
+		names = append(names, aws.ToString(b.Name))
+	}
+	return names, nil
+}
+
+// WithBucket returns a view of the store scoped to another bucket at its root
+// (no prefix), so the object browser can work across buckets on the same server.
+func (s *S3Store) WithBucket(bucket string) *S3Store {
+	cfg := s.cfg
+	cfg.Bucket = bucket
+	cfg.Prefix = ""
+	return &S3Store{client: s.client, cfg: cfg}
+}
+
+// Bucket returns the bucket this store is scoped to.
+func (s *S3Store) Bucket() string { return s.cfg.Bucket }
+
+// ListObjectsPage returns one page of objects and a continuation token for the
+// next page (empty when there are no more), so a browser can page through a
+// bucket of any size instead of loading it whole.
+func (s *S3Store) ListObjectsPage(ctx context.Context, prefix, token string, limit int32) ([]ObjectInfo, string, error) {
+	in := &s3.ListObjectsV2Input{
+		Bucket: aws.String(s.cfg.Bucket),
+		Prefix: aws.String(s.fullKey(prefix)),
+	}
+	if limit > 0 {
+		in.MaxKeys = aws.Int32(limit)
+	}
+	if token != "" {
+		in.ContinuationToken = aws.String(token)
+	}
+
+	out, err := s.client.ListObjectsV2(ctx, in)
+	if err != nil {
+		return nil, "", fmt.Errorf("s3 list page: %w", err)
+	}
+	objects := make([]ObjectInfo, 0, len(out.Contents))
+	for _, o := range out.Contents {
+		objects = append(objects, ObjectInfo{
+			Key:     s.relKey(aws.ToString(o.Key)),
+			Size:    aws.ToInt64(o.Size),
+			ModTime: aws.ToTime(o.LastModified),
+		})
+	}
+	next := ""
+	if aws.ToBool(out.IsTruncated) {
+		next = aws.ToString(out.NextContinuationToken)
+	}
+	return objects, next, nil
+}
+
+// BucketStats returns the object count and total size of the store's bucket.
+// Counting stops at limit (when > 0), reporting truncated=true, so the bucket
+// list stays responsive on very large buckets.
+func (s *S3Store) BucketStats(ctx context.Context, limit int) (count int, size int64, truncated bool, err error) {
+	paginator := s3.NewListObjectsV2Paginator(s.client, &s3.ListObjectsV2Input{Bucket: aws.String(s.cfg.Bucket)})
+	for paginator.HasMorePages() {
+		page, perr := paginator.NextPage(ctx)
+		if perr != nil {
+			return count, size, truncated, fmt.Errorf("s3 stats: %w", perr)
+		}
+		for _, o := range page.Contents {
+			count++
+			size += aws.ToInt64(o.Size)
+			if limit > 0 && count >= limit {
+				return count, size, true, nil
+			}
+		}
+	}
+	return count, size, false, nil
+}
+
+// DeleteBucket removes the store's bucket. S3 requires it to be empty.
+func (s *S3Store) DeleteBucket(ctx context.Context) error {
+	if _, err := s.client.DeleteBucket(ctx, &s3.DeleteBucketInput{Bucket: aws.String(s.cfg.Bucket)}); err != nil {
+		return fmt.Errorf("s3 delete bucket %q: %w", s.cfg.Bucket, err)
+	}
+	return nil
+}
+
+// ListObjects returns every object under prefix, unlike List which is scoped to
+// backup archives (.tar.gz). It backs the object browser, where a store's full
+// contents are shown.
+func (s *S3Store) ListObjects(ctx context.Context, prefix string) ([]ObjectInfo, error) {
+	var out []ObjectInfo
+	paginator := s3.NewListObjectsV2Paginator(s.client, &s3.ListObjectsV2Input{
+		Bucket: aws.String(s.cfg.Bucket),
+		Prefix: aws.String(s.fullKey(prefix)),
+	})
+	for paginator.HasMorePages() {
+		page, err := paginator.NextPage(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("s3 list %s: %w", prefix, err)
+		}
+		for _, obj := range page.Contents {
+			out = append(out, ObjectInfo{
+				Key:     s.relKey(aws.ToString(obj.Key)),
+				Size:    aws.ToInt64(obj.Size),
+				ModTime: aws.ToTime(obj.LastModified),
+			})
+		}
+	}
+	return out, nil
+}
+
 func (s *S3Store) Delete(ctx context.Context, key string) error {
 	_, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{
 		Bucket: aws.String(s.cfg.Bucket),
diff --git a/internal/docker/api.go b/internal/docker/api.go
index d7b17bb..d28cd6e 100644
--- a/internal/docker/api.go
+++ b/internal/docker/api.go
@@ -58,6 +58,42 @@ func (a *APIClient) FindContainer(ctx context.Context, project, service string)
 	return containers[0].ID, nil
 }
 
+// ContainerPrimaryIP returns the IP address of a project's first running
+// container on the given docker network. The agent runs on the host, so a
+// service that only exposes ports on an internal compose network (a self-hosted
+// object store, say) is reached by dialing the container's address directly.
+// When network is empty, the first attached network with an address is used.
+func (a *APIClient) ContainerPrimaryIP(ctx context.Context, project, network string) (string, error) {
+	f := filters.NewArgs(
+		filters.Arg("label", fmt.Sprintf("%s=%s", composeProjectLabel, project)),
+		filters.Arg("status", "running"),
+	)
+
+	containers, err := a.cli.ContainerList(ctx, container.ListOptions{Filters: f})
+	if err != nil {
+		return "", fmt.Errorf("failed to list containers: %w", err)
+	}
+	if len(containers) == 0 {
+		return "", fmt.Errorf("no running container found for project %q", project)
+	}
+
+	ns := containers[0].NetworkSettings
+	if ns == nil {
+		return "", fmt.Errorf("container for project %q has no network settings", project)
+	}
+	if network != "" {
+		if n, ok := ns.Networks[network]; ok && n.IPAddress != "" {
+			return n.IPAddress, nil
+		}
+	}
+	for _, n := range ns.Networks {
+		if n.IPAddress != "" {
+			return n.IPAddress, nil
+		}
+	}
+	return "", fmt.Errorf("container for project %q has no network address yet", project)
+}
+
 func (a *APIClient) ExecInContainer(ctx context.Context, containerID string, command string) (string, error) {
 	execConfig := container.ExecOptions{
 		Cmd:          []string{"sh", "-c", command},
diff --git a/internal/docker/compose.go b/internal/docker/compose.go
index 0aa24de..f52cdcb 100644
--- a/internal/docker/compose.go
+++ b/internal/docker/compose.go
@@ -420,8 +420,16 @@ func (c *ComposeExecutor) runCompose(deploymentPath string, opts []RunOption, ar
 	for _, opt := range opts {
 		opt(&ro)
 	}
+
+	// Expose the agent's own uid/gid to compose substitution so a template can
+	// run its container as the user that owns the deployment directory, keeping
+	// bind-mounted data host-manageable (deletable) instead of root-owned.
+	cmd.Env = append(os.Environ(),
+		fmt.Sprintf("FLATRUN_UID=%d", os.Getuid()),
+		fmt.Sprintf("FLATRUN_GID=%d", os.Getgid()),
+	)
 	if len(ro.extraEnv) > 0 {
-		cmd.Env = append(os.Environ(), ro.extraEnv...)
+		cmd.Env = append(cmd.Env, ro.extraEnv...)
 	}
 
 	if ro.lineSink != nil {
diff --git a/internal/docker/compose_yaml.go b/internal/docker/compose_yaml.go
index 3e00e1a..8c7a7fa 100644
--- a/internal/docker/compose_yaml.go
+++ b/internal/docker/compose_yaml.go
@@ -157,6 +157,66 @@ func ParseComposeYAML(content string) (map[string]interface{}, error) {
 	return compose, nil
 }
 
+// EnsureServiceEnvFile makes every service load envFile, so env written there
+// (e.g. an attached object store's connection details) actually reaches the
+// containers. An existing env_file (string or list form) is preserved and
+// envFile appended only when missing.
+func EnsureServiceEnvFile(content, envFile string) (string, error) {
+	compose, err := ParseComposeYAML(content)
+	if err != nil {
+		return content, err
+	}
+	services, ok := compose["services"].(map[string]interface{})
+	if !ok {
+		return content, nil
+	}
+
+	changed := false
+	for name, raw := range services {
+		service, ok := raw.(map[string]interface{})
+		if !ok {
+			continue
+		}
+
+		var files []string
+		switch ef := service["env_file"].(type) {
+		case string:
+			files = []string{ef}
+		case []interface{}:
+			for _, f := range ef {
+				if s, ok := f.(string); ok {
+					files = append(files, s)
+				}
+			}
+		}
+
+		found := false
+		for _, f := range files {
+			if f == envFile {
+				found = true
+				break
+			}
+		}
+		if found {
+			continue
+		}
+
+		files = append(files, envFile)
+		service["env_file"] = files
+		services[name] = service
+		changed = true
+	}
+
+	if !changed {
+		return content, nil
+	}
+	out, err := yaml.Marshal(compose)
+	if err != nil {
+		return content, err
+	}
+	return string(out), nil
+}
+
 // ContainerNameForService returns the DNS-resolvable container name for a service in
 // a deployment, matching what EnsureContainerNames assigns. An explicit container_name in
 // the compose wins; otherwise the EnsureContainerNames rule applies (primary service ->
diff --git a/internal/docker/manager.go b/internal/docker/manager.go
index 9755c3c..579a00c 100644
--- a/internal/docker/manager.go
+++ b/internal/docker/manager.go
@@ -54,6 +54,18 @@ func (m *Manager) indexContainersByProject(ctx context.Context) (containerIndex,
 	return index, nil
 }
 
+// ContainerPrimaryIP returns the first running container's address for a
+// deployment on the given docker network. A flatrun deploy names its compose
+// project after the deployment, so the project name is the deployment name.
+func (m *Manager) ContainerPrimaryIP(project, network string) (string, error) {
+	if m.apiClient == nil {
+		return "", fmt.Errorf("docker api client unavailable")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), statusReadTimeout)
+	defer cancel()
+	return m.apiClient.ContainerPrimaryIP(ctx, project, network)
+}
+
 // projectFor resolves a deployment's compose project name without shelling out.
 // It mirrors ComposeExecutor.getProjectName, except that the fallback probe for
 // an existing project reads the already-fetched index instead of running
diff --git a/pkg/config/config.go b/pkg/config/config.go
index 5f05151..0a9ecdf 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -180,11 +180,16 @@ type HealthConfig struct {
 }
 
 type InfrastructureConfig struct {
-	DefaultProxyNetwork    string               `yaml:"default_proxy_network" json:"default_proxy_network"`
-	DefaultDatabaseNetwork string               `yaml:"default_database_network" json:"default_database_network"`
-	Database               SharedDatabaseConfig `yaml:"database" json:"database"`
-	Redis                  SharedRedisConfig    `yaml:"redis" json:"redis"`
-	PowerDNS               PowerDNSConfig       `yaml:"powerdns" json:"powerdns"`
+	DefaultProxyNetwork    string `yaml:"default_proxy_network" json:"default_proxy_network"`
+	DefaultDatabaseNetwork string `yaml:"default_database_network" json:"default_database_network"`
+	// DefaultObjectStorageNetwork is the shared network self-hosted object
+	// stores join so apps can reach them by name. Empty reuses the database
+	// network (object storage is a data backend like a database); set it to run
+	// object stores on a dedicated network instead.
+	DefaultObjectStorageNetwork string               `yaml:"default_object_storage_network" json:"default_object_storage_network"`
+	Database                    SharedDatabaseConfig `yaml:"database" json:"database"`
+	Redis                       SharedRedisConfig    `yaml:"redis" json:"redis"`
+	PowerDNS                    PowerDNSConfig       `yaml:"powerdns" json:"powerdns"`
 }
 
 type PowerDNSConfig struct {
diff --git a/pkg/models/deployment.go b/pkg/models/deployment.go
index 2fd6049..b2f6110 100644
--- a/pkg/models/deployment.go
+++ b/pkg/models/deployment.go
@@ -176,18 +176,25 @@ type ContainerBackupPath struct {
 }
 
 type DatabaseBackupSpec struct {
-	Service     string `yaml:"service" json:"service"`
-	Type        string `yaml:"type" json:"type"`
-	HostEnv     string `yaml:"host_env,omitempty" json:"host_env,omitempty"`
-	PortEnv     string `yaml:"port_env,omitempty" json:"port_env,omitempty"`
-	UserEnv     string `yaml:"user_env,omitempty" json:"user_env,omitempty"`
-	PasswordEnv string `yaml:"password_env,omitempty" json:"password_env,omitempty"`
-	DatabaseEnv string `yaml:"database_env,omitempty" json:"database_env,omitempty"`
-	Host        string `yaml:"host,omitempty" json:"host,omitempty"`
-	Port        int    `yaml:"port,omitempty" json:"port,omitempty"`
-	User        string `yaml:"user,omitempty" json:"user,omitempty"`
-	Password    string `yaml:"password,omitempty" json:"password,omitempty"`
-	Database    string `yaml:"database,omitempty" json:"database,omitempty"`
+	Service string `yaml:"service" json:"service"`
+	Type    string `yaml:"type" json:"type"`
+	// Container, when set, is the exact container to exec the dump in, overriding
+	// the name derived from the deployment and service.
+	Container string `yaml:"container,omitempty" json:"container,omitempty"`
+	// AllDatabases dumps the whole server (pg_dumpall / mysqldump --all-databases)
+	// with root credentials, used to back up a database-server deployment such as
+	// the shared infrastructure database.
+	AllDatabases bool   `yaml:"all_databases,omitempty" json:"all_databases,omitempty"`
+	HostEnv      string `yaml:"host_env,omitempty" json:"host_env,omitempty"`
+	PortEnv      string `yaml:"port_env,omitempty" json:"port_env,omitempty"`
+	UserEnv      string `yaml:"user_env,omitempty" json:"user_env,omitempty"`
+	PasswordEnv  string `yaml:"password_env,omitempty" json:"password_env,omitempty"`
+	DatabaseEnv  string `yaml:"database_env,omitempty" json:"database_env,omitempty"`
+	Host         string `yaml:"host,omitempty" json:"host,omitempty"`
+	Port         int    `yaml:"port,omitempty" json:"port,omitempty"`
+	User         string `yaml:"user,omitempty" json:"user,omitempty"`
+	Password     string `yaml:"password,omitempty" json:"password,omitempty"`
+	Database     string `yaml:"database,omitempty" json:"database,omitempty"`
 }
 
 type BackupHookSpec struct {
diff --git a/templates/minio/docker-compose.yml b/templates/minio/docker-compose.yml
index 22d15ab..00cfc9d 100644
--- a/templates/minio/docker-compose.yml
+++ b/templates/minio/docker-compose.yml
@@ -3,6 +3,7 @@ services:
   minio:
     image: minio/minio:latest
     container_name: ${NAME}
+    user: "${FLATRUN_UID:-1000}:${FLATRUN_GID:-1000}"
     command: server /data --console-address ":9001"
     environment:
       MINIO_ROOT_USER: ${MINIO_ROOT_USER:-flatrun}
@@ -13,9 +14,10 @@ services:
       - "9000"
       - "9001"
     networks:
-      - proxy
+      - objstore
     restart: unless-stopped
 
 networks:
-  proxy:
+  objstore:
     external: true
+    name: ${FLATRUN_OBJECT_NETWORK:-database}
diff --git a/templates/minio/metadata.yml b/templates/minio/metadata.yml
index f1fb4f0..0285244 100644
--- a/templates/minio/metadata.yml
+++ b/templates/minio/metadata.yml
@@ -2,7 +2,13 @@ name: MinIO
 description: S3-compatible object storage you host yourself
 icon: pi pi-database
 logo: https://cdn.simpleicons.org/minio
-category: infrastructure
+category: storage
+object_store:
+  access_key_env: MINIO_ROOT_USER
+  secret_key_env: MINIO_ROOT_PASSWORD
+  api_port: 9000
+  region: us-east-1
+  use_path_style: true
 priority: 68
 container_port: 9000
 mounts:
diff --git a/templates/templates.go b/templates/templates.go
index 5168682..4900b4e 100644
--- a/templates/templates.go
+++ b/templates/templates.go
@@ -24,6 +24,7 @@ var Categories = []Category{
 	{ID: "framework", Name: "Frameworks", Icon: "pi pi-code", Priority: 90},
 	{ID: "runtime", Name: "Runtimes", Icon: "pi pi-cog", Priority: 80},
 	{ID: "infrastructure", Name: "Infrastructure", Icon: "pi pi-server", Priority: 70},
+	{ID: "storage", Name: "Storage", Icon: "pi pi-box", Priority: 65},
 	{ID: "database", Name: "Databases", Icon: "pi pi-database", Priority: 60},
 	{ID: "basic", Name: "Basic", Icon: "pi pi-file", Priority: 50},
 }