feat: createImage (#34)

PLAN.md

@@ -0,0 +1,145 @@
+# Plan: Implement CreateImage API
+
+## Context
+
+Users need the ability to create a new AMI from a running (or stopped) instance so they can launch new instances from snapshots of existing ones. This is the standard AWS `CreateImage` flow: snapshot the instance's root volume, then register a new AMI pointing at that snapshot. All the building blocks exist (CreateSnapshot, AMIMetadata, OpenFromSnapshot, cloneAMIToVolume) — they just need to be wired together.
+
+## Flow
+
+```
+AWS CLI: create-image --instance-id i-xxx --name "my-image"
+  → Gateway (ec2.go)
+    → NATS "ec2.CreateImage"
+      → Daemon handler
+        → Finds instance → gets root volume ID
+        → Calls snapshotService.CreateSnapshot(volumeID)
+        → Creates new AMI viperblock config with AMIMetadata.SnapshotID
+        → Stores as ami-xxx/config.json in S3
+        → Returns ami-xxx
+```
+
+## Changes
+
+### 1. Gateway: Add CreateImage action entry
+**File**: `hive/gateway/ec2.go`
+- Add `"CreateImage"` to the `ec2Actions` map, following the DescribeImages pattern
+
+### 2. Gateway: CreateImage handler
+**File**: `hive/gateway/ec2/image/CreateImage.go` (new)
+- `ValidateCreateImageInput()` — validate InstanceId is present and starts with `i-`
+- `CreateImage()` — create NATSImageService, call `svc.CreateImage(input)`
+- Follow exact pattern of `gateway/ec2/snapshot/CreateSnapshot.go`
+
+### 3. NATS service: Wire CreateImage to NATS
+**File**: `hive/handlers/ec2/image/service_nats.go`
+- Replace the `CreateImage` stub with a real NATS request to `"ec2.CreateImage"` topic
+- Use `utils.NATSRequest[ec2.CreateImageOutput]` with 120s timeout (snapshot can be slow)
+
+### 4. Daemon: Add NATS subscription and handler
+**File**: `hive/daemon/daemon.go`
+- Add `{"ec2.CreateImage", d.handleEC2CreateImage, "hive-workers"}` to subscriptions
+- Update `NewImageServiceImpl(d.config)` → `NewImageServiceImpl(d.config, d.natsConn)` (~line 515)
+
+**File**: `hive/daemon/daemon_handlers.go`
+- Add `handleEC2CreateImage` — this is a **stateful handler** (NOT handleNATSRequest) because it needs instance state to find the root volume
+- Steps:
+  1. Unmarshal `ec2.CreateImageInput`
+  2. Lock `d.Instances.Mu`, find instance in `d.Instances.VMS[*input.InstanceId]`, unlock
+  3. Validate instance exists and is in running or stopped state
+  4. Find root volume ID from `instance.Instance.BlockDeviceMappings` (first entry with `Ebs.VolumeId`)
+  5. Get instance status (running vs stopped) — pass to service so it knows which snapshot path to use
+  6. Get source AMI ID from `instance.Instance.ImageId` (for architecture, platform, etc.)
+  7. Call `d.imageService.CreateImageFromInstance(params)` passing the extracted data
+  8. Respond with JSON
+
+### 5. Image service: Update interface and add CreateImageParams
+**File**: `hive/handlers/ec2/image/service.go`
+- Keep `CreateImage(input *ec2.CreateImageInput)` signature unchanged on the interface (gateway NATS side doesn't use extra params)
+
+**File**: `hive/handlers/ec2/image/service_impl.go`
+- Add a `CreateImageParams` struct for the daemon-side call:
+  ```go
+  type CreateImageParams struct {
+      Input         *ec2.CreateImageInput
+      RootVolumeID  string
+      SourceImageID string
+      IsRunning     bool // true = use ebs.snapshot NATS, false = snapshot from S3 state
+  }
+  ```
+- The daemon handler calls `d.imageService.CreateImageFromInstance(params)` directly (not through the generic interface)
+
+### 6. Image service: Implement CreateImageFromInstance
+**File**: `hive/handlers/ec2/image/service_impl.go`
+
+Add `natsConn *nats.Conn` to `ImageServiceImpl` (for running-instance NATS snapshot calls and offline viperblock instantiation for stopped instances). Update constructor.
+
+Implement `CreateImageFromInstance(params CreateImageParams)`:
+1. Generate AMI ID: `utils.GenerateResourceID("ami")`
+2. Generate snapshot ID: `utils.GenerateResourceID("snap")`
+3. **Snapshot the root volume** — two paths:
+   - **Running instance**: NATS `ebs.snapshot.{volumeID}` — reuse the exact pattern from `snapshot/service_impl.go:191-218`:
+     - Marshal `config.EBSSnapshotRequest{Volume: rootVolumeID, SnapshotID: snapshotID}`
+     - Send NATS request to `ebs.snapshot.{rootVolumeID}` with 30s timeout
+     - Validate response
+   - **Stopped instance**: Volume is not mounted in viperblockd. Data is fully persisted in S3. Create snapshot offline:
+     - Instantiate viperblock via `newViperblock(rootVolumeID, ...)` (reuse pattern from `instance/service_impl.go:238-258`)
+     - `backend.Init()` + `LoadStateRequest("")` to load config + block map from S3
+     - Call `vb.CreateSnapshot(snapshotID)` directly — the flush/WAL steps are no-ops on a freshly loaded instance, it just serializes the block map into the snapshot checkpoint
+     - `vb.RemoveLocalFiles()` to clean up temp files
+4. **Store snapshot metadata** in `{snapshotID}/metadata.json` — same pattern as `snapshot/service_impl.go:220-252`
+5. **Read source AMI config** from S3 `{sourceImageID}/config.json` to get architecture, platform, etc.
+6. **Read root volume config** from S3 `{rootVolumeID}/config.json` to get volume size
+7. **Build AMI VolumeConfig** with populated `AMIMetadata`:
+   - `ImageID`: new ami-xxx
+   - `Name`: from input
+   - `Description`: from input
+   - `SnapshotID`: the snap-xxx just created
+   - `Architecture`, `PlatformDetails`, `Virtualization`, etc.: from source AMI (or defaults)
+   - `VolumeSizeGiB`: from root volume config
+   - `CreationDate`: now
+   - `RootDeviceType`: "ebs"
+   - `ImageOwnerAlias`: "self"
+8. **Store AMI config** to S3 as `{amiID}/config.json` — create a `VBState` with the VolumeConfig. This is what `DescribeImages` reads.
+9. Return `&ec2.CreateImageOutput{ImageId: &amiID}`
+
+### 7. Tests
+- Add unit test for `CreateImage` in `hive/handlers/ec2/image/` with mock object store
+- Add gateway validation test in `hive/gateway/ec2/image/`
+
+## Key Files
+
+| File | Change |
+|------|--------|
+| `hive/gateway/ec2.go` | Add CreateImage action |
+| `hive/gateway/ec2/image/CreateImage.go` | New gateway handler |
+| `hive/handlers/ec2/image/service.go` | Update interface |
+| `hive/handlers/ec2/image/service_nats.go` | Wire NATS request |
+| `hive/handlers/ec2/image/service_impl.go` | Core implementation |
+| `hive/daemon/daemon.go` | Add subscription, update constructor |
+| `hive/daemon/daemon_handlers.go` | Add stateful handler |
+
+## Existing Code to Reuse
+
+- `snapshot/service_impl.go:191-218` — EBS snapshot NATS pattern (`ebs.snapshot.{volumeID}`)
+- `snapshot/service_impl.go:107-123` — `putSnapshotConfig` S3 storage pattern
+- `image/service_impl.go:96-126` — Reading viperblock VBState from S3 (`config.json`)
+- `instance/service_impl.go:238-258` — `newViperblock()` for offline viperblock instantiation
+- `utils.GenerateResourceID()` — ID generation
+- `utils.NATSRequest[T]()` — Generic NATS request helper
+- `viperblock.VBState`, `viperblock.VolumeConfig`, `viperblock.AMIMetadata` — Existing types
+
+## Verification
+
+1. `make build` — confirm compilation
+2. `make test` — all existing tests pass
+3. Manual test:
+   ```bash
+   # Launch instance from existing AMI
+   aws ec2 run-instances --image-id ami-xxx --instance-type t2.micro --key-name mykey
+   # Create image from running instance
+   aws ec2 create-image --instance-id i-xxx --name "test-snapshot-image"
+   # Verify it appears
+   aws ec2 describe-images --owners self
+   # Launch new instance from the snapshot-based AMI
+   aws ec2 run-instances --image-id ami-yyy --instance-type t2.micro --key-name mykey
+   ```

hive/daemon/daemon.go

@@ -448,6 +448,7 @@ func (d *Daemon) subscribeAll() error {
 		{"ec2.DescribeKeyPairs", d.handleEC2DescribeKeyPairs, "hive-workers"},
 		{"ec2.ImportKeyPair", d.handleEC2ImportKeyPair, "hive-workers"},
 		{"ec2.DescribeImages", d.handleEC2DescribeImages, "hive-workers"},
+		{"ec2.CreateImage", d.handleEC2CreateImage, ""},
 		{"ec2.CreateVolume", d.handleEC2CreateVolume, "hive-workers"},
 		{"ec2.DescribeVolumes", d.handleEC2DescribeVolumes, "hive-workers"},
 		{"ec2.ModifyVolume", d.handleEC2ModifyVolume, "hive-workers"},
@@ -512,7 +513,7 @@ func (d *Daemon) Start() error {
 	store := objectstore.NewS3ObjectStoreFromConfig(d.config.Predastore.Host, d.config.Predastore.Region, d.config.AccessKey, d.config.SecretKey)
 	d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store)
 	d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config)
-	d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config)
+	d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config, d.natsConn)
 	d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn)
 	d.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImpl(d.config, d.natsConn)
 	d.tagsService = handlers_ec2_tags.NewTagsServiceImpl(d.config)
