diff --git a/CLAUDE.md b/CLAUDE.md
index e39aa56f..4537c08b 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -124,9 +124,12 @@ lstk proxies third-party IaC tools at the AWS emulator so they run against Local
 A REF is parsed by helpers in `internal/snapshot/destination.go`:
 - **local file** — absolute/relative path; the `.snapshot` extension is forced (any other extension is replaced). On load, `.zip` files saved by older lstk versions are still accepted.
 - **cloud snapshot** — `pod:` prefix (e.g. `pod:my-baseline`), stored on the LocalStack platform. Requires auth (`LOCALSTACK_AUTH_TOKEN` or `lstk login`).
+- **S3 remote** — `s3://bucket/prefix` (parsed to `KindS3`). The CLI never touches S3; the emulator performs the transfer. See the S3 remotes note below.
 
 `ParseDestination` (save), `ParseSource` (load), `ParseRemovable` (remove), and `ParseShowable` (show) share pod-name validation; `ParseRemovable` and `ParseShowable` reject local paths (via the shared `parseCloudOnly` helper) so those cloud-only commands never touch local files.
 
+**S3 remotes (save/load/list only).** `lstk snapshot save <pod-name> s3://bucket/prefix`, `load <pod-name> s3://bucket/prefix`, and `list s3://bucket/prefix` store snapshots in the user's own S3 bucket. The pod name (the snapshot's identity within the bucket) is a positional separate from the `s3://` location — required for load, auto-generated for save when omitted, unused for list. Credentials follow AWS CLI precedence (`resolveS3Credentials` in `cmd/snapshot.go`): `--profile <name>` wins, else static `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY` (optional `AWS_SESSION_TOKEN`), else the profile named by `AWS_PROFILE` (read via `internal/awsconfig.ReadProfileCredentials`/`CredentialsFromEnv`); only static credentials are supported (no SSO/assume-role/credential_process — those resolve only via the AWS SDK chain, not our ini parser); **never put credentials in the URL** (rejected by `parseS3`). The emulator-side S3 remote requires a remote to be registered by name first, so the CLI transparently upserts a deterministic remote (`POST /_localstack/pods/remotes/<name>`, name derived in `internal/snapshot/remote.go`) with a placeholder-templated URL, then passes the real credentials as ephemeral `remote_params` on each op — secrets stay out of the registered URL and out of any persisted state. `list s3://…` queries the emulator (`GET /_localstack/pods` with a remote body), not the platform API, so it **requires a running emulator** (unlike `snapshot list`). Before save/load/list, lstk runs a pre-flight bucket-existence check (`ensureBucketExists` → `RemoteClient.S3BucketExists`, an unsigned S3 `HEAD`: 404 ⇒ missing) and errors out rather than letting the emulator auto-create a bucket on a typo; local-testing endpoints (IP / `host.docker.internal`) are skipped, and a check that can't run degrades to a warning. Domain logic + client interface live in `internal/snapshot/remote.go`; the emulator HTTP impl is `internal/emulator/aws/remote.go`; command wiring/arg classification (`classifyRemoteArgs`, `resolveS3Credentials`) is in `cmd/snapshot.go`. ORAS and other remotes, and `remove`/`show`/versions on S3, are not yet supported.
+
 **Auto-load on start.** A `[[containers]]` block (AWS only) can set `snapshot = "pod:my-baseline"` (any load REF) to auto-load that snapshot after the emulator starts. The loader runs only when the emulator is freshly started this run (skipped when already running), mirroring v1's `AUTO_LOAD_POD`. `lstk start --snapshot REF` overrides the configured REF for one run; `lstk start --no-snapshot` skips it. Resolution lives in `resolveStartSnapshotRef`/`newSnapshotAutoLoader` in `cmd/snapshot.go`; the loader is threaded into the non-interactive start in `cmd/root.go` and into the TUI via `ui.RunOptions.PostStart`. `snapshot save` never writes back into config — the `snapshot` field is manual.
 
 # Code Style
diff --git a/README.md b/README.md
index 8cba41c6..711e22f0 100644
--- a/README.md
+++ b/README.md
@@ -45,7 +45,7 @@ Running `lstk` will automatically handle configuration setup and start LocalStac
 - **Interactive TUI** — a Bubble Tea-powered terminal UI shown in an interactive terminal for commands like `start`, `login`, `status`, etc.
 - **Plain output** for CI/CD and scripting (auto-detected in non-interactive environments or forced with `--non-interactive`)
 - **Log streaming** — tail emulator logs in real-time with `--follow`; use `--verbose` to show all logs without filtering
-- **Snapshots** — save, load, and remove emulator state as local files or named cloud snapshots (`pod:` prefix), and auto-load one on start
+- **Snapshots** — save, load, and remove emulator state as local files, named cloud snapshots (`pod:` prefix), or in your own S3 bucket (`s3://`), and auto-load one on start
 - **Browser-based login** — authenticate via browser and store credentials securely in the system keyring
 - **AWS CLI profile** — optionally configure a `localstack` profile in `~/.aws/` after start
 - **Terraform integration** — proxy Terraform commands to LocalStack with automatic AWS provider endpoint configuration
@@ -311,12 +311,19 @@ lstk snapshot save ./my-snapshot.snapshot
 # Save emulator state as a named cloud snapshot on the LocalStack platform
 lstk snapshot save pod:my-baseline
 
+# Save to your own S3 bucket (credentials from AWS_* env vars or --profile)
+lstk snapshot save my-pod s3://my-bucket/prefix
+
 # Load a snapshot back into the running emulator
 lstk snapshot load pod:my-baseline
+lstk snapshot load my-pod s3://my-bucket/prefix
 
 # List cloud snapshots on the LocalStack platform (--all for the whole organization)
 lstk snapshot list
 
+# List snapshots in your own S3 bucket (requires a running emulator)
+lstk snapshot list s3://my-bucket/prefix
+
 # Show metadata for a single cloud snapshot
 lstk snapshot show pod:my-baseline
 
@@ -345,23 +352,29 @@ lstk cdk synth
 
 Snapshots capture the running emulator's state so you can restore it later.
 
-A snapshot reference is either a **local file** or a **cloud snapshot**:
+A snapshot reference is a **local file**, a **cloud snapshot**, or an **S3 remote**:
 
 - **Local file** — an absolute or relative path. A `.snapshot` extension is added if omitted (snapshots saved as `.zip` by older lstk versions still load).
 - **Cloud snapshot** — a name with the `pod:` prefix (e.g. `pod:my-baseline`), stored on the LocalStack platform. Requires authentication (`LOCALSTACK_AUTH_TOKEN` or `lstk login`).
+- **S3 remote** — an `s3://bucket/prefix` location backed by your own S3 bucket. Supported by `save`, `load`, and `list`.
 
 ```bash
-# Save (local or cloud)
+# Save (local, cloud, or S3)
 lstk snapshot save ./my-snapshot.snapshot
 lstk snapshot save pod:my-baseline
+lstk snapshot save my-pod s3://my-bucket/prefix
 
 # Load (starts the emulator first if needed)
 lstk snapshot load pod:my-baseline
+lstk snapshot load my-pod s3://my-bucket/prefix
 
 # List cloud snapshots — only your own by default, --all for the whole organization
 lstk snapshot list
 lstk snapshot list --all
 
+# List snapshots in an S3 bucket
+lstk snapshot list s3://my-bucket/prefix
+
 # Show metadata for a single cloud snapshot
 lstk snapshot show pod:my-baseline
 
@@ -372,6 +385,26 @@ lstk snapshot remove pod:my-baseline --force  # skip the prompt (required in non
 
 `lstk snapshot load` supports merge strategies via `--merge` (`account-region-merge` (default), `overwrite`, `service-merge`) to control how snapshot state combines with running state.
 
+### S3 remotes
+
+`save`, `load`, and `list` can target your own S3 bucket with an `s3://bucket/prefix` location. The pod name (the snapshot's identity within the bucket) is a positional argument separate from the `s3://` location — required for `load`, auto-generated for `save` when omitted, and unused for `list`.
+
+Credentials come from `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` (and optional `AWS_SESSION_TOKEN`), or from a named profile via `--profile <name>`. **Never put credentials in the URL** — lstk rejects an `s3://` ref that embeds them. lstk itself never touches S3: the running emulator performs the transfer, so these commands require a running emulator, and `list s3://…` queries the emulator rather than the LocalStack platform.
+
+```bash
+export AWS_ACCESS_KEY_ID=...
+export AWS_SECRET_ACCESS_KEY=...
+
+lstk snapshot save my-pod s3://my-bucket/prefix
+lstk snapshot load my-pod s3://my-bucket/prefix
+lstk snapshot list s3://my-bucket/prefix
+
+# Or read credentials from a named AWS profile instead of env vars
+lstk snapshot save my-pod s3://my-bucket/prefix --profile my-aws-profile
+```
+
+The S3 bucket must already exist — lstk checks up front and errors out rather than creating it on a typo. `remove` and `show` are not yet supported for S3 remotes.
+
 ### Auto-load on start
 
 The AWS emulator can automatically load a snapshot whenever it starts. Set the `snapshot` field on its `[[containers]]` block to any snapshot reference — a local file or a `pod:` cloud snapshot:
diff --git a/cmd/snapshot.go b/cmd/snapshot.go
index a5600777..00862e61 100644
--- a/cmd/snapshot.go
+++ b/cmd/snapshot.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/localstack/lstk/internal/api"
+	"github.com/localstack/lstk/internal/awsconfig"
 	"github.com/localstack/lstk/internal/config"
 	"github.com/localstack/lstk/internal/container"
 	"github.com/localstack/lstk/internal/emulator/aws"
@@ -26,7 +27,12 @@ const snapshotSaveCanonical = "snapshot save"
 
 const snapshotListLong = `List Cloud Pod snapshots available on the LocalStack platform.
 
-By default only snapshots you created are listed. Pass --all to include all snapshots in your organisation.`
+By default only snapshots you created are listed. Pass --all to include all snapshots in your organisation.
+
+To list snapshots in your own S3 bucket, pass an s3:// location (requires a running emulator). Credentials are read from AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY, from --profile, or from the profile named by AWS_PROFILE:
+
+  lstk snapshot list s3://my-bucket/prefix
+  lstk snapshot list s3://my-bucket/prefix --profile my-aws-profile`
 
 const snapshotShowLong = `Show metadata for a cloud snapshot on the LocalStack platform.
 
@@ -52,7 +58,12 @@ Pass [destination] as an absolute or relative path for the exported file:
 
 To save to a remote pod on the LocalStack platform, use the pod: prefix:
 
-  lstk snapshot save pod:my-baseline    # saves as a named pod on the platform`
+  lstk snapshot save pod:my-baseline    # saves as a named pod on the platform
+
+To save to your own S3 bucket, pass an s3:// location with an optional pod name (auto-generated when omitted). Credentials are read from AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY, from --profile, or from the profile named by AWS_PROFILE; never put credentials in the URL.
+
+  lstk snapshot save my-pod s3://my-bucket/prefix
+  lstk snapshot save my-pod s3://my-bucket/prefix --profile my-aws-profile`
 
 const snapshotLoadCanonical = "snapshot load"
 
@@ -64,6 +75,11 @@ REF identifies the snapshot to load:
   lstk snapshot load ./checkpoint.snapshot   # loads from explicit path
   lstk snapshot load pod:my-baseline         # loads from LocalStack Cloud
 
+To load from your own S3 bucket, pass the pod name and an s3:// location. Credentials are read from AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY, from --profile, or from the profile named by AWS_PROFILE:
+
+  lstk snapshot load my-pod s3://my-bucket/prefix
+  lstk snapshot load my-pod s3://my-bucket/prefix --profile my-aws-profile
+
 Merge strategies control how snapshot state is combined with running state:
 
   --merge=account-region-merge  (default) snapshot wins on (service, account, region) overlap
@@ -176,11 +192,12 @@ func newSnapshotLoadCmd(cfg *env.Env, tel *telemetry.Client, logger log.Logger)
 		Use:     "load REF",
 		Short:   "Load a snapshot into the running emulator",
 		Long:    snapshotLoadLong,
-		Args:    cobra.ExactArgs(1),
+		Args:    cobra.RangeArgs(1, 2),
 		PreRunE: initConfig(nil),
 		RunE:    runSnapshotLoad(cfg, tel, logger),
 	}
 	addMergeFlag(cmd)
+	addProfileFlag(cmd)
 	return cmd
 }
 
@@ -189,12 +206,13 @@ func newLoadCmd(cfg *env.Env, tel *telemetry.Client, logger log.Logger) *cobra.C
 		Use:         "load REF",
 		Short:       "Load a snapshot into the running emulator",
 		Long:        snapshotLoadLong,
-		Args:        cobra.ExactArgs(1),
+		Args:        cobra.RangeArgs(1, 2),
 		PreRunE:     initConfig(nil),
 		RunE:        runSnapshotLoad(cfg, tel, logger),
 		Annotations: map[string]string{canonicalCommandAnnotation: snapshotLoadCanonical},
 	}
 	addMergeFlag(cmd)
+	addProfileFlag(cmd)
 	return cmd
 }
 
@@ -208,17 +226,53 @@ func runSnapshotLoad(cfg *env.Env, tel *telemetry.Client, logger log.Logger) fun
 		if err != nil {
 			return err
 		}
+		profile, err := cmd.Flags().GetString("profile")
+		if err != nil {
+			return err
+		}
+		if err := snapshot.ValidateMergeStrategy(strategy); err != nil {
+			return err
+		}
 
 		home, err := os.UserHomeDir()
 		if err != nil {
 			return err
 		}
