diff --git a/app/Console/Commands/MqttListenCommand.php b/app/Console/Commands/MqttListenCommand.php
index 57a7ce1..162236a 100644
--- a/app/Console/Commands/MqttListenCommand.php
+++ b/app/Console/Commands/MqttListenCommand.php
@@ -33,7 +33,9 @@ class MqttListenCommand extends Command
                 foreach (ShellyTopics::subscriptions() as $topic) {
                     $this->client->subscribe($topic, function (string $topic, string $message): void {
                         // H2: parse + dispatch only — never do DB/HTTP work in the loop.
-                        IngestShellyMessage::dispatch($topic, $message);
+                        // Stamp the receive time (µs) so out-of-order job completion can't
+                        // overwrite newer state with older (concurrent workers).
+                        IngestShellyMessage::dispatch($topic, $message, (int) round(microtime(true) * 1_000_000));
                     }, 0);
                 }
 
diff --git a/app/Jobs/IngestShellyMessage.php b/app/Jobs/IngestShellyMessage.php
index 296c615..2b29a84 100644
--- a/app/Jobs/IngestShellyMessage.php
+++ b/app/Jobs/IngestShellyMessage.php
@@ -5,14 +5,20 @@ namespace App\Jobs;
 use App\Events\DeviceStateChanged;
 use App\Models\Device;
 use App\Models\DeviceState;
+use App\Models\Entity;
 use App\Support\Mqtt\ShellyNormalizer;
 use App\Support\Mqtt\ShellyTopics;
 use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Database\UniqueConstraintViolationException;
 use Illuminate\Foundation\Queue\Queueable;
 
 /**
  * The single ingest path for Shelly status messages (H4). The MQTT callback only
  * dispatches this (H2); here we resolve the device, upsert current state and broadcast.
+ *
+ * `observedAt` is the µs receive time from the listener. Because concurrent Horizon
+ * workers can finish out of order, state is only written when this message is newer than
+ * what is stored — an older message can never overwrite a newer one.
  */
 class IngestShellyMessage implements ShouldQueue
 {
@@ -21,6 +27,7 @@ class IngestShellyMessage implements ShouldQueue
     public function __construct(
         public string $topic,
         public string $payload,
+        public ?int $observedAt = null,
     ) {}
 
     public function handle(): void
@@ -43,6 +50,8 @@ class IngestShellyMessage implements ShouldQueue
             return;
         }
 
+        $observedAt = $this->observedAt ?? (int) round(microtime(true) * 1_000_000);
+
         $device->forceFill(['last_seen_at' => now()])->save();
 
         foreach (ShellyNormalizer::normalize($component, $data) as $update) {
@@ -51,12 +60,46 @@ class IngestShellyMessage implements ShouldQueue
                 ['type' => $update['type']],
             );
 
-            DeviceState::updateOrCreate(
-                ['entity_id' => $entity->id],
-                ['state' => $update['state']],
-            );
-
-            DeviceStateChanged::dispatch($device->uuid, $entity->key, $update['state']);
+            if ($this->applyState($entity->id, $update['state'], $observedAt)) {
+                DeviceStateChanged::dispatch($device->uuid, $entity->key, $update['state']);
+            }
         }
     }