@@ -633,6 +634,37 @@ func (d *Daemon) restoreInstances() {
 			}
 		}
 
+		// Check if QEMU process is still alive from before the restart
+		if d.isInstanceProcessRunning(instance) {
+			slog.Info("Instance QEMU process still alive, reconnecting", "instance", instance.ID)
+			if err := d.reconnectInstance(instance); err != nil {
+				slog.Error("Failed to reconnect to running instance", "instanceId", instance.ID, "err", err)
+			}
+			continue
+		}
+
+		// QEMU is not running - handle transitional states from interrupted operations
+		switch instance.Status {
+		case vm.StateStopping:
+			instance.Status = vm.StateStopped
+			slog.Info("Instance was stopping, QEMU exited, marking stopped", "instance", instance.ID)
+			if err := d.WriteState(); err != nil {
+				slog.Error("Failed to write state", "instance", instance.ID, "error", err)
+			}
+			continue
+		case vm.StateShuttingDown:
+			instance.Status = vm.StateTerminated
+			slog.Info("Instance was shutting-down, QEMU exited, marking terminated", "instance", instance.ID)
+			if err := d.WriteState(); err != nil {
+				slog.Error("Failed to write state", "instance", instance.ID, "error", err)
+			}
+			continue
+		case vm.StateRunning:
+			// Was running but QEMU died - reset to pending so LaunchInstance can transition to running
+			instance.Status = vm.StatePending
+			slog.Info("Instance was running but QEMU exited, relaunching", "instance", instance.ID)
+		}
+
 		slog.Info("Launching instance", "instance", instance.ID)
 		if err := d.LaunchInstance(instance); err != nil {
 			slog.Error("Failed to launch instance", "instanceId", instance.ID, "err", err)
@@ -640,6 +672,46 @@ func (d *Daemon) restoreInstances() {
 	}
 }
 
