diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go
index 99f7bed7..24b4da4f 100644
--- a/go/vehicle-gateway/internal/gateway/mqtt_client.go
+++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go
@@ -176,6 +176,9 @@ func (c *MQTTClient) buildTLSConfig() (*tls.Config, error) {
 }
 
 func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []byte) {
+	messageCtx, cancelMessage := context.WithTimeout(context.WithoutCancel(ctx), frameOperationTimeout)
+	defer cancelMessage()
+
 	receivedAtMS := time.Now().UnixMilli()
 	env, err := yutongmqtt.ParseMessage(c.cfg.EndpointName, topic, payload, receivedAtMS)
 	if err != nil {
@@ -193,7 +196,7 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []
 		}
 		env.EventID = env.StableEventID()
 	} else {
-		resolved, resolveErr := c.cfg.Resolver.Resolve(ctx, env)
+		resolved, resolveErr := c.cfg.Resolver.Resolve(messageCtx, env)
 		if resolveErr != nil {
 			c.cfg.Logger.Warn("mqtt identity resolve failed", "topic", topic, "event_id", env.StableEventID(), "error", resolveErr)
 			if env.Parsed == nil {
@@ -205,14 +208,14 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []
 			env = resolved
 		}
 	}
-	if err := c.cfg.Sink.PublishRaw(ctx, env); err != nil {
+	if err := c.cfg.Sink.PublishRaw(messageCtx, env); err != nil {
 		c.cfg.Logger.Error("publish mqtt raw failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
 		return
 	}
 	if env.ParseStatus == envelope.ParseBadFrame {
 		return
 	}
-	if err := c.cfg.Sink.PublishUnified(ctx, env); err != nil {
+	if err := c.cfg.Sink.PublishUnified(messageCtx, env); err != nil {
 		c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
 	}
 }
diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go
index a5cbd696..f026aecb 100644
--- a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go
+++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go
@@ -69,6 +69,45 @@ func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) {
 	}
 }
 
+func TestMQTTClientUsesUncancelledMessageContextForReceivedMessage(t *testing.T) {
+	parent, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	resolver := &contextCheckingResolver{}
+	sink := &contextCheckingSink{}
+	client, err := NewMQTTClient(MQTTClientConfig{
+		EndpointName: "endpoint-a",
+		Broker:       "tcp://127.0.0.1:1883",
+		ClientID:     "test-client",
+		Topics:       []string{"/ytforward/shln/+"},
+		Sink:         sink,
+		Resolver:     resolver,
+		Logger:       slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
+	})
+	if err != nil {
+		t.Fatalf("NewMQTTClient() error = %v", err)
+	}
+
+	client.handleMessage(parent, "/ytforward/shln/dev1", []byte(`{
+	  "device":"LTEST000000000001",
+	  "time":"20260413100000",
+	  "data":{"METER_SPEED":52.3,"TOTAL_MILEAGE":123456.7}
+	}`))
+
+	if resolver.ctxErr != nil {
+		t.Fatalf("resolver saw cancelled context: %v", resolver.ctxErr)
+	}
+	if sink.rawCtxErr != nil {
+		t.Fatalf("raw publish saw cancelled context: %v", sink.rawCtxErr)
+	}
+	if sink.unifiedCtxErr != nil {
+		t.Fatalf("unified publish saw cancelled context: %v", sink.unifiedCtxErr)
+	}
+	if sink.rawCount != 1 || sink.unifiedCount != 1 {
+		t.Fatalf("raw=%d unified=%d", sink.rawCount, sink.unifiedCount)
+	}
+}
+
 func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) {
 	dir := t.TempDir()
 	caPath, certPath, keyPath := writeTestTLSMaterial(t, dir)