+
+    /**
+     * Monotonic, race-safe upsert. Returns true when this message's state was applied,
+     * false when it was stale (an equal-or-newer state is already stored).
+     */
+    private function applyState(int $entityId, array $state, int $observedAt): bool
+    {
+        $encoded = json_encode($state);
+
+        // Overwrite an existing row only when this message is newer.
+        $updated = DeviceState::query()
+            ->where('entity_id', $entityId)
+            ->where(fn ($q) => $q->whereNull('observed_at')->orWhere('observed_at', '<', $observedAt))
+            ->update(['state' => $encoded, 'observed_at' => $observedAt, 'updated_at' => now()]);
+
+        if ($updated > 0) {
+            return true;
+        }
+
+        // No row yet → insert (unique(entity_id) guards against a concurrent insert).
+        if (! DeviceState::where('entity_id', $entityId)->exists()) {
+            try {
+                DeviceState::create(['entity_id' => $entityId, 'state' => $state, 'observed_at' => $observedAt]);
+
+                return true;
+            } catch (UniqueConstraintViolationException) {
+                // Lost the insert race — fall through and retry the conditional update.
+                return DeviceState::query()
+                    ->where('entity_id', $entityId)
+                    ->where(fn ($q) => $q->whereNull('observed_at')->orWhere('observed_at', '<', $observedAt))
+                    ->update(['state' => $encoded, 'observed_at' => $observedAt, 'updated_at' => now()]) > 0;
+            }
+        }
+
+        // Existing row is equal-or-newer → this message is stale.
+        return false;
+    }
 }
diff --git a/app/Models/DeviceState.php b/app/Models/DeviceState.php
index b80ad7f..d64161f 100644
--- a/app/Models/DeviceState.php
+++ b/app/Models/DeviceState.php
@@ -10,9 +10,9 @@ use Illuminate\Database\Eloquent\Relations\BelongsTo;
  */
 class DeviceState extends Model
 {
-    protected $fillable = ['entity_id', 'state'];
+    protected $fillable = ['entity_id', 'state', 'observed_at'];
 
-    protected $casts = ['state' => 'array'];
+    protected $casts = ['state' => 'array', 'observed_at' => 'integer'];
 
     public function entity(): BelongsTo
     {
diff --git a/database/migrations/2026_07_17_203001_add_observed_at_to_device_states.php b/database/migrations/2026_07_17_203001_add_observed_at_to_device_states.php
new file mode 100644
index 0000000..3cfddbb
--- /dev/null
+++ b/database/migrations/2026_07_17_203001_add_observed_at_to_device_states.php
@@ -0,0 +1,24 @@
+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+return new class extends Migration
+{
+    public function up(): void
+    {
+        Schema::table('device_states', function (Blueprint $table) {
+            // Microseconds since epoch, stamped when the listener received the message.
+            // Used to reject out-of-order updates (concurrent Horizon workers).
+            $table->unsignedBigInteger('observed_at')->nullable()->after('state');
+        });
+    }
+
+    public function down(): void
+    {
+        Schema::table('device_states', function (Blueprint $table) {
+            $table->dropColumn('observed_at');
+        });
+    }
+};
diff --git a/tests/Feature/MqttTest.php b/tests/Feature/MqttTest.php
index 721cddf..c810d6b 100644
--- a/tests/Feature/MqttTest.php
+++ b/tests/Feature/MqttTest.php
@@ -34,6 +34,22 @@ class MqttTest extends TestCase
         $this->assertTrue($device->refresh()->last_seen_at->gt(now()->subMinute()));
     }
 
+    public function test_out_of_order_messages_do_not_overwrite_newer_state(): void
+    {
+        $device = Device::create([
+            'name' => 'Lampe', 'vendor' => 'Shelly', 'protocol' => 'mqtt',
+            'config' => ['mqtt_prefix' => 'shelly-lampe'], 'status' => 'active',
+        ]);
+
+        // Newer message (higher observed_at) says on=true …
+        (new IngestShellyMessage('shelly-lampe/status/switch:0', json_encode(['output' => true]), 2_000))->handle();
+        // … an older one arriving late must NOT overwrite it.
+        (new IngestShellyMessage('shelly-lampe/status/switch:0', json_encode(['output' => false]), 1_000))->handle();
+
+        $switch = $device->entities()->where('key', 'switch:0')->first();
+        $this->assertTrue(data_get($switch->state->state, 'on'));
+    }
+
     public function test_ingest_ignores_unknown_device(): void
     {
         (new IngestShellyMessage('shelly-nope/status/switch:0', json_encode(['output' => true])))->handle();