+// isInstanceProcessRunning checks if the QEMU process for an instance is still alive.
+func (d *Daemon) isInstanceProcessRunning(instance *vm.VM) bool {
+	pid, err := utils.ReadPidFile(instance.ID)
+	if err != nil || pid <= 0 {
+		return false
+	}
+	process, err := os.FindProcess(pid)
+	if err != nil {
+		return false
+	}
+	return process.Signal(syscall.Signal(0)) == nil
+}
+
+// reconnectInstance re-establishes QMP and NATS connections to a running QEMU instance
+// after a daemon restart. This bypasses the state machine since recovery is not a
+// normal state transition.
+func (d *Daemon) reconnectInstance(instance *vm.VM) error {
+	if err := d.CreateQMPClient(instance); err != nil {
+		return fmt.Errorf("failed to reconnect QMP: %w", err)
+	}
+
+	d.mu.Lock()
+	sub, err := d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events)
+	if err != nil {
+		d.mu.Unlock()
+		return fmt.Errorf("failed to subscribe to NATS: %w", err)
+	}
+	d.natsSubscriptions[instance.ID] = sub
+	d.mu.Unlock()
+
+	instance.Status = vm.StateRunning
+
+	if err := d.WriteState(); err != nil {
+		slog.Error("Failed to persist reconnected instance state", "instanceId", instance.ID, "err", err)
+	}
+
+	slog.Info("Successfully reconnected to running instance", "instance", instance.ID)
+	return nil
+}
+
 // awaitShutdown blocks until the daemon's shutdown wait group completes.
 func (d *Daemon) awaitShutdown() {
 	done := make(chan struct{})

hive/daemon/daemon_handlers.go

@@ -11,6 +11,7 @@ import (
 	"github.com/mulgadc/hive/hive/awserrors"
 	"github.com/mulgadc/hive/hive/config"
 	gateway_ec2_instance "github.com/mulgadc/hive/hive/gateway/ec2/instance"
+	handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image"
 	"github.com/mulgadc/hive/hive/qmp"
 	"github.com/mulgadc/hive/hive/utils"
 	"github.com/mulgadc/hive/hive/vm"
@@ -452,6 +453,17 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance
 func (d *Daemon) handleStartInstance(msg *nats.Msg, command qmp.Command, instance *vm.VM, respondWithError func(string)) {
 	slog.Info("Starting instance", "id", command.ID)
 
+	// Validate instance is in stopped state
+	d.Instances.Mu.Lock()
+	status := instance.Status
+	d.Instances.Mu.Unlock()
+
+	if status != vm.StateStopped {
+		slog.Error("StartInstance: instance not in stopped state", "instanceId", command.ID, "status", status)
+		respondWithError(awserrors.ErrorIncorrectInstanceState)
+		return
+	}
+
 	// Allocate resources
 	instanceType, ok := d.resourceMgr.instanceTypes[instance.InstanceType]
 	if ok {
@@ -814,6 +826,99 @@ func (d *Daemon) handleEC2DescribeImages(msg *nats.Msg) {
 	handleNATSRequest(msg, d.imageService.DescribeImages)
 }
 
+// handleEC2CreateImage is a stateful handler that extracts instance context
+// (root volume ID, source AMI, running state) before delegating to the image service.
+func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) {
+	slog.Debug("Received message", "subject", msg.Subject)
+
+	respondWithError := func(errCode string) {
+		if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil {
+			slog.Error("Failed to respond to NATS request", "err", err)
+		}
+	}
+
+	input := &ec2.CreateImageInput{}
+	if errResp := utils.UnmarshalJsonPayload(input, msg.Data); errResp != nil {
+		if err := msg.Respond(errResp); err != nil {
+			slog.Error("Failed to respond to NATS request", "err", err)
+		}
+		return
+	}
+
+	if input.InstanceId == nil || *input.InstanceId == "" {
+		respondWithError(awserrors.ErrorMissingParameter)
+		return
+	}
+
+	instanceID := *input.InstanceId
+
+	// Extract all instance context in a single critical section
+	d.Instances.Mu.Lock()
+	instance, ok := d.Instances.VMS[instanceID]
+	var status vm.InstanceState
+	var rootVolumeID, sourceImageID string
+	if ok {
+		status = instance.Status
+		if instance.Instance != nil {
+			for _, bdm := range instance.Instance.BlockDeviceMappings {
+				if bdm.Ebs != nil && bdm.Ebs.VolumeId != nil {
+					rootVolumeID = *bdm.Ebs.VolumeId
+					break
+				}
+			}
+			if instance.Instance.ImageId != nil {
+				sourceImageID = *instance.Instance.ImageId
+			}
+		}
+	}
+	d.Instances.Mu.Unlock()
+
+	if !ok {
+		slog.Error("CreateImage: instance not found", "instanceId", instanceID)
+		respondWithError(awserrors.ErrorInvalidInstanceIDNotFound)
+		return
+	}
+
+	if status != vm.StateRunning && status != vm.StateStopped {
+		slog.Error("CreateImage: instance not in valid state", "instanceId", instanceID, "status", status)
+		respondWithError(awserrors.ErrorIncorrectInstanceState)
+		return
+	}
+
+	if rootVolumeID == "" {
+		slog.Error("CreateImage: no root volume found", "instanceId", instanceID)
+		respondWithError(awserrors.ErrorServerInternal)
+		return
+	}
+
+	params := handlers_ec2_image.CreateImageParams{
+		Input:         input,
+		RootVolumeID:  rootVolumeID,
+		SourceImageID: sourceImageID,
+		IsRunning:     status == vm.StateRunning,
+	}
+
+	output, err := d.imageService.CreateImageFromInstance(params)
+	if err != nil {
+		slog.Error("CreateImage: service failed", "instanceId", instanceID, "err", err)
+		respondWithError(err.Error())
+		return
+	}
+
+	jsonResponse, err := json.Marshal(output)
+	if err != nil {
+		slog.Error("CreateImage: failed to marshal response", "err", err)
+		respondWithError(awserrors.ErrorServerInternal)
+		return
+	}
+
+	if err := msg.Respond(jsonResponse); err != nil {
+		slog.Error("Failed to respond to NATS request", "err", err)
+	}
+
+	slog.Info("CreateImage completed", "instanceId", instanceID, "imageId", *output.ImageId)
+}
+
 func (d *Daemon) handleEC2CreateVolume(msg *nats.Msg) {
 	handleNATSRequest(msg, d.volumeService.CreateVolume)
 }

hive/daemon/daemon_handlers_test.go

@@ -642,6 +642,148 @@ func TestHandleEC2DisableSerialConsoleAccess(t *testing.T) {
 	assert.False(t, *output.SerialConsoleAccessEnabled)
 }
 
+// --- handleEC2CreateImage tests ---
+
+func TestHandleEC2CreateImage_InstanceNotFound(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.CreateImageInput{
+		InstanceId: aws.String("i-nonexistent"),
+		Name:       aws.String("my-image"),
+	}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
+func TestHandleEC2CreateImage_MissingInstanceId(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.CreateImageInput{
+		Name: aws.String("my-image"),
+	}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
+func TestHandleEC2CreateImage_InvalidState(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	// Add an instance in "pending" state (not running or stopped)
+	daemon.Instances.Mu.Lock()
+	daemon.Instances.VMS["i-pending123"] = &vm.VM{
+		ID:     "i-pending123",
+		Status: vm.StatePending,
+		Instance: &ec2.Instance{
+			InstanceId: aws.String("i-pending123"),
+			ImageId:    aws.String("ami-source"),
+			BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
+				{
+					DeviceName: aws.String("/dev/sda1"),
+					Ebs:        &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-root123")},
+				},
+			},
+		},
+	}
+	daemon.Instances.Mu.Unlock()
+
+	sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.CreateImageInput{
+		InstanceId: aws.String("i-pending123"),
+		Name:       aws.String("my-image"),
+	}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
+func TestHandleEC2CreateImage_NoRootVolume(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	// Add instance with no block device mappings
+	daemon.Instances.Mu.Lock()
+	daemon.Instances.VMS["i-novol123"] = &vm.VM{
+		ID:     "i-novol123",
+		Status: vm.StateRunning,
+		Instance: &ec2.Instance{
+			InstanceId:          aws.String("i-novol123"),
+			ImageId:             aws.String("ami-source"),
+			BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{},
+		},
+	}
+	daemon.Instances.Mu.Unlock()
+
+	sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.CreateImageInput{
+		InstanceId: aws.String("i-novol123"),
+		Name:       aws.String("my-image"),
+	}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
+func TestHandleEC2CreateImage_MalformedJSON(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reply, err := daemon.natsConn.Request("ec2.CreateImage", []byte(`{bad json}`), 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
 // --- SetConfigPath test ---
 
 func TestSetConfigPath(t *testing.T) {

hive/gateway/ec2.go

@@ -87,6 +87,9 @@ var ec2Actions = map[string]EC2Handler{
 	"DescribeImages": ec2Handler(func(input *ec2.DescribeImagesInput, gw *GatewayConfig) (any, error) {
 		return gateway_ec2_image.DescribeImages(input, gw.NATSConn)
 	}),
+	"CreateImage": ec2Handler(func(input *ec2.CreateImageInput, gw *GatewayConfig) (any, error) {
+		return gateway_ec2_image.CreateImage(input, gw.NATSConn)
+	}),
 	"DescribeRegions": ec2Handler(func(input *ec2.DescribeRegionsInput, gw *GatewayConfig) (any, error) {
 		return gateway_ec2_zone.DescribeRegions(input, gw.Region)
 	}),

hive/gateway/ec2/image/CreateImage.go

@@ -0,0 +1,52 @@
+package gateway_ec2_image
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/hive/hive/awserrors"
+	handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image"
+	"github.com/nats-io/nats.go"
+)
+
+// ValidateCreateImageInput validates the input parameters for CreateImage
+func ValidateCreateImageInput(input *ec2.CreateImageInput) error {
+	if input == nil {
+		return errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+
+	if input.InstanceId == nil || *input.InstanceId == "" {
+		return errors.New(awserrors.ErrorMissingParameter)
+	}
+
+	if !strings.HasPrefix(*input.InstanceId, "i-") {
+		return errors.New(awserrors.ErrorInvalidInstanceIDMalformed)
+	}
+
+	if input.Name == nil || *input.Name == "" {
+		return errors.New(awserrors.ErrorMissingParameter)
+	}
+
+	return nil
+}
+
+// CreateImage handles the EC2 CreateImage API call
+func CreateImage(input *ec2.CreateImageInput, natsConn *nats.Conn) (ec2.CreateImageOutput, error) {
+	var output ec2.CreateImageOutput
+
+	if err := ValidateCreateImageInput(input); err != nil {
+		return output, err
+	}
+
+	svc := handlers_ec2_image.NewNATSImageService(natsConn)
+	result, err := svc.CreateImage(input)
+	if err != nil {
+		return output, err
+	}
+	if result == nil {
+		return output, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	return *result, nil
+}

hive/gateway/ec2/image/CreateImage_test.go

@@ -0,0 +1,97 @@
+package gateway_ec2_image
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/hive/hive/awserrors"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidateCreateImageInput(t *testing.T) {
+	tests := []struct {
+		name    string
+		input   *ec2.CreateImageInput
+		wantErr bool
+		errMsg  string
+	}{
+		{
+			name:    "NilInput",
+			input:   nil,
+			wantErr: true,
+			errMsg:  awserrors.ErrorInvalidParameterValue,
+		},
+		{
+			name:    "EmptyInput",
+			input:   &ec2.CreateImageInput{},
+			wantErr: true,
+			errMsg:  awserrors.ErrorMissingParameter,
+		},
+		{
+			name: "MissingName",
+			input: &ec2.CreateImageInput{
+				InstanceId: aws.String("i-1234567890abcdef0"),
+			},
+			wantErr: true,
+			errMsg:  awserrors.ErrorMissingParameter,
+		},
+		{
+			name: "InvalidInstanceIdFormat",
+			input: &ec2.CreateImageInput{
+				InstanceId: aws.String("invalid-id"),
+				Name:       aws.String("my-image"),
+			},
+			wantErr: true,
+			errMsg:  awserrors.ErrorInvalidInstanceIDMalformed,
+		},
+		{
+			name: "EmptyInstanceId",
+			input: &ec2.CreateImageInput{
+				InstanceId: aws.String(""),
+				Name:       aws.String("my-image"),
+			},
+			wantErr: true,
+			errMsg:  awserrors.ErrorMissingParameter,
+		},
+		{
+			name: "EmptyName",
+			input: &ec2.CreateImageInput{
+				InstanceId: aws.String("i-1234567890abcdef0"),
+				Name:       aws.String(""),
+			},
+			wantErr: true,
+			errMsg:  awserrors.ErrorMissingParameter,
+		},
+		{
+			name: "ValidInput",
+			input: &ec2.CreateImageInput{
+				InstanceId: aws.String("i-1234567890abcdef0"),
+				Name:       aws.String("my-image"),
+			},
+			wantErr: false,
+		},
+		{
+			name: "ValidInputWithDescription",
+			input: &ec2.CreateImageInput{
+				InstanceId:  aws.String("i-1234567890abcdef0"),
+				Name:        aws.String("my-image"),
+				Description: aws.String("A snapshot image"),
+			},
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := ValidateCreateImageInput(tt.input)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				assert.Equal(t, tt.errMsg, err.Error())
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

hive/handlers/ec2/image/service_impl.go

@@ -1,35 +1,51 @@
 package handlers_ec2_image
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"strings"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/mulgadc/hive/hive/awserrors"
 	"github.com/mulgadc/hive/hive/config"
+	handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot"
 	"github.com/mulgadc/hive/hive/objectstore"
 	"github.com/mulgadc/hive/hive/utils"
 	"github.com/mulgadc/viperblock/viperblock"
+	vbs3 "github.com/mulgadc/viperblock/viperblock/backends/s3"
+	"github.com/nats-io/nats.go"
 )
 
 // Ensure ImageServiceImpl implements ImageService
 var _ ImageService = (*ImageServiceImpl)(nil)
 
+// CreateImageParams holds parameters for creating an AMI from an instance.
+// Used by the daemon handler which extracts instance state before calling the service.
+type CreateImageParams struct {
+	Input         *ec2.CreateImageInput
+	RootVolumeID  string
+	SourceImageID string
+	IsRunning     bool // true = use ebs.snapshot NATS, false = snapshot from S3 state
+}
+
 // ImageServiceImpl handles AMI image operations with S3 storage
 type ImageServiceImpl struct {
 	config     *config.Config
 	store      objectstore.ObjectStore
+	natsConn   *nats.Conn
 	bucketName string
 	accountID  string // AWS account ID for S3 key storage path
 }
 
 // NewImageServiceImpl creates a new daemon-side image service
-func NewImageServiceImpl(cfg *config.Config) *ImageServiceImpl {
+func NewImageServiceImpl(cfg *config.Config, natsConn *nats.Conn) *ImageServiceImpl {
 	store := objectstore.NewS3ObjectStoreFromConfig(
 		cfg.Predastore.Host,
 		cfg.Predastore.Region,
@@ -40,6 +56,7 @@ func NewImageServiceImpl(cfg *config.Config) *ImageServiceImpl {
 	return &ImageServiceImpl{
 		config:     cfg,
 		store:      store,
+		natsConn:   natsConn,
 		bucketName: cfg.Predastore.Bucket,
 		accountID:  "123456789", // TODO: Implement proper account ID management
 	}
@@ -215,9 +232,275 @@ func (s *ImageServiceImpl) DescribeImages(input *ec2.DescribeImagesInput) (*ec2.
 	}, nil
 }
 
-// Stub implementations for other ImageService methods
+// CreateImage is the generic interface method — on the daemon side, the handler
+// calls CreateImageFromInstance directly with the extra instance context.
 func (s *ImageServiceImpl) CreateImage(input *ec2.CreateImageInput) (*ec2.CreateImageOutput, error) {
-	return nil, errors.New("CreateImage not yet implemented")
+	return nil, errors.New("CreateImage requires instance context; use CreateImageFromInstance on daemon side")
+}
+
+// CreateImageFromInstance creates an AMI from an instance by snapshotting the root
+// volume and storing a new AMI config in S3.
+func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*ec2.CreateImageOutput, error) {
+	input := params.Input
+	if input == nil || input.InstanceId == nil {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+
+	amiID := utils.GenerateResourceID("ami")
+	snapshotID := utils.GenerateResourceID("snap")
+
+	slog.Info("CreateImageFromInstance", "instanceId", *input.InstanceId,
+		"rootVolumeId", params.RootVolumeID, "amiId", amiID, "snapshotId", snapshotID,
+		"isRunning", params.IsRunning)
+
+	// Step 1: Snapshot the root volume
+	if params.IsRunning {
+		if err := s.snapshotRunningVolume(params.RootVolumeID, snapshotID); err != nil {
+			return nil, err
+		}
+	} else {
+		if err := s.snapshotStoppedVolume(params.RootVolumeID, snapshotID); err != nil {
+			return nil, err
+		}
+	}
+
+	// Step 2: Read source volume config for size
+	volumeConfig, err := s.getVolumeConfig(params.RootVolumeID)
+	if err != nil {
+		slog.Error("CreateImageFromInstance: failed to read volume config", "volumeId", params.RootVolumeID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	volumeSizeGiB := volumeConfig.VolumeMetadata.SizeGiB
+
+	// Step 3: Read source AMI config for architecture, platform, etc.
+	var sourceAMI viperblock.AMIMetadata
+	if params.SourceImageID != "" {
+		srcCfg, err := s.getAMIConfig(params.SourceImageID)
+		if err != nil {
+			slog.Error("CreateImageFromInstance: failed to read source AMI config", "sourceImageId", params.SourceImageID, "err", err)
+			return nil, errors.New(awserrors.ErrorServerInternal)
+		}
+		sourceAMI = srcCfg
+	} else {
+		sourceAMI = viperblock.AMIMetadata{
+			Architecture:    "x86_64",
+			PlatformDetails: "Linux/UNIX",
+			Virtualization:  "hvm",
+		}
+	}
+
+	// Step 4: Store snapshot metadata
+	if err := s.putSnapshotMetadata(snapshotID, params.RootVolumeID, volumeSizeGiB); err != nil {
+		slog.Error("CreateImageFromInstance: failed to write snapshot metadata", "snapshotId", snapshotID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	// Step 5: Build and store AMI config
+	name := aws.StringValue(input.Name)
+	description := aws.StringValue(input.Description)
+
+	amiConfig := viperblock.VBState{
+		VolumeConfig: viperblock.VolumeConfig{
+			AMIMetadata: viperblock.AMIMetadata{
+				ImageID:         amiID,
+				Name:            name,
+				Description:     description,
+				SnapshotID:      snapshotID,
+				Architecture:    sourceAMI.Architecture,
+				PlatformDetails: sourceAMI.PlatformDetails,
+				Virtualization:  sourceAMI.Virtualization,
+				VolumeSizeGiB:   volumeSizeGiB,
+				CreationDate:    time.Now(),
+				RootDeviceType:  "ebs",
+				ImageOwnerAlias: "self",
+			},
+		},
+	}
+
+	configData, err := json.Marshal(amiConfig)
+	if err != nil {
+		slog.Error("CreateImageFromInstance: failed to marshal AMI config", "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	configKey := fmt.Sprintf("%s/config.json", amiID)
+	_, err = s.store.PutObject(&s3.PutObjectInput{
+		Bucket:      aws.String(s.bucketName),
+		Key:         aws.String(configKey),
+		Body:        bytes.NewReader(configData),
+		ContentType: aws.String("application/json"),
+	})
+	if err != nil {
+		slog.Error("CreateImageFromInstance: failed to store AMI config", "amiId", amiID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	slog.Info("CreateImageFromInstance completed", "amiId", amiID, "snapshotId", snapshotID)
+
+	return &ec2.CreateImageOutput{
+		ImageId: aws.String(amiID),
+	}, nil
+}
+
+// snapshotRunningVolume triggers a snapshot via NATS on a running viperblockd instance
+func (s *ImageServiceImpl) snapshotRunningVolume(volumeID, snapshotID string) error {
+	if s.natsConn == nil {
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	snapReq := config.EBSSnapshotRequest{Volume: volumeID, SnapshotID: snapshotID}
+	snapData, err := json.Marshal(snapReq)
+	if err != nil {
+		slog.Error("snapshotRunningVolume: failed to marshal snapshot request", "volumeId", volumeID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	msg, err := s.natsConn.Request(fmt.Sprintf("ebs.snapshot.%s", volumeID), snapData, 30*time.Second)
+	if err != nil {
+		slog.Error("snapshotRunningVolume: NATS request failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	var snapResp config.EBSSnapshotResponse
+	if err := json.Unmarshal(msg.Data, &snapResp); err != nil {
+		slog.Error("snapshotRunningVolume: failed to unmarshal response", "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+	if !snapResp.Success || snapResp.Error != "" {
+		slog.Error("snapshotRunningVolume: snapshot failed", "volumeId", volumeID, "err", snapResp.Error)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	slog.Info("snapshotRunningVolume: snapshot created", "volumeId", volumeID, "snapshotId", snapshotID)
+	return nil
+}
+
+// snapshotStoppedVolume creates a snapshot offline by loading viperblock state from S3
+func (s *ImageServiceImpl) snapshotStoppedVolume(volumeID, snapshotID string) error {
+	if s.config == nil {
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	// Read volume config to get the size (required by viperblock.New)
+	volConfig, err := s.getVolumeConfig(volumeID)
+	if err != nil {
+		slog.Error("snapshotStoppedVolume: failed to read volume config", "volumeId", volumeID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+	volumeSize := volConfig.VolumeMetadata.SizeGiB * 1024 * 1024 * 1024
+
+	cfg := vbs3.S3Config{
+		VolumeName: volumeID,
+		VolumeSize: volumeSize,
+		Bucket:     s.config.Predastore.Bucket,
+		Region:     s.config.Predastore.Region,
+		AccessKey:  s.config.Predastore.AccessKey,
+		SecretKey:  s.config.Predastore.SecretKey,
+		Host:       s.config.Predastore.Host,
+	}
+
+	vbconfig := viperblock.VB{
+		VolumeName: volumeID,
+		VolumeSize: volumeSize,
+		BaseDir:    s.config.WalDir,
+		Cache:      viperblock.Cache{Config: viperblock.CacheConfig{Size: 0}},
+	}
+
+	vb, err := viperblock.New(&vbconfig, "s3", cfg)
+	if err != nil {
+		slog.Error("snapshotStoppedVolume: failed to create viperblock instance", "volumeId", volumeID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	if err := vb.Backend.Init(); err != nil {
+		slog.Error("snapshotStoppedVolume: failed to init backend", "volumeId", volumeID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	if _, err := vb.LoadStateRequest(""); err != nil {
+		slog.Error("snapshotStoppedVolume: failed to load state", "volumeId", volumeID, "err", err)
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	if _, err := vb.CreateSnapshot(snapshotID); err != nil {
+		slog.Error("snapshotStoppedVolume: CreateSnapshot failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err)
+		if cleanupErr := vb.RemoveLocalFiles(); cleanupErr != nil {
+			slog.Error("snapshotStoppedVolume: failed to remove local files after snapshot failure", "volumeId", volumeID, "err", cleanupErr)
+		}
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+
+	if err := vb.RemoveLocalFiles(); err != nil {
+		slog.Error("snapshotStoppedVolume: failed to remove local files", "volumeId", volumeID, "err", err)
+	}
+
+	slog.Info("snapshotStoppedVolume: snapshot created", "volumeId", volumeID, "snapshotId", snapshotID)
+	return nil
+}
+
+// getVolumeConfig reads a volume's VBState config from S3
+func (s *ImageServiceImpl) getVolumeConfig(volumeID string) (*viperblock.VolumeConfig, error) {
+	configKey := fmt.Sprintf("%s/config.json", volumeID)
+	result, err := s.store.GetObject(&s3.GetObjectInput{
+		Bucket: aws.String(s.bucketName),
+		Key:    aws.String(configKey),
+	})
+	if err != nil {
+		return nil, err
+	}
+	defer result.Body.Close()
+
+	var vbState viperblock.VBState
+	if err := json.NewDecoder(result.Body).Decode(&vbState); err != nil {
+		return nil, err
+	}
+	return &vbState.VolumeConfig, nil
+}
+
+// getAMIConfig reads an AMI's metadata from S3
+func (s *ImageServiceImpl) getAMIConfig(imageID string) (viperblock.AMIMetadata, error) {
+	configKey := fmt.Sprintf("%s/config.json", imageID)
+	result, err := s.store.GetObject(&s3.GetObjectInput{
+		Bucket: aws.String(s.bucketName),
+		Key:    aws.String(configKey),
+	})
+	if err != nil {
+		return viperblock.AMIMetadata{}, err
+	}
+	defer result.Body.Close()
+
+	var vbState viperblock.VBState
+	if err := json.NewDecoder(result.Body).Decode(&vbState); err != nil {
+		return viperblock.AMIMetadata{}, err
+	}
+	return vbState.VolumeConfig.AMIMetadata, nil
+}
+
+// putSnapshotMetadata stores snapshot metadata in S3 using the canonical SnapshotConfig type
+func (s *ImageServiceImpl) putSnapshotMetadata(snapshotID, volumeID string, volumeSizeGiB uint64) error {
+	cfg := handlers_ec2_snapshot.SnapshotConfig{
+		SnapshotID: snapshotID,
+		VolumeID:   volumeID,
+		VolumeSize: utils.SafeUint64ToInt64(volumeSizeGiB),
+		State:      "completed",
+		Progress:   "100%",
+		StartTime:  time.Now(),
+		OwnerID:    s.accountID,
+	}
+
+	data, err := json.Marshal(cfg)
+	if err != nil {
+		return err
+	}
+
+	key := fmt.Sprintf("%s/metadata.json", snapshotID)
+	_, err = s.store.PutObject(&s3.PutObjectInput{
+		Bucket:      aws.String(s.bucketName),
+		Key:         aws.String(key),
+		Body:        bytes.NewReader(data),
+		ContentType: aws.String("application/json"),
+	})
+	return err
 }
 
 func (s *ImageServiceImpl) CopyImage(input *ec2.CopyImageInput) (*ec2.CopyImageOutput, error) {

hive/handlers/ec2/image/service_impl_test.go

@@ -0,0 +1,261 @@
+package handlers_ec2_image
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	awss3 "github.com/aws/aws-sdk-go/service/s3"
+	"github.com/mulgadc/hive/hive/awserrors"
+	handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot"
+	"github.com/mulgadc/hive/hive/objectstore"
+	"github.com/mulgadc/viperblock/viperblock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const testBucket = "test-bucket"
+const testAccountID = "123456789"
+
+// setupTestImageService creates an image service with in-memory storage for testing
+func setupTestImageService(t *testing.T) (*ImageServiceImpl, *objectstore.MemoryObjectStore) {
+	store := objectstore.NewMemoryObjectStore()
+	svc := NewImageServiceImplWithStore(store, testBucket, testAccountID)
+	return svc, store
+}
+
+// createTestVolumeConfig creates a test volume config in the mock store
+func createTestVolumeConfig(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string, sizeGiB int) {
+	volumeState := viperblock.VBState{
+		VolumeConfig: viperblock.VolumeConfig{
+			VolumeMetadata: viperblock.VolumeMetadata{
+				SizeGiB: uint64(sizeGiB),
+			},
+		},
+	}
+	data, err := json.Marshal(volumeState)
+	require.NoError(t, err)
+
+	_, err = store.PutObject(&awss3.PutObjectInput{
+		Bucket:      aws.String(testBucket),
+		Key:         aws.String(volumeID + "/config.json"),
+		Body:        strings.NewReader(string(data)),
+		ContentType: aws.String("application/json"),
+	})
+	require.NoError(t, err)
+}
+
+// createTestAMIConfig creates a test AMI config in the mock store
+func createTestAMIConfig(t *testing.T, store *objectstore.MemoryObjectStore, imageID string) {
+	amiState := viperblock.VBState{
+		VolumeConfig: viperblock.VolumeConfig{
+			AMIMetadata: viperblock.AMIMetadata{
+				ImageID:         imageID,
+				Name:            "test-ami",
+				Architecture:    "x86_64",
+				PlatformDetails: "Linux/UNIX",
+				Virtualization:  "hvm",
+				RootDeviceType:  "ebs",
+				VolumeSizeGiB:   8,
+			},
+		},
+	}
+	data, err := json.Marshal(amiState)
+	require.NoError(t, err)
+
+	_, err = store.PutObject(&awss3.PutObjectInput{
+		Bucket:      aws.String(testBucket),
+		Key:         aws.String(imageID + "/config.json"),
+		Body:        strings.NewReader(string(data)),
+		ContentType: aws.String("application/json"),
+	})
+	require.NoError(t, err)
+}
+
+func TestCreateImageFromInstance_NilInput(t *testing.T) {
+	svc, _ := setupTestImageService(t)
+
+	_, err := svc.CreateImageFromInstance(CreateImageParams{})
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
+}
+
+func TestCreateImageFromInstance_RunningInstance_NoNATS(t *testing.T) {
+	svc, store := setupTestImageService(t)
+
+	// Create volume and AMI configs
+	createTestVolumeConfig(t, store, "vol-root123", 10)
+	createTestAMIConfig(t, store, "ami-source123")
+
+	// Running instance without NATS should fail (natsConn is nil)
+	_, err := svc.CreateImageFromInstance(CreateImageParams{
+		Input: &ec2.CreateImageInput{
+			InstanceId: aws.String("i-test123"),
+			Name:       aws.String("my-image"),
+		},
+		RootVolumeID:  "vol-root123",
+		SourceImageID: "ami-source123",
+		IsRunning:     true,
+	})
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+}
+
+func TestCreateImageFromInstance_StoppedInstance_NoConfig(t *testing.T) {
+	svc, _ := setupTestImageService(t)
+
+	// Stopped instance without config will fail (no config to create viperblock)
+	_, err := svc.CreateImageFromInstance(CreateImageParams{
+		Input: &ec2.CreateImageInput{
+			InstanceId: aws.String("i-test123"),
+			Name:       aws.String("my-image"),
+		},
+		RootVolumeID:  "vol-nonexistent",
+		SourceImageID: "ami-source123",
+		IsRunning:     false,
+	})
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+}
+
+func TestDescribeImages_AfterCreate(t *testing.T) {
+	svc, store := setupTestImageService(t)
+
+	// Manually create an AMI config in the store (simulates what CreateImageFromInstance does)
+	amiID := "ami-testimage123"
+	createTestAMIConfig(t, store, amiID)
+
+	// Describe images should find it
+	result, err := svc.DescribeImages(&ec2.DescribeImagesInput{
+		ImageIds: []*string{aws.String(amiID)},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	require.Len(t, result.Images, 1)
+	assert.Equal(t, amiID, *result.Images[0].ImageId)
+	assert.Equal(t, "test-ami", *result.Images[0].Name)
+	assert.Equal(t, "x86_64", *result.Images[0].Architecture)
+}
+
+func TestGetVolumeConfig(t *testing.T) {
+	svc, store := setupTestImageService(t)
+
+	createTestVolumeConfig(t, store, "vol-abc123", 20)
+
+	cfg, err := svc.getVolumeConfig("vol-abc123")
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+	assert.Equal(t, uint64(20), cfg.VolumeMetadata.SizeGiB)
+}
+
+func TestGetVolumeConfig_NotFound(t *testing.T) {
+	svc, _ := setupTestImageService(t)
+
+	_, err := svc.getVolumeConfig("vol-nonexistent")
+	require.Error(t, err)
+}
+
+func TestGetAMIConfig(t *testing.T) {
+	svc, store := setupTestImageService(t)
+
+	createTestAMIConfig(t, store, "ami-abc123")
+
+	meta, err := svc.getAMIConfig("ami-abc123")
+	require.NoError(t, err)
+	assert.Equal(t, "ami-abc123", meta.ImageID)
+	assert.Equal(t, "test-ami", meta.Name)
+	assert.Equal(t, "x86_64", meta.Architecture)
+	assert.Equal(t, "Linux/UNIX", meta.PlatformDetails)
+	assert.Equal(t, "hvm", meta.Virtualization)
+}
+
+func TestGetAMIConfig_NotFound(t *testing.T) {
+	svc, _ := setupTestImageService(t)
+
+	_, err := svc.getAMIConfig("ami-nonexistent")
+	require.Error(t, err)
+}
+
+func TestPutSnapshotMetadata(t *testing.T) {
+	svc, store := setupTestImageService(t)
+
+	err := svc.putSnapshotMetadata("snap-abc123", "vol-xyz789", 10)
+	require.NoError(t, err)
+
+	// Verify the metadata was written correctly
+	result, err := store.GetObject(&awss3.GetObjectInput{
+		Bucket: aws.String(testBucket),
+		Key:    aws.String("snap-abc123/metadata.json"),
+	})
+	require.NoError(t, err)
+	defer result.Body.Close()
+
+	var cfg handlers_ec2_snapshot.SnapshotConfig
+	err = json.NewDecoder(result.Body).Decode(&cfg)
+	require.NoError(t, err)
+	assert.Equal(t, "snap-abc123", cfg.SnapshotID)
+	assert.Equal(t, "vol-xyz789", cfg.VolumeID)
+	assert.Equal(t, int64(10), cfg.VolumeSize)
+	assert.Equal(t, "completed", cfg.State)
+	assert.Equal(t, "100%", cfg.Progress)
+	assert.Equal(t, testAccountID, cfg.OwnerID)
+}
+
+func TestCreateImageFromInstance_SourceAMIReadFailure(t *testing.T) {
+	svc, store := setupTestImageService(t)
+
+	// Create volume config but NOT the source AMI config
+	createTestVolumeConfig(t, store, "vol-root123", 10)
+
+	// With non-empty SourceImageID, missing AMI config should be a hard error
+	_, err := svc.CreateImageFromInstance(CreateImageParams{
+		Input: &ec2.CreateImageInput{
+			InstanceId: aws.String("i-test123"),
+			Name:       aws.String("my-image"),
+		},
+		RootVolumeID:  "vol-root123",
+		SourceImageID: "ami-nonexistent",
+		IsRunning:     true, // will fail at snapshot step first (no NATS)
+	})
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+}
+
+func TestDescribeImages_FilterByOwnerSelf(t *testing.T) {
+	svc, store := setupTestImageService(t)
+
+	// Create AMI with "self" owner alias
+	amiID := "ami-selfowned"
+	amiState := viperblock.VBState{
+		VolumeConfig: viperblock.VolumeConfig{
+			AMIMetadata: viperblock.AMIMetadata{
+				ImageID:         amiID,
+				Name:            "self-owned-ami",
+				ImageOwnerAlias: "self",
+				Architecture:    "x86_64",
+				PlatformDetails: "Linux/UNIX",
+				RootDeviceType:  "ebs",
+				VolumeSizeGiB:   8,
+			},
+		},
+	}
+	data, err := json.Marshal(amiState)
+	require.NoError(t, err)
+	_, err = store.PutObject(&awss3.PutObjectInput{
+		Bucket: aws.String(testBucket),
+		Key:    aws.String(amiID + "/config.json"),
+		Body:   strings.NewReader(string(data)),
+	})
+	require.NoError(t, err)
+
+	// Filter by "self"
+	result, err := svc.DescribeImages(&ec2.DescribeImagesInput{
+		Owners: []*string{aws.String("self")},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, result)
+	assert.Len(t, result.Images, 1)
+	assert.Equal(t, amiID, *result.Images[0].ImageId)
+}

hive/handlers/ec2/image/service_nats.go

@@ -1,7 +1,7 @@
 package handlers_ec2_image
 
 import (
-	"fmt"
+	"errors"
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -23,31 +23,30 @@ func (s *NATSImageService) DescribeImages(input *ec2.DescribeImagesInput) (*ec2.
 	return utils.NATSRequest[ec2.DescribeImagesOutput](s.natsConn, "ec2.DescribeImages", input, 30*time.Second)
 }
 
-// Stub implementations for unimplemented ImageService methods
 func (s *NATSImageService) CreateImage(input *ec2.CreateImageInput) (*ec2.CreateImageOutput, error) {
-	return nil, fmt.Errorf("CreateImage not yet implemented")
+	return utils.NATSRequest[ec2.CreateImageOutput](s.natsConn, "ec2.CreateImage", input, 120*time.Second)
 }
 
 func (s *NATSImageService) CopyImage(input *ec2.CopyImageInput) (*ec2.CopyImageOutput, error) {
-	return nil, fmt.Errorf("CopyImage not yet implemented")
+	return nil, errors.New("CopyImage not yet implemented")
 }
 
 func (s *NATSImageService) DescribeImageAttribute(input *ec2.DescribeImageAttributeInput) (*ec2.DescribeImageAttributeOutput, error) {
-	return nil, fmt.Errorf("DescribeImageAttribute not yet implemented")
+	return nil, errors.New("DescribeImageAttribute not yet implemented")
 }
 
 func (s *NATSImageService) RegisterImage(input *ec2.RegisterImageInput) (*ec2.RegisterImageOutput, error) {
-	return nil, fmt.Errorf("RegisterImage not yet implemented")
+	return nil, errors.New("RegisterImage not yet implemented")
 }
 
 func (s *NATSImageService) DeregisterImage(input *ec2.DeregisterImageInput) (*ec2.DeregisterImageOutput, error) {
-	return nil, fmt.Errorf("DeregisterImage not yet implemented")
+	return nil, errors.New("DeregisterImage not yet implemented")
 }
 
 func (s *NATSImageService) ModifyImageAttribute(input *ec2.ModifyImageAttributeInput) (*ec2.ModifyImageAttributeOutput, error) {
-	return nil, fmt.Errorf("ModifyImageAttribute not yet implemented")
+	return nil, errors.New("ModifyImageAttribute not yet implemented")
 }
 
 func (s *NATSImageService) ResetImageAttribute(input *ec2.ResetImageAttributeInput) (*ec2.ResetImageAttributeOutput, error) {
-	return nil, fmt.Errorf("ResetImageAttribute not yet implemented")
+	return nil, errors.New("ResetImageAttribute not yet implemented")
 }

scripts/stop-dev.sh

@@ -65,6 +65,27 @@ stop_service "hive-ui" "$PID_DIR"
 # Stop Hive daemon/gateway (it will terminate running instances, unmount nbd devices)
 stop_service "hive" "$PID_DIR"
 
+# Wait for all QEMU instances to exit before stopping infrastructure services.
+# The daemon sends system_powerdown to VMs during shutdown, but may be killed
+# before VMs fully exit. We must wait here so viperblock/predastore aren't
+# pulled out from under running VMs.
+if pgrep -x qemu-system-x86_64 > /dev/null 2>&1; then
+    echo "⏳ Waiting for QEMU instances to exit..."
+    timeout=120
+    elapsed=0
+    while pgrep -x qemu-system-x86_64 > /dev/null 2>&1; do
+        if [ $elapsed -ge $timeout ]; then
+            echo "⚠️  Timeout waiting for QEMU processes, force killing..."
+            pkill -9 -x qemu-system-x86_64 2>/dev/null || true
+            sleep 1
+            break
+        fi
+        sleep 1
+        ((elapsed++))
+    done
+    echo "✅ All QEMU instances exited"
+fi
+
 # Stop AWSGW
 stop_service "awsgw" "$PID_DIR"