-		src, err := snapshot.ParseSource(args[0], home)
+
+		podName, s3URL, isRemote, err := classifyRemoteArgs(args)
 		if err != nil {
 			return err
 		}
 
-		if err := snapshot.ValidateMergeStrategy(strategy); err != nil {
+		if isRemote {
+			if podName == "" {
+				return fmt.Errorf("a pod name is required to load from S3: lstk snapshot load <pod-name> %s", s3URL)
+			}
+			if err := snapshot.ValidatePodName(podName); err != nil {
+				return err
+			}
+			src, err := snapshot.ParseSource(s3URL, home)
+			if err != nil {
+				return err
+			}
+			creds, err := resolveS3Credentials(profile)
+			if err != nil {
+				return err
+			}
+			rt, client, host, containers, appConfig, err := resolveSnapshotDeps(cmd.Context(), cfg)
+			if err != nil {
+				return err
+			}
+			starter := buildStarter(cfg, rt, appConfig, logger, tel)
+			if isInteractiveMode(cfg) {
+				return ui.RunSnapshotLoadRemoteS3(cmd.Context(), rt, containers, client, host, podName, src.Value, creds, cfg.AuthToken, strategy, starter)
+			}
+			sink := output.NewPlainSink(os.Stdout)
+			return snapshot.LoadRemoteS3(cmd.Context(), rt, containers, client, host, podName, src.Value, creds, cfg.AuthToken, strategy, starter, sink)
+		}
+
+		src, err := snapshot.ParseSource(args[0], home)
+		if err != nil {
 			return err
 		}
 
@@ -324,16 +378,83 @@ func resolveSnapshotDeps(ctx context.Context, cfg *env.Env) (rt runtime.Runtime,
 	return rt, aws.NewClient(), host, []config.ContainerConfig{target}, appConfig, nil
 }
 
+// addProfileFlag registers the --profile flag used to source AWS credentials for
+// S3 remote snapshots.
+func addProfileFlag(cmd *cobra.Command) {
+	cmd.Flags().String("profile", "", "AWS profile to read S3 credentials from (defaults to AWS_* env vars, then AWS_PROFILE)")
+}
+
+// classifyRemoteArgs inspects positional args for an s3:// location. When one is
+// present, it returns the S3 URL and the optional pod name (the other positional);
+// ok is false when no s3:// ref is given, so the caller uses the local/pod path.
+func classifyRemoteArgs(args []string) (podName, s3URL string, ok bool, err error) {
+	for _, a := range args {
+		if snapshot.IsS3Ref(a) {
+			if s3URL != "" {
+				return "", "", false, fmt.Errorf("only one s3:// location may be given")
+			}
+			s3URL = a
+			continue
+		}
+		if podName != "" {
+			return "", "", false, fmt.Errorf("unexpected extra argument %q", a)
+		}
+		podName = a
+	}
+	if s3URL == "" {
+		return "", "", false, nil
+	}
+	return podName, s3URL, true, nil
+}
+
+// resolveS3Credentials reads AWS credentials for an S3 remote, following the
+// AWS CLI precedence: an explicit --profile flag wins; otherwise static AWS_*
+// environment variables win; otherwise the profile named by AWS_PROFILE is used.
+func resolveS3Credentials(profile string) (snapshot.S3Credentials, error) {
+	var (
+		creds awsconfig.Credentials
+		err   error
+	)
+	switch {
+	case profile != "":
+		creds, err = awsconfig.ReadProfileCredentials(profile)
+		if err != nil {
+			return snapshot.S3Credentials{}, err
+		}
+	default:
+		creds, err = awsconfig.CredentialsFromEnv()
+		if errors.Is(err, awsconfig.ErrNoCredentials) {
+			if envProfile := os.Getenv("AWS_PROFILE"); envProfile != "" {
+				creds, err = awsconfig.ReadProfileCredentials(envProfile)
+				if err != nil {
+					return snapshot.S3Credentials{}, err
+				}
+				break
+			}
+			return snapshot.S3Credentials{}, fmt.Errorf("AWS credentials required for S3 snapshots: set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY, set AWS_PROFILE, or pass --profile <name>")
+		}
+		if err != nil {
+			return snapshot.S3Credentials{}, err
+		}
+	}
+	return snapshot.S3Credentials{
+		AccessKeyID:     creds.AccessKeyID,
+		SecretAccessKey: creds.SecretAccessKey,
+		SessionToken:    creds.SessionToken,
+	}, nil
+}
+
 func newSnapshotListCmd(cfg *env.Env, logger log.Logger) *cobra.Command {
 	cmd := &cobra.Command{
-		Use:     "list",
+		Use:     "list [s3://bucket/prefix]",
 		Short:   "List Cloud Pod snapshots available on the LocalStack platform",
 		Long:    snapshotListLong,
-		Args:    cobra.NoArgs,
+		Args:    cobra.MaximumNArgs(1),
 		PreRunE: initConfig(nil),
 		RunE:    runSnapshotList(cfg, logger),
 	}
 	cmd.Flags().Bool("all", false, "List all snapshots in the organisation")
+	addProfileFlag(cmd)
 	return cmd
 }
 
@@ -343,6 +464,38 @@ func runSnapshotList(cfg *env.Env, logger log.Logger) func(*cobra.Command, []str
 		if err != nil {
 			return err
 		}
+		profile, err := cmd.Flags().GetString("profile")
+		if err != nil {
+			return err
+		}
+
+		if len(args) == 1 && snapshot.IsS3Ref(args[0]) {
+			home, err := os.UserHomeDir()
+			if err != nil {
+				return err
+			}
+			src, err := snapshot.ParseSource(args[0], home)
+			if err != nil {
+				return err
+			}
+			creds, err := resolveS3Credentials(profile)
+			if err != nil {
+				return err
+			}
+			rt, client, host, containers, _, err := resolveSnapshotDeps(cmd.Context(), cfg)
+			if err != nil {
+				return err
+			}
+			if isInteractiveMode(cfg) {
+				return ui.RunSnapshotListRemoteS3(cmd.Context(), rt, containers, client, host, src.Value, creds, cfg.AuthToken)
+			}
+			sink := output.NewPlainSink(os.Stdout)
+			return snapshot.ListRemoteS3(cmd.Context(), rt, containers, client, host, src.Value, creds, cfg.AuthToken, sink)
+		}
+		if len(args) == 1 {
+			return fmt.Errorf("unexpected argument %q: snapshot list takes an optional s3:// location", args[0])
+		}
+
 		creator := "me"
 		if all {
 			creator = ""
@@ -393,39 +546,78 @@ func runSnapshotShow(cfg *env.Env, logger log.Logger) func(*cobra.Command, []str
 }
 
 func newSnapshotSaveCmd(cfg *env.Env) *cobra.Command {
-	return &cobra.Command{
+	cmd := &cobra.Command{
 		Use:     "save [destination]",
 		Short:   "Save a snapshot of the emulator state",
 		Long:    snapshotSaveLong,
-		Args:    cobra.MaximumNArgs(1),
+		Args:    cobra.MaximumNArgs(2),
 		PreRunE: initConfig(nil),
 		RunE:    runSnapshotSave(cfg),
 	}
+	addProfileFlag(cmd)
+	return cmd
 }
 
 func newSaveCmd(cfg *env.Env) *cobra.Command {
-	return &cobra.Command{
+	cmd := &cobra.Command{
 		Use:         "save [destination]",
 		Short:       "Save a snapshot of the emulator state",
 		Long:        snapshotSaveLong,
-		Args:        cobra.MaximumNArgs(1),
+		Args:        cobra.MaximumNArgs(2),
 		PreRunE:     initConfig(nil),
 		RunE:        runSnapshotSave(cfg),
 		Annotations: map[string]string{canonicalCommandAnnotation: snapshotSaveCanonical},
 	}
+	addProfileFlag(cmd)
+	return cmd
 }
 
 func runSnapshotSave(cfg *env.Env) func(*cobra.Command, []string) error {
 	return func(cmd *cobra.Command, args []string) error {
-		var destArg string
-		if len(args) > 0 {
-			destArg = args[0]
+		profile, err := cmd.Flags().GetString("profile")
+		if err != nil {
+			return err
+		}
+
+		podName, s3URL, isRemote, err := classifyRemoteArgs(args)
+		if err != nil {
+			return err
 		}
 
 		home, err := os.UserHomeDir()
 		if err != nil {
 			return err
 		}
+
+		if isRemote {
+			dest, err := snapshot.ParseDestination(s3URL, home, time.Now())
+			if err != nil {
+				return err
+			}
+			if podName == "" {
+				podName = snapshot.DefaultRemotePodName(time.Now())
+			} else if err := snapshot.ValidatePodName(podName); err != nil {
+				return err
+			}
+			creds, err := resolveS3Credentials(profile)
+			if err != nil {
+				return err
+			}
+			rt, client, host, containers, _, err := resolveSnapshotDeps(cmd.Context(), cfg)
+			if err != nil {
+				return err
+			}
+			if isInteractiveMode(cfg) {
+				return ui.RunSnapshotSaveRemoteS3(cmd.Context(), rt, containers, client, host, podName, dest.Value, creds, cfg.AuthToken)
+			}
+			sink := output.NewPlainSink(os.Stdout)
+			return snapshot.SaveRemoteS3(cmd.Context(), rt, containers, client, host, podName, dest.Value, creds, cfg.AuthToken, sink)
+		}
+
+		var destArg string
+		if len(args) > 0 {
+			destArg = args[0]
+		}
 		dest, err := snapshot.ParseDestination(destArg, home, time.Now())
 		if err != nil {
 			return err
diff --git a/cmd/snapshot_credentials_test.go b/cmd/snapshot_credentials_test.go
new file mode 100644
index 00000000..15668f35
--- /dev/null
+++ b/cmd/snapshot_credentials_test.go
@@ -0,0 +1,78 @@
+package cmd
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// writeProfileCreds writes a credentials file with the given profile section and
+// points AWS_SHARED_CREDENTIALS_FILE at it (config file absent).
+func writeProfileCreds(t *testing.T, profile, body string) {
+	t.Helper()
+	dir := t.TempDir()
+	credsPath := filepath.Join(dir, "credentials")
+	require.NoError(t, os.WriteFile(credsPath, []byte("["+profile+"]\n"+body), 0600))
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", credsPath)
+	t.Setenv("AWS_CONFIG_FILE", filepath.Join(dir, "config")) // absent
+}
+
+// clearStaticCreds unsets the static AWS credential env vars so each test starts
+// from a known state.
+func clearStaticCreds(t *testing.T) {
+	t.Helper()
+	t.Setenv("AWS_ACCESS_KEY_ID", "")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "")
+	t.Setenv("AWS_SESSION_TOKEN", "")
+	t.Setenv("AWS_PROFILE", "")
+}
+
+func TestResolveS3Credentials_AWSProfileEnv(t *testing.T) {
+	clearStaticCreds(t)
+	writeProfileCreds(t, "work", "aws_access_key_id = AKIAWORK\naws_secret_access_key = worksecret\naws_session_token = worktoken\n")
+	t.Setenv("AWS_PROFILE", "work")
+
+	creds, err := resolveS3Credentials("")
+	require.NoError(t, err)
+	assert.Equal(t, "AKIAWORK", creds.AccessKeyID)
+	assert.Equal(t, "worksecret", creds.SecretAccessKey)
+	assert.Equal(t, "worktoken", creds.SessionToken)
+}
+
+func TestResolveS3Credentials_StaticEnvWinsOverAWSProfile(t *testing.T) {
+	clearStaticCreds(t)
+	writeProfileCreds(t, "work", "aws_access_key_id = AKIAWORK\naws_secret_access_key = worksecret\n")
+	t.Setenv("AWS_PROFILE", "work")
+	t.Setenv("AWS_ACCESS_KEY_ID", "AKIAENV")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "envsecret")
+
+	creds, err := resolveS3Credentials("")
+	require.NoError(t, err)
+	assert.Equal(t, "AKIAENV", creds.AccessKeyID)
+	assert.Equal(t, "envsecret", creds.SecretAccessKey)
+}
+
+func TestResolveS3Credentials_FlagWinsOverAWSProfile(t *testing.T) {
+	clearStaticCreds(t)
+	writeProfileCreds(t, "flagprofile", "aws_access_key_id = AKIAFLAG\naws_secret_access_key = flagsecret\n")
+	t.Setenv("AWS_PROFILE", "missing")
+
+	creds, err := resolveS3Credentials("flagprofile")
+	require.NoError(t, err)
+	assert.Equal(t, "AKIAFLAG", creds.AccessKeyID)
+	assert.Equal(t, "flagsecret", creds.SecretAccessKey)
+}
+
+func TestResolveS3Credentials_NoneSet(t *testing.T) {
+	clearStaticCreds(t)
+	dir := t.TempDir()
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(dir, "credentials")) // absent
+	t.Setenv("AWS_CONFIG_FILE", filepath.Join(dir, "config"))                  // absent
+
+	_, err := resolveS3Credentials("")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "AWS_PROFILE")
+}
diff --git a/internal/awsconfig/credentials.go b/internal/awsconfig/credentials.go
new file mode 100644
index 00000000..08f9a087
--- /dev/null
+++ b/internal/awsconfig/credentials.go
@@ -0,0 +1,127 @@
+package awsconfig
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"gopkg.in/ini.v1"
+)
+
+// Credentials holds static AWS credentials read from the environment or a profile.
+// SessionToken is optional.
+type Credentials struct {
+	AccessKeyID     string
+	SecretAccessKey string
+	SessionToken    string
+}
+
+// credentialsFilePath returns the shared credentials file path, honoring
+// AWS_SHARED_CREDENTIALS_FILE, defaulting to ~/.aws/credentials.
+func credentialsFilePath() (string, error) {
+	if p := os.Getenv("AWS_SHARED_CREDENTIALS_FILE"); p != "" {
+		return p, nil
+	}
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return "", err
+	}
+	return filepath.Join(home, ".aws", "credentials"), nil
+}
+
+// configFilePath returns the AWS config file path, honoring AWS_CONFIG_FILE,
+// defaulting to ~/.aws/config.
+func configFilePath() (string, error) {
+	if p := os.Getenv("AWS_CONFIG_FILE"); p != "" {
+		return p, nil
+	}
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return "", err
+	}
+	return filepath.Join(home, ".aws", "config"), nil
+}
+
+// readCredsFromSection extracts credentials from an ini section, returning ok=false
+// when the access key or secret key is absent.
+func readCredsFromSection(s *ini.Section) (Credentials, bool) {
+	access := s.Key("aws_access_key_id").String()
+	secret := s.Key("aws_secret_access_key").String()
+	if access == "" || secret == "" {
+		return Credentials{}, false
+	}
+	return Credentials{
+		AccessKeyID:     access,
+		SecretAccessKey: secret,
+		SessionToken:    s.Key("aws_session_token").String(),
+	}, true
+}
+
+// lookupProfile loads file at path and returns credentials from the section named
+// sectionName, if both the file and a complete credential pair are present.
+func lookupProfile(path, sectionName string) (Credentials, bool) {
+	f, err := ini.Load(path)
+	if err != nil {
+		return Credentials{}, false
+	}
+	s, err := f.GetSection(sectionName)
+	if err != nil {
+		return Credentials{}, false
+	}
+	return readCredsFromSection(s)
+}
+
+// ReadProfileCredentials resolves AWS credentials for the named profile from the
+// shared credentials file (~/.aws/credentials, section "[<profile>]"), falling back
+// to the config file (~/.aws/config, section "[profile <profile>]", or "[default]"
+// for the default profile). It honors AWS_SHARED_CREDENTIALS_FILE and AWS_CONFIG_FILE.
+func ReadProfileCredentials(profile string) (Credentials, error) {
+	if profile == "" {
+		profile = "default"
+	}
+
+	credsPath, err := credentialsFilePath()
+	if err != nil {
+		return Credentials{}, err
+	}
+	if creds, ok := lookupProfile(credsPath, profile); ok {
+		return creds, nil
+	}
+
+	configPath, err := configFilePath()
+	if err != nil {
+		return Credentials{}, err
+	}
+	// The config file names the default profile "[default]" and others
+	// "[profile <name>]".
+	configSection := "profile " + profile
+	if profile == "default" {
+		configSection = "default"
+	}
+	if creds, ok := lookupProfile(configPath, configSection); ok {
+		return creds, nil
+	}
+
+	return Credentials{}, fmt.Errorf("could not find AWS credentials for profile %q in %s or %s", profile, credsPath, configPath)
+}
+
+// ErrNoCredentials is returned by CredentialsFromEnv when the required AWS
+// credential environment variables are not set.
+var ErrNoCredentials = errors.New("AWS credentials not found in environment")
+
+// CredentialsFromEnv reads credentials from AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY,
+// and the optional AWS_SESSION_TOKEN. It returns ErrNoCredentials when either of the
+// two required variables is unset.
+func CredentialsFromEnv() (Credentials, error) {
+	access := os.Getenv("AWS_ACCESS_KEY_ID")
+	secret := os.Getenv("AWS_SECRET_ACCESS_KEY")
+	if access == "" || secret == "" {
+		return Credentials{}, ErrNoCredentials
+	}
+	return Credentials{
+		AccessKeyID:     access,
+		SecretAccessKey: secret,
+		SessionToken:    os.Getenv("AWS_SESSION_TOKEN"),
+	}, nil
+}
diff --git a/internal/awsconfig/credentials_test.go b/internal/awsconfig/credentials_test.go
new file mode 100644
index 00000000..fe90661f
--- /dev/null
+++ b/internal/awsconfig/credentials_test.go
@@ -0,0 +1,89 @@
+package awsconfig
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCredentialsFromEnv(t *testing.T) {
+	t.Setenv("AWS_ACCESS_KEY_ID", "AKIA123")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "secret")
+	t.Setenv("AWS_SESSION_TOKEN", "token")
+
+	creds, err := CredentialsFromEnv()
+	require.NoError(t, err)
+	assert.Equal(t, "AKIA123", creds.AccessKeyID)
+	assert.Equal(t, "secret", creds.SecretAccessKey)
+	assert.Equal(t, "token", creds.SessionToken)
+}
+
+func TestCredentialsFromEnv_Missing(t *testing.T) {
+	t.Setenv("AWS_ACCESS_KEY_ID", "")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "")
+
+	_, err := CredentialsFromEnv()
+	require.ErrorIs(t, err, ErrNoCredentials)
+}
+
+func TestReadProfileCredentials_FromCredentialsFile(t *testing.T) {
+	dir := t.TempDir()
+	credsPath := filepath.Join(dir, "credentials")
+	require.NoError(t, os.WriteFile(credsPath, []byte(`[work]
+aws_access_key_id = AKIAWORK
+aws_secret_access_key = worksecret
+aws_session_token = worktoken
+`), 0600))
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", credsPath)
+	t.Setenv("AWS_CONFIG_FILE", filepath.Join(dir, "config")) // absent
+
+	creds, err := ReadProfileCredentials("work")
+	require.NoError(t, err)
+	assert.Equal(t, "AKIAWORK", creds.AccessKeyID)
+	assert.Equal(t, "worksecret", creds.SecretAccessKey)
+	assert.Equal(t, "worktoken", creds.SessionToken)
+}
+
+func TestReadProfileCredentials_FallsBackToConfigFile(t *testing.T) {
+	dir := t.TempDir()
+	configPath := filepath.Join(dir, "config")
+	require.NoError(t, os.WriteFile(configPath, []byte(`[profile work]
+aws_access_key_id = AKIACONF
+aws_secret_access_key = confsecret
+`), 0600))
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(dir, "credentials")) // absent
+	t.Setenv("AWS_CONFIG_FILE", configPath)
+
+	creds, err := ReadProfileCredentials("work")
+	require.NoError(t, err)
+	assert.Equal(t, "AKIACONF", creds.AccessKeyID)
+	assert.Equal(t, "confsecret", creds.SecretAccessKey)
+}
+
+func TestReadProfileCredentials_DefaultProfileInConfig(t *testing.T) {
+	dir := t.TempDir()
+	configPath := filepath.Join(dir, "config")
+	require.NoError(t, os.WriteFile(configPath, []byte(`[default]
+aws_access_key_id = AKIADEF
+aws_secret_access_key = defsecret
+`), 0600))
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(dir, "credentials")) // absent
+	t.Setenv("AWS_CONFIG_FILE", configPath)
+
+	creds, err := ReadProfileCredentials("")
+	require.NoError(t, err)
+	assert.Equal(t, "AKIADEF", creds.AccessKeyID)
+}
+
+func TestReadProfileCredentials_Missing(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(dir, "credentials"))
+	t.Setenv("AWS_CONFIG_FILE", filepath.Join(dir, "config"))
+
+	_, err := ReadProfileCredentials("ghost")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "ghost")
+}
diff --git a/internal/emulator/aws/client.go b/internal/emulator/aws/client.go
index 07c1f092..38a6a22d 100644
--- a/internal/emulator/aws/client.go
+++ b/internal/emulator/aws/client.go
@@ -20,6 +20,9 @@ import (
 
 type Client struct {
 	http *http.Client
+	// s3BucketURLTemplate builds the URL for an S3 bucket existence check; it
+	// contains a single %s for the bucket name. Overridable in tests.
+	s3BucketURLTemplate string
 }
 
 func NewClient() *Client {
@@ -32,6 +35,7 @@ func NewClient() *Client {
 				}),
 			),
 		},
+		s3BucketURLTemplate: "https://%s.s3.amazonaws.com/",
 	}
 }
 
@@ -248,138 +252,11 @@ func isPodNotFoundMsg(msg string) bool {
 }
 
 func (c *Client) LoadPodSnapshot(ctx context.Context, host, podName, authToken, strategy string) ([]string, error) {
-	url := fmt.Sprintf("http://%s/_localstack/pods/%s", host, podName)
-	if strategy != "" {
-		url += "?merge=" + strategy
-	}
-	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader([]byte("{}")))
-	if err != nil {
-		return nil, fmt.Errorf("create request: %w", err)
-	}
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(":"+authToken)))
-
-	resp, err := c.http.Do(req)
-	if err != nil {
-		return nil, fmt.Errorf("connect to LocalStack: %w", err)
-	}
-	defer func() { _ = resp.Body.Close() }()
-
-	if resp.StatusCode == http.StatusUnprocessableEntity {
-		body, _ := io.ReadAll(resp.Body)
-		return nil, fmt.Errorf("%w: %s", snapshot.ErrIncompatibleSnapshot, strings.TrimSpace(string(body)))
-	}
-	if resp.StatusCode != http.StatusOK {
-		body, _ := io.ReadAll(resp.Body)
-		return nil, fmt.Errorf("pod load failed (HTTP %d): %s", resp.StatusCode, strings.TrimSpace(string(body)))
-	}
-
-	var services []string
-	scanner := bufio.NewScanner(resp.Body)
-	buf := make([]byte, 1024*1024)
-	scanner.Buffer(buf, 1024*1024)
-	for scanner.Scan() {
-		line := strings.TrimSpace(scanner.Text())
-		if line == "" {
-			continue
-		}
-		var event struct {
-			Event   string `json:"event"`
-			Service string `json:"service"`
-			Status  string `json:"status"`
-			Message string `json:"message"`
-		}
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			continue
-		}
-		switch event.Event {
-		case "service":
-			switch event.Status {
-			case "ok":
-				services = append(services, event.Service)
-			case "error":
-				if isInvalidSnapshotFileMsg(event.Message) {
-					return nil, snapshot.ErrInvalidSnapshotFile
-				}
-				return nil, fmt.Errorf("load failed for service %s: %s", event.Service, event.Message)
-			}
-		case "completion":
-			if event.Status != "ok" {
-				if isInvalidSnapshotFileMsg(event.Message) {
-					return nil, snapshot.ErrInvalidSnapshotFile
-				}
-				if isPodNotFoundMsg(event.Message) {
-					return nil, snapshot.ErrPodNotFound
-				}
-				return nil, fmt.Errorf("pod load failed: %s", event.Message)
-			}
-			return services, nil
-		}
-	}
-	if err := scanner.Err(); err != nil {
-		return nil, fmt.Errorf("reading response: %w", err)
-	}
-	return services, nil
+	return c.doPodLoad(ctx, host, podName, authToken, strategy, []byte("{}"))
 }
 
 func (c *Client) SavePodSnapshot(ctx context.Context, host, podName, authToken string) (snapshot.PodSaveResult, error) {
-	url := fmt.Sprintf("http://%s/_localstack/pods/%s", host, podName)
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader([]byte("{}")))
-	if err != nil {
-		return snapshot.PodSaveResult{}, fmt.Errorf("create request: %w", err)
-	}
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(":"+authToken)))
-
-	resp, err := c.http.Do(req)
-	if err != nil {
-		return snapshot.PodSaveResult{}, fmt.Errorf("connect to LocalStack: %w", err)
-	}
-	defer func() { _ = resp.Body.Close() }()
-
-	if resp.StatusCode != http.StatusOK {
-		body, _ := io.ReadAll(resp.Body)
-		return snapshot.PodSaveResult{}, fmt.Errorf("pod save failed (HTTP %d): %s", resp.StatusCode, strings.TrimSpace(string(body)))
-	}
-
-	// The response is a newline-delimited JSON stream. We scan until we find a
-	// completion event and surface any server-side error as a Go error.
-	scanner := bufio.NewScanner(resp.Body)
-	buf := make([]byte, 1024*1024)
-	scanner.Buffer(buf, 1024*1024)
-	for scanner.Scan() {
-		line := strings.TrimSpace(scanner.Text())
-		if line == "" {
-			continue
-		}
-		var event struct {
-			Event   string `json:"event"`
-			Status  string `json:"status"`
-			Message string `json:"message"`
-			Info    struct {
-				Version  int      `json:"version"`
-				Services []string `json:"services"`
-				Size     int64    `json:"size"`
-			} `json:"info"`
-		}
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			continue
-		}
-		if event.Event == "completion" {
-			if event.Status != "ok" {
-				return snapshot.PodSaveResult{}, fmt.Errorf("pod save failed: %s", event.Message)
-			}
-			return snapshot.PodSaveResult{
-				Version:  event.Info.Version,
-				Services: event.Info.Services,
-				Size:     event.Info.Size,
-			}, nil
-		}
-	}
-	if err := scanner.Err(); err != nil {
-		return snapshot.PodSaveResult{}, fmt.Errorf("reading response: %w", err)
-	}
-	return snapshot.PodSaveResult{}, fmt.Errorf("pod save: server closed stream without a completion event")
+	return c.doPodSave(ctx, host, podName, authToken, []byte("{}"))
 }
 
 func (c *Client) RemovePodSnapshot(ctx context.Context, host, podName, authToken string) error {
diff --git a/internal/emulator/aws/remote.go b/internal/emulator/aws/remote.go
new file mode 100644
index 00000000..d5a92fee
--- /dev/null
+++ b/internal/emulator/aws/remote.go
@@ -0,0 +1,302 @@
+package aws
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"github.com/localstack/lstk/internal/snapshot"
+)
+
+// remotePayload is the "remote" object sent in pod save/load/list request bodies
+// to target a named remote with ephemeral credential params.
+type remotePayload struct {
+	RemoteName   string            `json:"remote_name"`
+	RemoteParams map[string]string `json:"remote_params,omitempty"`
+}
+
+// podRequestBody is the JSON body for pod save/load/list. Remote is omitted for
+// platform (pod:) operations, in which case the body is "{}".
+type podRequestBody struct {
+	Remote *remotePayload `json:"remote,omitempty"`
+}
+
+// marshalPodBody builds the request body for a pod operation. When remoteName is
+// empty it returns "{}" (the platform default remote).
+func marshalPodBody(remoteName string, params map[string]string) ([]byte, error) {
+	body := podRequestBody{}
+	if remoteName != "" {
+		body.Remote = &remotePayload{RemoteName: remoteName, RemoteParams: params}
+	}
+	return json.Marshal(body)
+}
+
+// setBasicAuth sets the LocalStack Basic auth header when a token is present.
+// S3 remotes do not require a platform token, so it is optional.
+func setBasicAuth(req *http.Request, authToken string) {
+	if authToken == "" {
+		return
+	}
+	req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(":"+authToken)))
+}
+
+// S3BucketExists reports whether an S3 bucket exists, via an unsigned HEAD to the
+// S3 endpoint: a 404 means the bucket does not exist; any other status (200, 403,
+// or a redirect for a bucket in another region) means it does. This lets lstk
+// reject a missing bucket up front instead of letting the emulator auto-create it.
+func (c *Client) S3BucketExists(ctx context.Context, bucket string) (bool, error) {
+	url := fmt.Sprintf(c.s3BucketURLTemplate, bucket)
+	req, err := http.NewRequestWithContext(ctx, http.MethodHead, url, nil)
+	if err != nil {
+		return false, fmt.Errorf("create request: %w", err)
+	}
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return false, fmt.Errorf("connect to S3: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode == http.StatusNotFound {
+		return false, nil
+	}
+	return true, nil
+}
+
+// RegisterRemote upserts a named remote on the running emulator. The emulator
+// persists it (idempotently replacing any same-named entry) so subsequent
+// save/load/list calls can reference it by name.
+func (c *Client) RegisterRemote(ctx context.Context, host, name, remoteURL string) error {
+	url := fmt.Sprintf("http://%s/_localstack/pods/remotes/%s", host, name)
+	payload, err := json.Marshal(map[string]any{
+		"name":       name,
+		"protocols":  []string{"s3"},
+		"remote_url": remoteURL,
+	})
+	if err != nil {
+		return fmt.Errorf("marshal remote request: %w", err)
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload))
+	if err != nil {
+		return fmt.Errorf("create request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return fmt.Errorf("connect to LocalStack: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("register remote failed (HTTP %d): %s", resp.StatusCode, strings.TrimSpace(string(body)))
+	}
+	return nil
+}
+
+// SavePodRemote saves the running state to podName on the named remote.
+func (c *Client) SavePodRemote(ctx context.Context, host, podName, remoteName string, params map[string]string, authToken string) (snapshot.PodSaveResult, error) {
+	body, err := marshalPodBody(remoteName, params)
+	if err != nil {
+		return snapshot.PodSaveResult{}, fmt.Errorf("marshal request: %w", err)
+	}
+	return c.doPodSave(ctx, host, podName, authToken, body)
+}
+
+// LoadPodRemote loads podName from the named remote with the given merge strategy.
+func (c *Client) LoadPodRemote(ctx context.Context, host, podName, remoteName string, params map[string]string, authToken, strategy string) ([]string, error) {
+	body, err := marshalPodBody(remoteName, params)
+	if err != nil {
+		return nil, fmt.Errorf("marshal request: %w", err)
+	}
+	return c.doPodLoad(ctx, host, podName, authToken, strategy, body)
+}
+
+// ListPodsRemote lists the snapshots stored on the named remote via
+// GET /_localstack/pods (with the remote passed in the request body).
+func (c *Client) ListPodsRemote(ctx context.Context, host, remoteName string, params map[string]string, authToken, creator string) ([]snapshot.RemotePod, error) {
+	url := fmt.Sprintf("http://%s/_localstack/pods", host)
+	if creator != "" {
+		url += "?creator=" + creator
+	}
+	body, err := marshalPodBody(remoteName, params)
+	if err != nil {
+		return nil, fmt.Errorf("marshal request: %w", err)
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, bytes.NewReader(body))
+	if err != nil {
+		return nil, fmt.Errorf("create request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	setBasicAuth(req, authToken)
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("connect to LocalStack: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusOK {
+		respBody, _ := io.ReadAll(resp.Body)
+		return nil, fmt.Errorf("list pods failed (HTTP %d): %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
+	}
+
+	var parsed struct {
+		CloudPods []struct {
+			PodName    string `json:"pod_name"`
+			MaxVersion int    `json:"max_version"`
+		} `json:"cloudpods"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
+		return nil, fmt.Errorf("decode response: %w", err)
+	}
+	pods := make([]snapshot.RemotePod, len(parsed.CloudPods))
+	for i, p := range parsed.CloudPods {
+		pods[i] = snapshot.RemotePod{Name: p.PodName, MaxVersion: p.MaxVersion}
+	}
+	return pods, nil
+}
+
+// doPodSave issues POST /_localstack/pods/{name} with the given JSON body and
+// parses the NDJSON response stream into a PodSaveResult.
+func (c *Client) doPodSave(ctx context.Context, host, podName, authToken string, body []byte) (snapshot.PodSaveResult, error) {
+	url := fmt.Sprintf("http://%s/_localstack/pods/%s", host, podName)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
+	if err != nil {
+		return snapshot.PodSaveResult{}, fmt.Errorf("create request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	setBasicAuth(req, authToken)
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return snapshot.PodSaveResult{}, fmt.Errorf("connect to LocalStack: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusOK {
+		respBody, _ := io.ReadAll(resp.Body)
+		return snapshot.PodSaveResult{}, fmt.Errorf("pod save failed (HTTP %d): %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
+	}
+
+	// The response is a newline-delimited JSON stream. We scan until we find a
+	// completion event and surface any server-side error as a Go error.
+	scanner := bufio.NewScanner(resp.Body)
+	buf := make([]byte, 1024*1024)
+	scanner.Buffer(buf, 1024*1024)
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" {
+			continue
+		}
+		var event struct {
+			Event   string `json:"event"`
+			Status  string `json:"status"`
+			Message string `json:"message"`
+			Info    struct {
+				Version  int      `json:"version"`
+				Services []string `json:"services"`
+				Size     int64    `json:"size"`
+			} `json:"info"`
+		}
+		if err := json.Unmarshal([]byte(line), &event); err != nil {
+			continue
+		}
+		if event.Event == "completion" {
+			if event.Status != "ok" {
+				return snapshot.PodSaveResult{}, fmt.Errorf("pod save failed: %s", event.Message)
+			}
+			return snapshot.PodSaveResult{
+				Version:  event.Info.Version,
+				Services: event.Info.Services,
+				Size:     event.Info.Size,
+			}, nil
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return snapshot.PodSaveResult{}, fmt.Errorf("reading response: %w", err)
+	}
+	return snapshot.PodSaveResult{}, fmt.Errorf("pod save: server closed stream without a completion event")
+}
+
+// doPodLoad issues PUT /_localstack/pods/{name}[?merge=strategy] with the given
+// JSON body and parses the NDJSON response stream into the list of services.
+func (c *Client) doPodLoad(ctx context.Context, host, podName, authToken, strategy string, body []byte) ([]string, error) {
+	url := fmt.Sprintf("http://%s/_localstack/pods/%s", host, podName)
+	if strategy != "" {
+		url += "?merge=" + strategy
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader(body))
+	if err != nil {
+		return nil, fmt.Errorf("create request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	setBasicAuth(req, authToken)
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("connect to LocalStack: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode == http.StatusUnprocessableEntity {
+		respBody, _ := io.ReadAll(resp.Body)
+		return nil, fmt.Errorf("%w: %s", snapshot.ErrIncompatibleSnapshot, strings.TrimSpace(string(respBody)))
+	}
+	if resp.StatusCode != http.StatusOK {
+		respBody, _ := io.ReadAll(resp.Body)
+		return nil, fmt.Errorf("pod load failed (HTTP %d): %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
+	}
+
+	var services []string
+	scanner := bufio.NewScanner(resp.Body)
+	buf := make([]byte, 1024*1024)
+	scanner.Buffer(buf, 1024*1024)
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" {
+			continue
+		}
+		var event struct {
+			Event   string `json:"event"`
+			Service string `json:"service"`
+			Status  string `json:"status"`
+			Message string `json:"message"`
+		}
+		if err := json.Unmarshal([]byte(line), &event); err != nil {
+			continue
+		}
+		switch event.Event {
+		case "service":
+			switch event.Status {
+			case "ok":
+				services = append(services, event.Service)
+			case "error":
+				if isInvalidSnapshotFileMsg(event.Message) {
+					return nil, snapshot.ErrInvalidSnapshotFile
+				}
+				return nil, fmt.Errorf("load failed for service %s: %s", event.Service, event.Message)
+			}
+		case "completion":
+			if event.Status != "ok" {
+				if isInvalidSnapshotFileMsg(event.Message) {
+					return nil, snapshot.ErrInvalidSnapshotFile
+				}
+				if isPodNotFoundMsg(event.Message) {
+					return nil, snapshot.ErrPodNotFound
+				}
+				return nil, fmt.Errorf("pod load failed: %s", event.Message)
+			}
+			return services, nil
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return nil, fmt.Errorf("reading response: %w", err)
+	}
+	return services, nil
+}
diff --git a/internal/emulator/aws/remote_test.go b/internal/emulator/aws/remote_test.go
new file mode 100644
index 00000000..2040f7cd
--- /dev/null
+++ b/internal/emulator/aws/remote_test.go
@@ -0,0 +1,115 @@
+package aws
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRegisterRemote(t *testing.T) {
+	t.Parallel()
+	var gotPath string
+	var gotBody map[string]any
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath = r.URL.Path
+		body, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(body, &gotBody)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	c := NewClient()
+	err := c.RegisterRemote(context.Background(), server.Listener.Addr().String(), "lstk-s3-abc", "s3://bucket/?access_key_id={access_key_id}")
+	require.NoError(t, err)
+	assert.Equal(t, "/_localstack/pods/remotes/lstk-s3-abc", gotPath)
+	assert.Equal(t, []any{"s3"}, gotBody["protocols"])
+	assert.Equal(t, "s3://bucket/?access_key_id={access_key_id}", gotBody["remote_url"])
+}
+
+func TestSavePodRemote_SendsRemoteBody(t *testing.T) {
+	t.Parallel()
+	var gotBody podRequestBody
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		require.NoError(t, json.Unmarshal(body, &gotBody))
+		w.Header().Set("Content-Type", "application/x-ndjson")
+		_, _ = fmt.Fprintln(w, `{"event": "completion", "status": "ok", "info": {"version": 1, "services": ["s3"], "size": 10}}`)
+	}))
+	defer server.Close()
+
+	c := NewClient()
+	params := map[string]string{"access_key_id": "AKIA", "secret_access_key": "shh"}
+	res, err := c.SavePodRemote(context.Background(), server.Listener.Addr().String(), "my-pod", "lstk-s3-abc", params, "")
+	require.NoError(t, err)
+	assert.Equal(t, 1, res.Version)
+	require.NotNil(t, gotBody.Remote)
+	assert.Equal(t, "lstk-s3-abc", gotBody.Remote.RemoteName)
+	assert.Equal(t, "AKIA", gotBody.Remote.RemoteParams["access_key_id"])
+	assert.Equal(t, "shh", gotBody.Remote.RemoteParams["secret_access_key"])
+}
+
+func TestSavePodSnapshot_SendsEmptyRemote(t *testing.T) {
+	t.Parallel()
+	var gotBody podRequestBody
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		require.NoError(t, json.Unmarshal(body, &gotBody))
+		assert.Equal(t, "Basic OnRoZS10b2tlbg==", r.Header.Get("Authorization")) // base64(":the-token")
+		w.Header().Set("Content-Type", "application/x-ndjson")
+		_, _ = fmt.Fprintln(w, `{"event": "completion", "status": "ok", "info": {"version": 2}}`)
+	}))
+	defer server.Close()
+
+	c := NewClient()
+	_, err := c.SavePodSnapshot(context.Background(), server.Listener.Addr().String(), "my-pod", "the-token")
+	require.NoError(t, err)
+	assert.Nil(t, gotBody.Remote, "platform pod save must not include a remote payload")
+}
+
+func TestS3BucketExists(t *testing.T) {
+	t.Parallel()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// 404 for a missing bucket; 403 (access denied, unsigned) means it exists.
+		if r.URL.Path == "/missing" {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		w.WriteHeader(http.StatusForbidden)
+	}))
+	defer server.Close()
+
+	c := NewClient()
+	c.s3BucketURLTemplate = server.URL + "/%s"
+
+	exists, err := c.S3BucketExists(context.Background(), "exists")
+	require.NoError(t, err)
+	assert.True(t, exists)
+
+	missing, err := c.S3BucketExists(context.Background(), "missing")
+	require.NoError(t, err)
+	assert.False(t, missing)
+}
+
+func TestListPodsRemote(t *testing.T) {
+	t.Parallel()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "/_localstack/pods", r.URL.Path)
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = fmt.Fprintln(w, `{"cloudpods": [{"pod_name": "a", "max_version": 3}, {"pod_name": "b", "max_version": 1}]}`)
+	}))
+	defer server.Close()
+
+	c := NewClient()
+	pods, err := c.ListPodsRemote(context.Background(), server.Listener.Addr().String(), "lstk-s3-abc", map[string]string{"access_key_id": "x"}, "", "")
+	require.NoError(t, err)
+	require.Len(t, pods, 2)
+	assert.Equal(t, "a", pods[0].Name)
+	assert.Equal(t, 3, pods[0].MaxVersion)
+}
diff --git a/internal/output/events.go b/internal/output/events.go
index c4455f93..42b9c71b 100644
--- a/internal/output/events.go
+++ b/internal/output/events.go
@@ -84,6 +84,17 @@ type PodSnapshotSavedEvent struct {
 	Size     int64
 }
 
+// RemoteSnapshotSavedEvent reports a snapshot saved to a remote storage backend
+// (e.g. an S3 bucket). Location is the user-facing remote target (e.g. an s3:// URL)
+// and PodName is the snapshot's identity within that remote.
+type RemoteSnapshotSavedEvent struct {
+	PodName  string
+	Location string
+	Version  int
+	Services []string
+	Size     int64
+}
+
 // DeferredEvent wraps another event so that the TUI renders it after the interface
 // exits rather than inline. Plain sinks format the inner event immediately.
 type DeferredEvent struct {
@@ -130,23 +141,24 @@ type AuthCompleteEvent struct{}
 // so Sink.Emit rejects unknown types at compile time.
 type Event interface{ sealedEvent() }
 
-func (MessageEvent) sealedEvent()            {}
-func (SpinnerEvent) sealedEvent()            {}
-func (ErrorEvent) sealedEvent()              {}
-func (AuthEvent) sealedEvent()               {}
-func (AuthCompleteEvent) sealedEvent()       {}
-func (InstanceInfoEvent) sealedEvent()       {}
-func (TableEvent) sealedEvent()              {}
-func (ResourceSummaryEvent) sealedEvent()    {}
-func (PodSnapshotSavedEvent) sealedEvent()   {}
-func (DeferredEvent) sealedEvent()           {}
-func (SnapshotLoadedEvent) sealedEvent()     {}
-func (PodSnapshotRemovedEvent) sealedEvent() {}
-func (SnapshotShownEvent) sealedEvent()      {}
-func (ContainerStatusEvent) sealedEvent()    {}
-func (ProgressEvent) sealedEvent()           {}
-func (UserInputRequestEvent) sealedEvent()   {}
-func (LogLineEvent) sealedEvent()            {}
+func (MessageEvent) sealedEvent()             {}
+func (SpinnerEvent) sealedEvent()             {}
+func (ErrorEvent) sealedEvent()               {}
+func (AuthEvent) sealedEvent()                {}
+func (AuthCompleteEvent) sealedEvent()        {}
+func (InstanceInfoEvent) sealedEvent()        {}
+func (TableEvent) sealedEvent()               {}
+func (ResourceSummaryEvent) sealedEvent()     {}
+func (PodSnapshotSavedEvent) sealedEvent()    {}
+func (RemoteSnapshotSavedEvent) sealedEvent() {}
+func (DeferredEvent) sealedEvent()            {}
+func (SnapshotLoadedEvent) sealedEvent()      {}
+func (PodSnapshotRemovedEvent) sealedEvent()  {}
+func (SnapshotShownEvent) sealedEvent()       {}
+func (ContainerStatusEvent) sealedEvent()     {}
+func (ProgressEvent) sealedEvent()            {}
+func (UserInputRequestEvent) sealedEvent()    {}
+func (LogLineEvent) sealedEvent()             {}
 
 type Sink interface {
 	Emit(event Event)
diff --git a/internal/output/plain_format.go b/internal/output/plain_format.go
index 1aff3adb..e11b8385 100644
--- a/internal/output/plain_format.go
+++ b/internal/output/plain_format.go
@@ -42,6 +42,8 @@ func FormatEventLine(event Event) (string, bool) {
 		return formatResourceSummary(e), true
 	case PodSnapshotSavedEvent:
 		return formatPodSnapshotSaved(e), true
+	case RemoteSnapshotSavedEvent:
+		return formatRemoteSnapshotSaved(e), true
 	case SnapshotLoadedEvent:
 		return formatSnapshotLoaded(e), true
 	case DeferredEvent:
@@ -230,6 +232,21 @@ func formatPodSnapshotSaved(e PodSnapshotSavedEvent) string {
 	return sb.String()
 }
 
+func formatRemoteSnapshotSaved(e RemoteSnapshotSavedEvent) string {
+	var sb strings.Builder
+	sb.WriteString(SuccessMarker() + fmt.Sprintf(" Snapshot saved to %s as %q", e.Location, e.PodName))
+	if e.Version > 0 {
+		sb.WriteString(fmt.Sprintf("\n• Version: %d", e.Version))
+	}
+	if len(e.Services) > 0 {
+		sb.WriteString("\n• Services: " + strings.Join(e.Services, ", "))
+	}
+	if e.Size > 0 {
+		sb.WriteString("\n• Size: " + formatBytes(e.Size))
+	}
+	return sb.String()
+}
+
 func formatPodSnapshotRemoved(e PodSnapshotRemovedEvent) string {
 	return SuccessMarker() + fmt.Sprintf(" Cloud snapshot 'pod:%s' deleted", e.PodName)
 }
diff --git a/internal/output/plain_format_test.go b/internal/output/plain_format_test.go
index fdd49e88..4f1c9b62 100644
--- a/internal/output/plain_format_test.go
+++ b/internal/output/plain_format_test.go
@@ -211,6 +211,27 @@ func TestFormatEventLine(t *testing.T) {
 			want:   SuccessMarker() + " Snapshot saved to pod:minimal-pod",
 			wantOK: true,
 		},
+		{
+			name: "remote snapshot saved full",
+			event: RemoteSnapshotSavedEvent{
+				PodName:  "my-baseline",
+				Location: "s3://my-bucket/prefix",
+				Version:  3,
+				Services: []string{"dynamodb", "s3", "sqs"},
+				Size:     2621440,
+			},
+			want:   SuccessMarker() + " Snapshot saved to s3://my-bucket/prefix as \"my-baseline\"\n• Version: 3\n• Services: dynamodb, s3, sqs\n• Size: 2.5 MB",
+			wantOK: true,
+		},
+		{
+			name: "remote snapshot saved omits zero fields",
+			event: RemoteSnapshotSavedEvent{
+				PodName:  "minimal-pod",
+				Location: "s3://my-bucket",
+			},
+			want:   SuccessMarker() + " Snapshot saved to s3://my-bucket as \"minimal-pod\"",
+			wantOK: true,
+		},
 
 		// snapshot load events
 		{
diff --git a/internal/snapshot/destination.go b/internal/snapshot/destination.go
index 798c1dc9..55a4ae45 100644
--- a/internal/snapshot/destination.go
+++ b/internal/snapshot/destination.go
@@ -4,6 +4,7 @@ import (
 	"crypto/rand"
 	"errors"
 	"fmt"
+	"net/url"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -15,10 +16,14 @@ import (
 var ErrHomeNotSet = errors.New("home directory is not set")
 
 var (
-	// ErrRemoteNotSupported is returned for known remote schemes (s3://, oras://).
+	// ErrRemoteNotSupported is returned for remote schemes that are not yet
+	// implemented (e.g. oras://). S3 (s3://) is supported.
 	ErrRemoteNotSupported = errors.New("remote destinations are not yet supported — coming soon")
 	// ErrUnknownScheme is returned for unrecognized URL schemes.
 	ErrUnknownScheme = errors.New("unrecognized destination scheme")
+	// ErrCredentialsInS3URL is returned when an s3:// ref embeds credential query
+	// params. Credentials must come from the environment or --profile, never the URL.
+	ErrCredentialsInS3URL = errors.New("do not put credentials in the s3:// URL; use AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY or --profile")
 
 	validPodName = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9_-]*$`)
 )
@@ -43,16 +48,61 @@ type DestinationKind int
 const (
 	KindLocal DestinationKind = iota
 	KindPod
+	KindS3
 )
 
 // Destination is the parsed result of a user-supplied snapshot destination.
 // For KindLocal, Value is an absolute local file path with a .snapshot extension.
 // For KindPod, Value is the validated pod name (without the "pod:" prefix).
+// For KindS3, Value is the validated s3:// URL (bucket + optional key prefix), with
+// no credential query params — credentials are supplied separately at runtime.
 type Destination struct {
 	Kind  DestinationKind
 	Value string
 }
 
+// IsS3Ref reports whether ref is an s3:// reference. Used at the command boundary
+// to classify positional args into a pod name and an S3 location.
+func IsS3Ref(ref string) bool {
+	return strings.HasPrefix(strings.ToLower(ref), "s3://")
+}
+
+// ValidatePodName validates a user-supplied pod name (the identity of a snapshot
+// on a remote), using the same rules as pod: refs.
+func ValidatePodName(name string) error {
+	if !validPodName.MatchString(name) {
+		return fmt.Errorf("invalid pod name %q: use letters, digits, hyphens, and underscores only, starting with a letter or digit", name)
+	}
+	return nil
+}
+
+// DefaultRemotePodName generates a timestamped pod name used when saving to a
+// remote without an explicit name, mirroring local snapshot auto-naming.
+func DefaultRemotePodName(now time.Time) string {
+	b := make([]byte, 2)
+	_, _ = rand.Read(b)
+	return "snapshot-" + now.UTC().Format("2006-01-02T15-04-05") + "-" + fmt.Sprintf("%x", b)[:3]
+}
+
+// parseS3 validates an s3:// URL and returns it as a KindS3 destination. The bucket
+// must be present and the URL must not contain credential query params.
+func parseS3(ref string) (Destination, error) {
+	u, err := url.Parse(ref)
+	if err != nil {
+		return Destination{}, fmt.Errorf("invalid s3:// URL %q: %w", ref, err)
+	}
+	if u.Host == "" {
+		return Destination{}, fmt.Errorf("invalid s3:// URL %q: missing bucket name", ref)
+	}
+	q := u.Query()
+	for _, k := range []string{"access_key_id", "secret_access_key", "session_token"} {
+		if q.Has(k) {
+			return Destination{}, ErrCredentialsInS3URL
+		}
+	}
+	return Destination{Kind: KindS3, Value: ref}, nil
+}
+
 // ParseRemovable parses a ref for snapshot remove. Only cloud (pod:) refs are accepted;
 // local file paths are rejected because the CLI cannot delete local files.
 // cwd and home are used to produce a human-readable path in error messages.
@@ -76,7 +126,15 @@ func parseCloudOnly(ref, cwd, home, action string) (Destination, error) {
 		abs = withSnapshotExt(abs)
 		return Destination{}, fmt.Errorf("'%s' resolves to a local file (%s); CLI cannot %s", ref, displayPath(abs, cwd, home), action)
 	}
-	return ParseSource(ref, home)
+	dest, err := ParseSource(ref, home)
+	if err != nil {
+		return Destination{}, err
+	}
+	// remove/show are cloud (pod:) only; S3 remotes are not yet supported here.
+	if dest.Kind == KindS3 {
+		return Destination{}, ErrRemoteNotSupported
+	}
+	return dest, nil
 }
 
 // displayPath shortens abs for human-readable output:
@@ -112,12 +170,13 @@ func ParseSource(ref, home string) (Destination, error) {
 		return Destination{}, fmt.Errorf("'%s' is not a valid reference. Aliases use a single colon. Did you mean:\npod:%s", ref, podName)
 	case strings.HasPrefix(lower, "pod:"):
 		podName := ref[len("pod:"):]
-		if !validPodName.MatchString(podName) {
-			return Destination{}, fmt.Errorf("invalid pod name %q: use letters, digits, hyphens, and underscores only, starting with a letter or digit", podName)
+		if err := ValidatePodName(podName); err != nil {
+			return Destination{}, err
 		}
 		return Destination{Kind: KindPod, Value: podName}, nil
-	case strings.HasPrefix(lower, "s3://"),
-		strings.HasPrefix(lower, "oras://"):
+	case strings.HasPrefix(lower, "s3://"):
+		return parseS3(ref)
+	case strings.HasPrefix(lower, "oras://"):
 		return Destination{}, ErrRemoteNotSupported
 	case strings.Contains(lower, "://"):
 		scheme, _, _ := strings.Cut(ref, "://")
@@ -174,12 +233,13 @@ func ParseDestination(dest, home string, now time.Time) (Destination, error) {
 			return Destination{}, fmt.Errorf("'%s' is not a valid reference. Aliases use a single colon. Did you mean:\npod:%s", dest, podName)
 		case strings.HasPrefix(lower, "pod:"):
 			podName := dest[len("pod:"):]
-			if !validPodName.MatchString(podName) {
-				return Destination{}, fmt.Errorf("invalid pod name %q: use letters, digits, hyphens, and underscores only, starting with a letter or digit", podName)
+			if err := ValidatePodName(podName); err != nil {
+				return Destination{}, err
 			}
 			return Destination{Kind: KindPod, Value: podName}, nil
-		case strings.HasPrefix(lower, "s3://"),
-			strings.HasPrefix(lower, "oras://"):
+		case strings.HasPrefix(lower, "s3://"):
+			return parseS3(dest)
+		case strings.HasPrefix(lower, "oras://"):
 			return Destination{}, ErrRemoteNotSupported
 		case strings.Contains(lower, "://"):
 			scheme, _, _ := strings.Cut(dest, "://")
diff --git a/internal/snapshot/destination_test.go b/internal/snapshot/destination_test.go
index 5777579e..bd563fe0 100644
--- a/internal/snapshot/destination_test.go
+++ b/internal/snapshot/destination_test.go
@@ -171,9 +171,20 @@ func TestParseSource(t *testing.T) {
 
 		// --- remote schemes ---
 		{
-			name:          "s3:// not supported",
-			input:         "s3://bucket/key",
-			wantRemoteErr: true,
+			name:     "s3:// is an S3 remote",
+			input:    "s3://bucket/key",
+			wantKind: snapshot.KindS3,
+			wantPath: "s3://bucket/key",
+		},
+		{
+			name:    "s3:// rejects embedded credentials",
+			input:   "s3://bucket/key?access_key_id=AKIA&secret_access_key=zzz",
+			wantErr: "do not put credentials",
+		},
+		{
+			name:    "s3:// requires a bucket",
+			input:   "s3:///key",
+			wantErr: "missing bucket",
 		},
 		{
 			name:          "oras:// not supported",
@@ -432,12 +443,19 @@ func TestParseDestination(t *testing.T) {
 
 		// --- remote: s3 ---
 		{
-			input:         "s3://bucket/key",
-			wantRemoteErr: true,
+			input:    "s3://bucket/key",
+			wantKind: snapshot.KindS3,
+			wantPath: "s3://bucket/key",
 		},
 		{
-			input:         "S3://bucket/key",
-			wantRemoteErr: true,
+			input:    "S3://bucket/key",
+			wantKind: snapshot.KindS3,
+			wantPath: "S3://bucket/key",
+		},
+		{
+			name:    "s3:// rejects embedded credentials",
+			input:   "s3://bucket/key?secret_access_key=zzz",
+			wantErr: "do not put credentials",
 		},
 
 		// --- remote: oras ---
@@ -583,3 +601,16 @@ func TestParseDestinationTildeWithoutHome(t *testing.T) {
 		})
 	}
 }
+
+func TestDefaultRemotePodName(t *testing.T) {
+	t.Parallel()
+
+	now := time.Date(2026, 5, 11, 21, 4, 32, 0, time.UTC)
+	name := snapshot.DefaultRemotePodName(now)
+
+	assert.True(t, strings.HasPrefix(name, "snapshot-2026-05-11T21-04-32-"), "got %q", name)
+	// The generated name must be a valid pod name.
+	require.NoError(t, snapshot.ValidatePodName(name))
+	// The random suffix should make repeated calls distinct.
+	assert.NotEqual(t, name, snapshot.DefaultRemotePodName(now))
+}
diff --git a/internal/snapshot/export_test.go b/internal/snapshot/export_test.go
index 5959cd60..11d22bb1 100644
--- a/internal/snapshot/export_test.go
+++ b/internal/snapshot/export_test.go
@@ -1,3 +1,8 @@
 package snapshot
 
 var DisplayPath = displayPath
+
+var (
+	TemplatedRemoteURL = templatedRemoteURL
+	RemoteName         = remoteName
+)
diff --git a/internal/snapshot/mock_remote_client_test.go b/internal/snapshot/mock_remote_client_test.go
new file mode 100644
index 00000000..12f2c3ab
--- /dev/null
+++ b/internal/snapshot/mock_remote_client_test.go
@@ -0,0 +1,116 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: remote.go
+//
+// Generated by this command:
+//
+//	mockgen -source=remote.go -destination=mock_remote_client_test.go -package=snapshot_test
+//
+
+// Package snapshot_test is a generated GoMock package.
+package snapshot_test
+
+import (
+	context "context"
+	reflect "reflect"
+
+	snapshot "github.com/localstack/lstk/internal/snapshot"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockRemoteClient is a mock of RemoteClient interface.
+type MockRemoteClient struct {
+	ctrl     *gomock.Controller
+	recorder *MockRemoteClientMockRecorder
+	isgomock struct{}
+}
+
+// MockRemoteClientMockRecorder is the mock recorder for MockRemoteClient.
+type MockRemoteClientMockRecorder struct {
+	mock *MockRemoteClient
+}
+
+// NewMockRemoteClient creates a new mock instance.
+func NewMockRemoteClient(ctrl *gomock.Controller) *MockRemoteClient {
+	mock := &MockRemoteClient{ctrl: ctrl}
+	mock.recorder = &MockRemoteClientMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockRemoteClient) EXPECT() *MockRemoteClientMockRecorder {
+	return m.recorder
+}
+
+// ListPodsRemote mocks base method.
+func (m *MockRemoteClient) ListPodsRemote(ctx context.Context, host, remoteName string, params map[string]string, authToken, creator string) ([]snapshot.RemotePod, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ListPodsRemote", ctx, host, remoteName, params, authToken, creator)
+	ret0, _ := ret[0].([]snapshot.RemotePod)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ListPodsRemote indicates an expected call of ListPodsRemote.
+func (mr *MockRemoteClientMockRecorder) ListPodsRemote(ctx, host, remoteName, params, authToken, creator any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListPodsRemote", reflect.TypeOf((*MockRemoteClient)(nil).ListPodsRemote), ctx, host, remoteName, params, authToken, creator)
+}
+
+// LoadPodRemote mocks base method.
+func (m *MockRemoteClient) LoadPodRemote(ctx context.Context, host, podName, remoteName string, params map[string]string, authToken, strategy string) ([]string, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "LoadPodRemote", ctx, host, podName, remoteName, params, authToken, strategy)
+	ret0, _ := ret[0].([]string)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// LoadPodRemote indicates an expected call of LoadPodRemote.
+func (mr *MockRemoteClientMockRecorder) LoadPodRemote(ctx, host, podName, remoteName, params, authToken, strategy any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "LoadPodRemote", reflect.TypeOf((*MockRemoteClient)(nil).LoadPodRemote), ctx, host, podName, remoteName, params, authToken, strategy)
+}
+
+// RegisterRemote mocks base method.
+func (m *MockRemoteClient) RegisterRemote(ctx context.Context, host, name, remoteURL string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RegisterRemote", ctx, host, name, remoteURL)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RegisterRemote indicates an expected call of RegisterRemote.
+func (mr *MockRemoteClientMockRecorder) RegisterRemote(ctx, host, name, remoteURL any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RegisterRemote", reflect.TypeOf((*MockRemoteClient)(nil).RegisterRemote), ctx, host, name, remoteURL)
+}
+
+// S3BucketExists mocks base method.
+func (m *MockRemoteClient) S3BucketExists(ctx context.Context, bucket string) (bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "S3BucketExists", ctx, bucket)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// S3BucketExists indicates an expected call of S3BucketExists.
+func (mr *MockRemoteClientMockRecorder) S3BucketExists(ctx, bucket any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "S3BucketExists", reflect.TypeOf((*MockRemoteClient)(nil).S3BucketExists), ctx, bucket)
+}
+
+// SavePodRemote mocks base method.
+func (m *MockRemoteClient) SavePodRemote(ctx context.Context, host, podName, remoteName string, params map[string]string, authToken string) (snapshot.PodSaveResult, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SavePodRemote", ctx, host, podName, remoteName, params, authToken)
+	ret0, _ := ret[0].(snapshot.PodSaveResult)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// SavePodRemote indicates an expected call of SavePodRemote.
+func (mr *MockRemoteClientMockRecorder) SavePodRemote(ctx, host, podName, remoteName, params, authToken any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SavePodRemote", reflect.TypeOf((*MockRemoteClient)(nil).SavePodRemote), ctx, host, podName, remoteName, params, authToken)
+}
diff --git a/internal/snapshot/remote.go b/internal/snapshot/remote.go
new file mode 100644
index 00000000..e3642d8e
--- /dev/null
+++ b/internal/snapshot/remote.go
@@ -0,0 +1,239 @@
+package snapshot
+
+//go:generate mockgen -source=remote.go -destination=mock_remote_client_test.go -package=snapshot_test
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"net"
+	"net/url"
+	"strings"
+
+	"github.com/localstack/lstk/internal/config"
+	"github.com/localstack/lstk/internal/output"
+	"github.com/localstack/lstk/internal/runtime"
+)
+
+// Credential param names rendered into the remote URL template by the emulator.
+// These must match the backend's S3 remote contract.
+const (
+	paramAccessKeyID     = "access_key_id"
+	paramSecretAccessKey = "secret_access_key"
+	paramSessionToken    = "session_token"
+)
+
+// S3Credentials are the AWS credentials sent to the emulator for an S3 remote.
+// They are passed as ephemeral per-request parameters and never persisted.
+type S3Credentials struct {
+	AccessKeyID     string
+	SecretAccessKey string
+	SessionToken    string
+}
+
+// params builds the remote_params map for the request body, omitting the session
+// token when absent.
+func (c S3Credentials) params() map[string]string {
+	p := map[string]string{
+		paramAccessKeyID:     c.AccessKeyID,
+		paramSecretAccessKey: c.SecretAccessKey,
+	}
+	if c.SessionToken != "" {
+		p[paramSessionToken] = c.SessionToken
+	}
+	return p
+}
+
+// RemotePod is a snapshot listed on a remote storage backend.
+type RemotePod struct {
+	Name       string
+	MaxVersion int
+}
+
+// RemoteClient is satisfied by aws.Client. It manages remote registration and the
+// pod operations that target a named remote.
+type RemoteClient interface {
+	// S3BucketExists reports whether the named S3 bucket exists. Used to reject a
+	// missing bucket before an operation, rather than letting the emulator
+	// auto-create it.
+	S3BucketExists(ctx context.Context, bucket string) (bool, error)
+	// RegisterRemote upserts a named remote on the running emulator
+	// (POST /_localstack/pods/remotes/<name>). remoteURL may contain {placeholder}
+	// tokens that the emulator renders with the per-request params.
+	RegisterRemote(ctx context.Context, host, name, remoteURL string) error
+	// SavePodRemote saves the running state to podName on the named remote.
+	SavePodRemote(ctx context.Context, host, podName, remoteName string, params map[string]string, authToken string) (PodSaveResult, error)
+	// LoadPodRemote loads podName from the named remote with the given merge strategy.
+	LoadPodRemote(ctx context.Context, host, podName, remoteName string, params map[string]string, authToken, strategy string) ([]string, error)
+	// ListPodsRemote lists the snapshots stored on the named remote.
+	ListPodsRemote(ctx context.Context, host, remoteName string, params map[string]string, authToken, creator string) ([]RemotePod, error)
+}
+
+// s3RemoteBucket extracts the bucket name from an s3:// URL and reports whether
+// the URL targets a local-testing endpoint (an IP or host.docker.internal), which
+// mirrors the emulator's own detection. For local endpoints the bucket is the
+// first path segment; for real AWS it is the host.
+func s3RemoteBucket(s3URL string) (bucket string, isLocalEndpoint bool) {
+	u, err := url.Parse(s3URL)
+	if err != nil {
+		return "", false
+	}
+	host := u.Hostname()
+	if host == "host.docker.internal" || net.ParseIP(host) != nil {
+		return strings.Trim(u.Path, "/"), true
+	}
+	return host, false
+}
+
+// ensureBucketExists returns an error when s3URL points at a real-AWS bucket that
+// does not exist, so lstk never relies on the emulator auto-creating it. Local
+// testing endpoints are skipped, and a check that cannot be performed (e.g. no
+// network) is surfaced as a warning rather than a hard failure.
+func ensureBucketExists(ctx context.Context, client RemoteClient, s3URL string, sink output.Sink) error {
+	bucket, isLocal := s3RemoteBucket(s3URL)
+	if isLocal || bucket == "" {
+		return nil
+	}
+	exists, err := client.S3BucketExists(ctx, bucket)
+	if err != nil {
+		sink.Emit(output.MessageEvent{Severity: output.SeverityWarning, Text: fmt.Sprintf("could not verify S3 bucket %q exists: %v", bucket, err)})
+		return nil
+	}
+	if !exists {
+		return fmt.Errorf("S3 bucket %q does not exist — create it first; lstk does not create buckets automatically", bucket)
+	}
+	return nil
+}
+
+// remoteName derives a deterministic registry name for an s3:// URL so the
+// emulator's remote registry holds at most one entry per bucket/prefix and
+// re-registration is an idempotent overwrite.
+func remoteName(s3URL string) string {
+	sum := sha256.Sum256([]byte(s3URL))
+	return "lstk-s3-" + hex.EncodeToString(sum[:])[:10]
+}
+
+// templatedRemoteURL appends credential placeholders to the s3:// URL so the
+// emulator injects the ephemeral params at runtime. Secrets never appear here —
+// only "{access_key_id}"-style tokens, kept literal (not percent-encoded) so the
+// backend's str.format rendering recognizes them.
+func templatedRemoteURL(s3URL string, hasSessionToken bool) string {
+	template := paramAccessKeyID + "={" + paramAccessKeyID + "}" +
+		"&" + paramSecretAccessKey + "={" + paramSecretAccessKey + "}"
+	if hasSessionToken {
+		template += "&" + paramSessionToken + "={" + paramSessionToken + "}"
+	}
+	sep := "?"
+	if strings.Contains(s3URL, "?") {
+		sep = "&"
+	}
+	return s3URL + sep + template
+}
+
+// SaveRemoteS3 saves the running emulator's state to podName in the S3 bucket
+// identified by s3URL, using the given credentials. An auth token is optional for
+// S3 remotes (the S3 credentials are the auth); it is forwarded when present.
+func SaveRemoteS3(ctx context.Context, rt runtime.Runtime, containers []config.ContainerConfig, client RemoteClient, host, podName, s3URL string, creds S3Credentials, authToken string, sink output.Sink) error {
+	if err := ensureBucketExists(ctx, client, s3URL, sink); err != nil {
+		return err
+	}
+	name := remoteName(s3URL)
+	remoteURL := templatedRemoteURL(s3URL, creds.SessionToken != "")
+	var result PodSaveResult
+	return save(ctx, rt, containers, sink,
+		fmt.Sprintf("Saving snapshot to %s...", s3URL),
+		func() {
+			sink.Emit(output.RemoteSnapshotSavedEvent{
+				PodName:  podName,
+				Location: s3URL,
+				Version:  result.Version,
+				Services: result.Services,
+				Size:     result.Size,
+			})
+		},
+		func() error {
+			if err := client.RegisterRemote(ctx, host, name, remoteURL); err != nil {
+				return fmt.Errorf("register S3 remote: %w", err)
+			}
+			var err error
+			result, err = client.SavePodRemote(ctx, host, podName, name, creds.params(), authToken)
+			return err
+		},
+	)
+}
+
+// LoadRemoteS3 loads podName from the S3 bucket identified by s3URL into the
+// running emulator, starting it first if needed.
+func LoadRemoteS3(ctx context.Context, rt runtime.Runtime, containers []config.ContainerConfig, client RemoteClient, host, podName, s3URL string, creds S3Credentials, authToken, strategy string, starter Starter, sink output.Sink) error {
+	if err := ensureBucketExists(ctx, client, s3URL, sink); err != nil {
+		return err
+	}
+	name := remoteName(s3URL)
+	remoteURL := templatedRemoteURL(s3URL, creds.SessionToken != "")
+	var services []string
+	return load(ctx, rt, containers, sink, starter,
+		fmt.Sprintf("Loading snapshot %q from %s...", podName, s3URL),
+		func() {
+			sink.Emit(output.SnapshotLoadedEvent{
+				Source:   fmt.Sprintf("%s (%s)", s3URL, podName),
+				Services: services,
+			})
+		},
+		func() error {
+			if err := client.RegisterRemote(ctx, host, name, remoteURL); err != nil {
+				return fmt.Errorf("register S3 remote: %w", err)
+			}
+			var err error
+			services, err = client.LoadPodRemote(ctx, host, podName, name, creds.params(), authToken, strategy)
+			return err
+		},
+	)
+}
+
+// ListRemoteS3 lists the snapshots stored in the S3 bucket identified by s3URL.
+// Unlike List (which queries the platform API), this requires a running emulator
+// because the emulator performs the S3 listing.
+func ListRemoteS3(ctx context.Context, rt runtime.Runtime, containers []config.ContainerConfig, client RemoteClient, host, s3URL string, creds S3Credentials, authToken string, sink output.Sink) error {
+	if err := rt.IsHealthy(ctx); err != nil {
+		rt.EmitUnhealthyError(sink, err)
+		return output.NewSilentError(fmt.Errorf("runtime not healthy: %w", err))
+	}
+
+	if err := ensureBucketExists(ctx, client, s3URL, sink); err != nil {
+		return err
+	}
+
+	name := remoteName(s3URL)
+	remoteURL := templatedRemoteURL(s3URL, creds.SessionToken != "")
+
+	sink.Emit(output.SpinnerStart("Fetching snapshots"))
+	if err := client.RegisterRemote(ctx, host, name, remoteURL); err != nil {
+		sink.Emit(output.SpinnerStop())
+		return fmt.Errorf("register S3 remote: %w", err)
+	}
+	pods, err := client.ListPodsRemote(ctx, host, name, creds.params(), authToken, "")
+	sink.Emit(output.SpinnerStop())
+	if err != nil {
+		return fmt.Errorf("list snapshots on %s: %w", s3URL, err)
+	}
+
+	if len(pods) == 0 {
+		sink.Emit(output.DeferredEvent{Inner: output.MessageEvent{Severity: output.SeverityNote, Text: fmt.Sprintf("No snapshots found on %s", s3URL)}})
+		return nil
+	}
+	noun := "snapshots"
+	if len(pods) == 1 {
+		noun = "snapshot"
+	}
+	rows := make([][]string, len(pods))
+	for i, p := range pods {
+		rows[i] = []string{p.Name, fmt.Sprintf("%d", p.MaxVersion)}
+	}
+	sink.Emit(output.DeferredEvent{Inner: output.MessageEvent{Severity: output.SeveritySecondary, Text: fmt.Sprintf("~ %d %s\n", len(pods), noun)}})
+	sink.Emit(output.DeferredEvent{Inner: output.TableEvent{
+		Headers: []string{"Name", "Version"},
+		Rows:    rows,
+	}})
+	return nil
+}
diff --git a/internal/snapshot/remote_test.go b/internal/snapshot/remote_test.go
new file mode 100644
index 00000000..836e267b
--- /dev/null
+++ b/internal/snapshot/remote_test.go
@@ -0,0 +1,209 @@
+package snapshot_test
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/localstack/lstk/internal/output"
+	"github.com/localstack/lstk/internal/runtime"
+	"github.com/localstack/lstk/internal/snapshot"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+)
+
+func TestTemplatedRemoteURL(t *testing.T) {
+	t.Parallel()
+
+	// Placeholders must stay as literal {tokens} (not percent-encoded) so the
+	// backend's str.format rendering recognizes them.
+	got := snapshot.TemplatedRemoteURL("s3://my-bucket/prefix", false)
+	assert.Equal(t, "s3://my-bucket/prefix?access_key_id={access_key_id}&secret_access_key={secret_access_key}", got)
+
+	withToken := snapshot.TemplatedRemoteURL("s3://my-bucket/prefix", true)
+	assert.Contains(t, withToken, "session_token={session_token}")
+
+	// An existing query is preserved with an & separator.
+	withQuery := snapshot.TemplatedRemoteURL("s3://my-bucket/prefix?region=eu-west-1", false)
+	assert.True(t, strings.HasPrefix(withQuery, "s3://my-bucket/prefix?region=eu-west-1&access_key_id="))
+}
+
+func TestRemoteName_DeterministicAndDistinct(t *testing.T) {
+	t.Parallel()
+	a := snapshot.RemoteName("s3://bucket/one")
+	b := snapshot.RemoteName("s3://bucket/one")
+	c := snapshot.RemoteName("s3://bucket/two")
+	assert.Equal(t, a, b, "same URL yields the same remote name")
+	assert.NotEqual(t, a, c, "different URLs yield different remote names")
+	assert.True(t, strings.HasPrefix(a, "lstk-s3-"))
+}
+
+func TestSaveRemoteS3_RegistersAndSaves(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	client := NewMockRemoteClient(ctrl)
+
+	const s3URL = "s3://my-bucket/prefix"
+	wantName := snapshot.RemoteName(s3URL)
+
+	client.EXPECT().S3BucketExists(gomock.Any(), "my-bucket").Return(true, nil)
+	var gotURL string
+	var gotParams map[string]string
+	client.EXPECT().RegisterRemote(gomock.Any(), gomock.Any(), wantName, gomock.Any()).DoAndReturn(
+		func(_ context.Context, _, _, remoteURL string) error {
+			gotURL = remoteURL
+			return nil
+		},
+	)
+	client.EXPECT().SavePodRemote(gomock.Any(), gomock.Any(), "my-pod", wantName, gomock.Any(), "").DoAndReturn(
+		func(_ context.Context, _, _, _ string, params map[string]string, _ string) (snapshot.PodSaveResult, error) {
+			gotParams = params
+			return snapshot.PodSaveResult{Version: 1, Services: []string{"s3"}, Size: 42}, nil
+		},
+	)
+
+	creds := snapshot.S3Credentials{AccessKeyID: "AKIA123", SecretAccessKey: "supersecret"}
+	sink, getEvents := captureEvents(t)
+	err := snapshot.SaveRemoteS3(context.Background(), healthyRunningMock(t), awsContainers, client, "", "my-pod", s3URL, creds, "", sink)
+	require.NoError(t, err)
+
+	// The registered URL must carry placeholders, never the secret values.
+	assert.Contains(t, gotURL, "{access_key_id}")
+	assert.NotContains(t, gotURL, "supersecret")
+	assert.NotContains(t, gotURL, "AKIA123")
+	// The secrets travel only as ephemeral params.
+	assert.Equal(t, "AKIA123", gotParams["access_key_id"])
+	assert.Equal(t, "supersecret", gotParams["secret_access_key"])
+	_, hasToken := gotParams["session_token"]
+	assert.False(t, hasToken, "session_token omitted when empty")
+
+	var saved *output.RemoteSnapshotSavedEvent
+	for _, e := range getEvents() {
+		if ev, ok := e.(output.RemoteSnapshotSavedEvent); ok {
+			saved = &ev
+		}
+	}
+	require.NotNil(t, saved)
+	assert.Equal(t, "my-pod", saved.PodName)
+	assert.Equal(t, s3URL, saved.Location)
+}
+
+func TestSaveRemoteS3_SessionTokenIncluded(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	client := NewMockRemoteClient(ctrl)
+
+	client.EXPECT().S3BucketExists(gomock.Any(), "bucket").Return(true, nil)
+	var gotURL string
+	client.EXPECT().RegisterRemote(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ context.Context, _, _, remoteURL string) error {
+			gotURL = remoteURL
+			return nil
+		},
+	)
+	client.EXPECT().SavePodRemote(gomock.Any(), gomock.Any(), "my-pod", gomock.Any(), gomock.Any(), "").DoAndReturn(
+		func(_ context.Context, _, _, _ string, params map[string]string, _ string) (snapshot.PodSaveResult, error) {
+			assert.Equal(t, "tok", params["session_token"])
+			return snapshot.PodSaveResult{}, nil
+		},
+	)
+
+	creds := snapshot.S3Credentials{AccessKeyID: "a", SecretAccessKey: "b", SessionToken: "tok"}
+	err := snapshot.SaveRemoteS3(context.Background(), healthyRunningMock(t), awsContainers, client, "", "my-pod", "s3://bucket", creds, "", output.NewPlainSink(io.Discard))
+	require.NoError(t, err)
+	assert.Contains(t, gotURL, "session_token={session_token}")
+}
+
+func TestSaveRemoteS3_RegisterError(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	client := NewMockRemoteClient(ctrl)
+	client.EXPECT().S3BucketExists(gomock.Any(), "bucket").Return(true, nil)
+	client.EXPECT().RegisterRemote(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("boom"))
+
+	err := snapshot.SaveRemoteS3(context.Background(), healthyRunningMock(t), awsContainers, client, "", "my-pod", "s3://bucket", snapshot.S3Credentials{AccessKeyID: "a", SecretAccessKey: "b"}, "", output.NewPlainSink(io.Discard))
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "register S3 remote")
+}
+
+func TestSaveRemoteS3_BucketMissingFails(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	client := NewMockRemoteClient(ctrl)
+	// Bucket does not exist → save must fail without registering or uploading.
+	// The check runs before any runtime interaction, so a bare runtime mock is used.
+	client.EXPECT().S3BucketExists(gomock.Any(), "missing-bucket").Return(false, nil)
+
+	err := snapshot.SaveRemoteS3(context.Background(), runtime.NewMockRuntime(ctrl), awsContainers, client, "", "my-pod", "s3://missing-bucket", snapshot.S3Credentials{AccessKeyID: "a", SecretAccessKey: "b"}, "", output.NewPlainSink(io.Discard))
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "does not exist")
+}
+
+func TestSaveRemoteS3_LocalEndpointSkipsBucketCheck(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	client := NewMockRemoteClient(ctrl)
+	// No S3BucketExists expectation: the local-testing endpoint must skip the check.
+	client.EXPECT().RegisterRemote(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
+	client.EXPECT().SavePodRemote(gomock.Any(), gomock.Any(), "my-pod", gomock.Any(), gomock.Any(), "").Return(snapshot.PodSaveResult{}, nil)
+
+	err := snapshot.SaveRemoteS3(context.Background(), healthyRunningMock(t), awsContainers, client, "", "my-pod", "s3://host.docker.internal:4566/my-bucket", snapshot.S3Credentials{AccessKeyID: "a", SecretAccessKey: "b"}, "", output.NewPlainSink(io.Discard))
+	require.NoError(t, err)
+}
+
+func TestSaveRemoteS3_CheckErrorWarnsAndProceeds(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	client := NewMockRemoteClient(ctrl)
+	// A check that cannot be performed degrades to a warning, not a hard failure.
+	client.EXPECT().S3BucketExists(gomock.Any(), "bucket").Return(false, fmt.Errorf("no network"))
+	client.EXPECT().RegisterRemote(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
+	client.EXPECT().SavePodRemote(gomock.Any(), gomock.Any(), "my-pod", gomock.Any(), gomock.Any(), "").Return(snapshot.PodSaveResult{}, nil)
+
+	sink, getEvents := captureEvents(t)
+	err := snapshot.SaveRemoteS3(context.Background(), healthyRunningMock(t), awsContainers, client, "", "my-pod", "s3://bucket", snapshot.S3Credentials{AccessKeyID: "a", SecretAccessKey: "b"}, "", sink)
+	require.NoError(t, err)
+
+	var warned bool
+	for _, e := range getEvents() {
+		if ev, ok := e.(output.MessageEvent); ok && ev.Severity == output.SeverityWarning {
+			warned = true
+		}
+	}
+	assert.True(t, warned, "a warning should be emitted when the bucket check fails")
+}
+
+func TestListRemoteS3_RendersTable(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	client := NewMockRemoteClient(ctrl)
+
+	mockRT := runtime.NewMockRuntime(ctrl)
+	mockRT.EXPECT().IsHealthy(gomock.Any()).Return(nil)
+
+	client.EXPECT().S3BucketExists(gomock.Any(), "bucket").Return(true, nil)
+	client.EXPECT().RegisterRemote(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
+	client.EXPECT().ListPodsRemote(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), "").Return(
+		[]snapshot.RemotePod{{Name: "pod-a", MaxVersion: 3}, {Name: "pod-b", MaxVersion: 1}}, nil,
+	)
+
+	sink, getEvents := captureEvents(t)
+	err := snapshot.ListRemoteS3(context.Background(), mockRT, awsContainers, client, "", "s3://bucket", snapshot.S3Credentials{AccessKeyID: "a", SecretAccessKey: "b"}, "", sink)
+	require.NoError(t, err)
+
+	var table *output.TableEvent
+	for _, e := range getEvents() {
+		if d, ok := e.(output.DeferredEvent); ok {
+			if tbl, ok := d.Inner.(output.TableEvent); ok {
+				table = &tbl
+			}
+		}
+	}
+	require.NotNil(t, table)
+	assert.Equal(t, []string{"Name", "Version"}, table.Headers)
+	assert.Len(t, table.Rows, 2)
+	assert.Equal(t, []string{"pod-a", "3"}, table.Rows[0])
+}
diff --git a/internal/ui/app.go b/internal/ui/app.go
index 70acd13a..ae812592 100644
--- a/internal/ui/app.go
+++ b/internal/ui/app.go
@@ -299,7 +299,7 @@ func (a App) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 		a.addLine(styledLine{text: style.Render(text)})
 		a.addLine(blank)
 		return a, nil
-	case output.InstanceInfoEvent, output.PodSnapshotSavedEvent, output.SnapshotLoadedEvent:
+	case output.InstanceInfoEvent, output.PodSnapshotSavedEvent, output.RemoteSnapshotSavedEvent, output.SnapshotLoadedEvent:
 		if line, ok := output.FormatEventLine(msg.(output.Event)); ok {
 			a.addSuccessLines(line)
 		}
diff --git a/internal/ui/run_snapshot_remote.go b/internal/ui/run_snapshot_remote.go
new file mode 100644
index 00000000..b650cdce
--- /dev/null
+++ b/internal/ui/run_snapshot_remote.go
@@ -0,0 +1,28 @@
+package ui
+
+import (
+	"context"
+
+	"github.com/localstack/lstk/internal/config"
+	"github.com/localstack/lstk/internal/output"
+	"github.com/localstack/lstk/internal/runtime"
+	"github.com/localstack/lstk/internal/snapshot"
+)
+
+func RunSnapshotSaveRemoteS3(parentCtx context.Context, rt runtime.Runtime, containers []config.ContainerConfig, client snapshot.RemoteClient, host, podName, s3URL string, creds snapshot.S3Credentials, authToken string) error {
+	return runWithTUI(parentCtx, withoutHeader(), func(ctx context.Context, sink output.Sink) error {
+		return snapshot.SaveRemoteS3(ctx, rt, containers, client, host, podName, s3URL, creds, authToken, sink)
+	})
+}
+
+func RunSnapshotLoadRemoteS3(parentCtx context.Context, rt runtime.Runtime, containers []config.ContainerConfig, client snapshot.RemoteClient, host, podName, s3URL string, creds snapshot.S3Credentials, authToken, strategy string, starter snapshot.Starter) error {
+	return runWithTUI(parentCtx, withoutHeader(), func(ctx context.Context, sink output.Sink) error {
+		return snapshot.LoadRemoteS3(ctx, rt, containers, client, host, podName, s3URL, creds, authToken, strategy, starter, sink)
+	})
+}
+
+func RunSnapshotListRemoteS3(parentCtx context.Context, rt runtime.Runtime, containers []config.ContainerConfig, client snapshot.RemoteClient, host, s3URL string, creds snapshot.S3Credentials, authToken string) error {
+	return runWithTUI(parentCtx, withoutHeader(), func(ctx context.Context, sink output.Sink) error {
+		return snapshot.ListRemoteS3(ctx, rt, containers, client, host, s3URL, creds, authToken, sink)
+	})
+}
diff --git a/test/integration/env/env.go b/test/integration/env/env.go
index dcfeb9cf..c92631b2 100644
--- a/test/integration/env/env.go
+++ b/test/integration/env/env.go
@@ -9,17 +9,19 @@ import (
 type Key string
 
 const (
-	AuthToken         Key = "LOCALSTACK_AUTH_TOKEN"
-	LocalStackHost    Key = "LOCALSTACK_HOST"
-	APIEndpoint       Key = "LSTK_API_ENDPOINT"
-	Keyring           Key = "LSTK_KEYRING"
-	CI                Key = "CI"
-	AnalyticsEndpoint Key = "LSTK_ANALYTICS_ENDPOINT"
-	DisableEvents     Key = "LOCALSTACK_DISABLE_EVENTS"
-	Home              Key = "HOME"
-	Persistence       Key = "LOCALSTACK_PERSISTENCE"
-	Otel              Key = "LSTK_OTEL"
-	OtelEndpoint      Key = "OTEL_EXPORTER_OTLP_ENDPOINT"
+	AuthToken          Key = "LOCALSTACK_AUTH_TOKEN"
+	LocalStackHost     Key = "LOCALSTACK_HOST"
+	APIEndpoint        Key = "LSTK_API_ENDPOINT"
+	Keyring            Key = "LSTK_KEYRING"
+	CI                 Key = "CI"
+	AnalyticsEndpoint  Key = "LSTK_ANALYTICS_ENDPOINT"
+	DisableEvents      Key = "LOCALSTACK_DISABLE_EVENTS"
+	Home               Key = "HOME"
+	Persistence        Key = "LOCALSTACK_PERSISTENCE"
+	Otel               Key = "LSTK_OTEL"
+	OtelEndpoint       Key = "OTEL_EXPORTER_OTLP_ENDPOINT"
+	AWSAccessKeyID     Key = "AWS_ACCESS_KEY_ID"
+	AWSSecretAccessKey Key = "AWS_SECRET_ACCESS_KEY"
 )
 
 // UnreachableAnalyticsEndpoint is a closed local port used as the default
diff --git a/test/integration/snapshot_load_test.go b/test/integration/snapshot_load_test.go
index bf51eff5..820c3106 100644
--- a/test/integration/snapshot_load_test.go
+++ b/test/integration/snapshot_load_test.go
@@ -108,7 +108,7 @@ func writeTestSnapFile(t *testing.T, dir, name string) string {
 
 func TestSnapshotLoadRemoteRejected(t *testing.T) {
 	t.Parallel()
-	for _, ref := range []string{"s3://bucket/key", "oras://registry/image"} {
+	for _, ref := range []string{"oras://registry/image"} {
 		t.Run(ref, func(t *testing.T) {
 			t.Parallel()
 			ctx := testContext(t)
@@ -122,6 +122,19 @@ func TestSnapshotLoadRemoteRejected(t *testing.T) {
 	}
 }
 
+// Loading from S3 requires a pod name (the snapshot's identity within the
+// bucket); the s3:// location alone is not enough.
+func TestSnapshotLoadS3RequiresPodName(t *testing.T) {
+	t.Parallel()
+	ctx := testContext(t)
+	_, stderr, err := runLstk(t, ctx, t.TempDir(),
+		testEnvWithHome(t.TempDir(), ""),
+		"--non-interactive", "snapshot", "load", "s3://bucket/key",
+	)
+	requireExitCode(t, 1, err)
+	assert.Contains(t, stderr, "pod name is required")
+}
+
 func TestSnapshotLoadPodNoAuthToken(t *testing.T) {
 	t.Parallel()
 	ctx := testContext(t)
diff --git a/test/integration/snapshot_save_test.go b/test/integration/snapshot_save_test.go
index 4c4d9c7a..c5b727e8 100644
--- a/test/integration/snapshot_save_test.go
+++ b/test/integration/snapshot_save_test.go
@@ -3,11 +3,14 @@ package integration_test
 import (
 	"archive/zip"
 	"bytes"
+	"encoding/json"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/localstack/lstk/test/integration/env"
@@ -173,7 +176,6 @@ func TestSnapshotSaveOverwritesExistingFile(t *testing.T) {
 func TestSnapshotSaveRemoteRejected(t *testing.T) {
 	t.Parallel()
 	for _, dest := range []string{
-		"s3://my-bucket/my-snap",
 		"oras://registry/my-snap",
 	} {
 		t.Run(dest, func(t *testing.T) {
@@ -187,6 +189,110 @@ func TestSnapshotSaveRemoteRejected(t *testing.T) {
 	}
 }
 
+// TestSnapshotSaveS3MissingCredentials and the credential/URL validation tests run
+// before any Docker interaction, so they need no running emulator.
+func TestSnapshotSaveS3MissingCredentials(t *testing.T) {
+	t.Parallel()
+	ctx := testContext(t)
+
+	_, stderr, err := runLstk(t, ctx, t.TempDir(),
+		env.Environ(testEnvWithHome(t.TempDir(), "")).Without(env.AWSAccessKeyID, env.AWSSecretAccessKey),
+		"--non-interactive", "snapshot", "save", "my-pod", "s3://my-bucket/prefix",
+	)
+	requireExitCode(t, 1, err)
+	assert.Contains(t, stderr, "AWS credentials required")
+}
+
+func TestSnapshotSaveS3CredentialsInURLRejected(t *testing.T) {
+	t.Parallel()
+	ctx := testContext(t)
+
+	_, stderr, err := runLstk(t, ctx, t.TempDir(),
+		env.Environ(testEnvWithHome(t.TempDir(), "")).
+			With(env.AWSAccessKeyID, "AKIA").
+			With(env.AWSSecretAccessKey, "secret"),
+		"--non-interactive", "snapshot", "save", "my-pod", "s3://my-bucket/prefix?access_key_id=AKIA&secret_access_key=secret",
+	)
+	requireExitCode(t, 1, err)
+	assert.Contains(t, stderr, "do not put credentials")
+}
+
+// mockPodS3Server handles the remote registration plus pod save against an S3
+// remote, capturing the registered remote URL and the save request body so the
+// test can assert the wire contract (placeholders in the URL, secrets only in the
+// ephemeral params).
+func mockPodS3Server(t *testing.T) (*httptest.Server, func() (remoteURL string, saveBody []byte)) {
+	t.Helper()
+	var (
+		mu        sync.Mutex
+		remoteURL string
+		saveBody  []byte
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case strings.HasPrefix(r.URL.Path, "/_localstack/pods/remotes/") && r.Method == http.MethodPost:
+			body, _ := io.ReadAll(r.Body)
+			var parsed struct {
+				RemoteURL string `json:"remote_url"`
+			}
+			_ = json.Unmarshal(body, &parsed)
+			mu.Lock()
+			remoteURL = parsed.RemoteURL
+			mu.Unlock()
+			w.WriteHeader(http.StatusOK)
+		case strings.HasPrefix(r.URL.Path, "/_localstack/pods/") && r.Method == http.MethodPost:
+			body, _ := io.ReadAll(r.Body)
+			mu.Lock()
+			saveBody = body
+			mu.Unlock()
+			w.Header().Set("Content-Type", "application/x-ndjson")
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte(`{"event":"completion","status":"ok","operation":"save","info":{"name":"my-pod","version":1,"services":["s3"],"size":2048}}` + "\n"))
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	t.Cleanup(srv.Close)
+	return srv, func() (string, []byte) {
+		mu.Lock()
+		defer mu.Unlock()
+		return remoteURL, saveBody
+	}
+}
+
+func TestSnapshotSaveS3Success(t *testing.T) {
+	requireDocker(t)
+	cleanup()
+	t.Cleanup(cleanup)
+
+	ctx := testContext(t)
+	startTestContainer(t, ctx)
+	srv, captured := mockPodS3Server(t)
+
+	// A local-testing endpoint (host.docker.internal) makes lstk skip the
+	// bucket-existence pre-flight, so the test never reaches real AWS S3.
+	const s3URL = "s3://host.docker.internal:4566/my-bucket"
+	stdout, stderr, err := runLstk(t, ctx, t.TempDir(),
+		env.Environ(testEnvWithHome(t.TempDir(), "")).
+			With(env.LocalStackHost, lsHost(srv)).
+			With(env.AWSAccessKeyID, "AKIAEXAMPLE").
+			With(env.AWSSecretAccessKey, "topsecret"),
+		"--non-interactive", "snapshot", "save", "my-pod", s3URL,
+	)
+	require.NoError(t, err, "lstk snapshot save to s3 failed: %s", stderr)
+	assert.Contains(t, stdout, "Snapshot saved to "+s3URL)
+	assert.Contains(t, stdout, "my-pod")
+
+	remoteURL, saveBody := captured()
+	// The registered URL carries placeholders, never the secret values.
+	assert.Contains(t, remoteURL, "{access_key_id}")
+	assert.NotContains(t, remoteURL, "topsecret")
+	assert.NotContains(t, remoteURL, "AKIAEXAMPLE")
+	// The secrets travel only in the ephemeral save params.
+	assert.Contains(t, string(saveBody), "topsecret")
+	assert.Contains(t, string(saveBody), "remote_params")
+}
+
 // mockPodSaveServer returns a test server that handles POST /_localstack/pods/{name}
 // and responds with a streaming completion event. respondOK controls whether the
 // completion event reports success or a server-side error.